How to handle the closing of a sync connection?

[ghost]

No description provided.

[MightyPork]

I'm facing the same problem.

I managed to shut down the sender by injecting a OwnedMessage::Close(None) into the mpsc channel. But how do I shut down the receiver?

---

Edit: I found a solution! This should really be better documented.

let (mut ws_reader, mut ws_writer) = client.split().unwrap();
let (pipe_tx, pipe_rx) = mpsc::channel();
let pipe_tx1 = pipe_tx.clone(); // control pipe for controlled shutdown

let send_loop = thread::spawn(move || {
    'sender: loop {
        let message = match pipe_rx.recv() {
            Ok(m) => m,
            Err(e) => {
                break 'sender; // !!! break instead of return
            }
        };

        match message {
            OwnedMessage::Close(_) => {
                let _ = ws_writer.send_message(&message);
                break 'sender; // !!! break instead of return
            }
            _ => (),
        }

        match ws_writer.send_message(&message) {
            Ok(()) => (),
            Err(e) => {
                let _ = ws_writer.send_message(&Message::close());
                break 'sender; // !!! break instead of return
            }
        }
    }

    // !!! this is the magic line
    ws_writer.shutdown_all();
});

let receive_loop = thread::spawn(move || {
    for message in ws_reader.incoming_messages() {
        let message = match message {
            Ok(m) => m,
            Err(e) => {
                let _ = pipe_tx.send(OwnedMessage::Close(None));
                return;
            }
        };

        match message {
            OwnedMessage::Close(_) => {
                let _ = pipe_tx.send(OwnedMessage::Close(None));
                return;
            }
            OwnedMessage::Ping(data) => {
                match pipe_tx.send(OwnedMessage::Pong(data)) {
                    Ok(()) => (),
                    Err(e) => {
                        return;
                    }
                }
            }
            _ => println!("Received: {:?}", message), // TODO handle data
        }
    }
});

thread::sleep(Duration::from_millis(5000));

println!("Trying to close");
let _ = pipe_tx1.send(OwnedMessage::Close(None));

println!("Waiting for threads");

send_loop.join();
println!("Sender closed");
receive_loop.join();
println!("Receiver closed");

println!("Terminated.");