pub fn spawn_sender_loop(
    timeout_duration: Duration,
    mut websocket: WebSocket,
    mut ws_outgoing_receiver: mpsc::UnboundedReceiver<(
        AbsintheWSRequest,
        Option<oneshot::Receiver<bool>>,
    )>,
    mut ws_disconnect_receiver: mpsc::UnboundedReceiver<()>,
    message_broker_link: mpsc::UnboundedSender<BrokerAction>,
) {
    tokio::spawn(async move {
        // The idea is that try_recv will only work when it receives a disconnect signal
        // This is a bit ugly imo and we should probably change in the future
        while ws_disconnect_receiver.recv().now_or_never().is_none() {
            // let ready_map = HashMap::new();
            let next_outgoing = ws_outgoing_receiver.recv();
            let next_incoming = tokio::time::timeout(timeout_duration, websocket.next());
            tokio::pin!(next_outgoing);
            tokio::pin!(next_incoming);
            match select(next_outgoing, next_incoming).await {
                Either::Left((outgoing, _)) => {
                    if let Some((request, _ready_rx)) = outgoing {
                        if let Ok(request_raw) = serde_json::to_string(&request) {
                            // If sending fails, pass error through broker and global channel
                            match websocket.send(Message::Text(request_raw)).await {
                                Ok(_) => {
                                    trace!(id = ?request.message_id(), "SEND");
                                }
                                Err(e) => {
                                    error!(error = %e, "SEND channel error");
                                    let error = ProtocolError("failed to send message on WS connection, likely disconnected");
                                    let _ =
                                        message_broker_link.send(BrokerAction::Message(Err(error)));
                                    break;
                                }
                            }
                        } else {
                            error!(request = ?request, "SEND invalid request");
                        }
                    } else {
                        error!("SEND channel error");
                        let error =
                            ProtocolError("outgoing channel died or errored, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
                Either::Right((incoming, _)) => {
                    if let Ok(incoming) = incoming {
                        if let Some(Ok(message)) = incoming {
                            let raw_response = message.into_text().map_err(|e| {
                                ProtocolError::coerce_static_from_str(e.to_string().as_str())
                            });
                            let response: Result<AbsintheWSResponse> = raw_response.and_then(|r| {
                                serde_json::from_str(&r).map_err(|e| {
                                    ProtocolError::coerce_static_from_str(e.to_string().as_str())
                                })
                            });
                            match response {
                                Ok(response) => {
                                    trace!(id = ?response.message_id(), "RECV success");
                                    let _ = message_broker_link
                                        .send(BrokerAction::Message(Ok(response)));
                                }
                                Err(e) => {
                                    error!(error = %e, "RECV invalid response message");
                                    let _ = message_broker_link.send(BrokerAction::Message(Err(e)));
                                    break;
                                }
                            }
                        } else {
                            error!("RECV channel error");
                            let error = ProtocolError(
                                "incoming channel died or errored, likely disconnected",
                            );
                            let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                            break;
                        }
                    } else {
                        error!("RECV timed out");
                        let error = ProtocolError("incoming WS timed out, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
            };
        }
        error!("DISCONNECT");
        let error = ProtocolError("Disconnected.");
        message_broker_link
            .send(BrokerAction::Message(Err(error)))
            .ok();
    });
}
// this will receive messages to send out over websockets on one channel, and pass incoming ws messages
// back up to client on another channel.
this will receive messages to send out over websockets on one channel, and pass incoming ws messages
back up to client on another channel.
this will receive messages to send out over websockets on one channel and pass incoming ws messages back up to client on another channel .
pub fn spawn_sender_loop(
    timeout_duration: Duration,
    mut websocket: WebSocket,
    mut ws_outgoing_receiver: mpsc::UnboundedReceiver<(
        AbsintheWSRequest,
        Option<oneshot::Receiver<bool>>,
    )>,
    mut ws_disconnect_receiver: mpsc::UnboundedReceiver<()>,
    message_broker_link: mpsc::UnboundedSender<BrokerAction>,
) {
    tokio::spawn(async move {
        // The idea is that try_recv will only work when it receives a disconnect signal
        // This is a bit ugly imo and we should probably change in the future
        while ws_disconnect_receiver.recv().now_or_never().is_none() {
            // let ready_map = HashMap::new();
            let next_outgoing = ws_outgoing_receiver.recv();
            let next_incoming = tokio::time::timeout(timeout_duration, websocket.next());
            tokio::pin!(next_outgoing);
            tokio::pin!(next_incoming);
            match select(next_outgoing, next_incoming).await {
                Either::Left((outgoing, _)) => {
                    if let Some((request, _ready_rx)) = outgoing {
                        if let Ok(request_raw) = serde_json::to_string(&request) {
                            // If sending fails, pass error through broker and global channel
                            match websocket.send(Message::Text(request_raw)).await {
                                Ok(_) => {
                                    trace!(id = ?request.message_id(), "SEND");
                                }
                                Err(e) => {
                                    error!(error = %e, "SEND channel error");
                                    let error = ProtocolError("failed to send message on WS connection, likely disconnected");
                                    let _ =
                                        message_broker_link.send(BrokerAction::Message(Err(error)));
                                    break;
                                }
                            }
                        } else {
                            error!(request = ?request, "SEND invalid request");
                        }
                    } else {
                        error!("SEND channel error");
                        let error =
                            ProtocolError("outgoing channel died or errored, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
                Either::Right((incoming, _)) => {
                    if let Ok(incoming) = incoming {
                        if let Some(Ok(message)) = incoming {
                            let raw_response = message.into_text().map_err(|e| {
                                ProtocolError::coerce_static_from_str(e.to_string().as_str())
                            });
                            let response: Result<AbsintheWSResponse> = raw_response.and_then(|r| {
                                serde_json::from_str(&r).map_err(|e| {
                                    ProtocolError::coerce_static_from_str(e.to_string().as_str())
                                })
                            });
                            match response {
                                Ok(response) => {
                                    trace!(id = ?response.message_id(), "RECV success");
                                    let _ = message_broker_link
                                        .send(BrokerAction::Message(Ok(response)));
                                }
                                Err(e) => {
                                    error!(error = %e, "RECV invalid response message");
                                    let _ = message_broker_link.send(BrokerAction::Message(Err(e)));
                                    break;
                                }
                            }
                        } else {
                            error!("RECV channel error");
                            let error = ProtocolError(
                                "incoming channel died or errored, likely disconnected",
                            );
                            let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                            break;
                        }
                    } else {
                        error!("RECV timed out");
                        let error = ProtocolError("incoming WS timed out, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
            };
        }
        error!("DISCONNECT");
        let error = ProtocolError("Disconnected.");
        message_broker_link
            .send(BrokerAction::Message(Err(error)))
            .ok();
    });
}
pub fn spawn_sender_loop ( timeout_duration : Duration , mut websocket : WebSocket , mut ws_outgoing_receiver : mpsc :: UnboundedReceiver < ( AbsintheWSRequest , Option < oneshot :: Receiver < bool > > , ) > , mut ws_disconnect_receiver : mpsc :: UnboundedReceiver < ( ) > , message_broker_link : mpsc :: UnboundedSender < BrokerAction > , ) { tokio :: spawn ( async move { // The idea is that try_recv will only work when it receives a disconnect signal // This is a bit ugly imo and we should probably change in the future while ws_disconnect_receiver . recv ( ) . now_or_never ( ) . is_none ( ) { // let ready_map = HashMap::new(); let next_outgoing = ws_outgoing_receiver . recv ( ) ; let next_incoming = tokio :: time :: timeout ( timeout_duration , websocket . next ( ) ) ; tokio :: pin ! ( next_outgoing ) ; tokio :: pin ! ( next_incoming ) ; match select ( next_outgoing , next_incoming ) . await { Either :: Left ( ( outgoing , _ ) ) => { if let Some ( ( request , _ready_rx ) ) = outgoing { if let Ok ( request_raw ) = serde_json :: to_string ( & request ) { // If sending fails, pass error through broker and global channel match websocket . send ( Message :: Text ( request_raw ) ) . await { Ok ( _ ) => { trace ! ( id request message_id ( ) " " ) ; } Err ( e ) => { error ! ( error e " " ) ; let error = ProtocolError ( " " ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) ; break ; } } } else { error ! ( request request " " ) ; } } else { error ! ( " " ) ; let error = ProtocolError ( " " ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) ; break ; } } Either :: Right ( ( incoming , _ ) ) => { if let Ok ( incoming ) = incoming { if let Some ( Ok ( message ) ) = incoming { let raw_response = message . into_text ( ) . map_err ( | e | { ProtocolError :: coerce_static_from_str ( e . to_string ( ) . as_str ( ) ) } ) ; let response : Result < AbsintheWSResponse > = raw_response . and_then ( | r | { serde_json :: from_str ( & r ) . map_err ( | e | { ProtocolError :: coerce_static_from_str ( e . to_string ( ) . as_str ( ) ) } ) } ) ; match response { Ok ( response ) => { trace ! ( id response message_id ( ) " " ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Ok ( response ) ) ) ; } Err ( e ) => { error ! ( error e " " ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( e ) ) ) ; break ; } } } else { error ! ( " " ) ; let error = ProtocolError ( " " , ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) ; break ; } } else { error ! ( " " ) ; let error = ProtocolError ( " " ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) ; break ; } } } ; } error ! ( " " ) ; let error = ProtocolError ( " " ) ; message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) . ok ( ) ; } ) ; }
this will receive messages to send out over websockets on one channel, and pass incoming ws messages
back up to client on another channel.
this will receive messages to send out over websockets on one channel and pass incoming ws messages back up to client on another channel .
timeout_duration
Duration
websocket
WebSocket
ws_outgoing_receiver
mpsc::UnboundedReceiver<(
        AbsintheWSRequest,
        Option<oneshot::Receiver<bool>>,
    )>
ws_disconnect_receiver
mpsc::UnboundedReceiver<()>
message_broker_link
mpsc::UnboundedSender<BrokerAction>