pub fn spawn_sender_loop(
    mut websocket: WebSocket,
    mut ws_outgoing_receiver: UnboundedReceiver<AbsintheWSRequest>,
    message_broker_link: UnboundedSender<BrokerAction>,
) {
    tokio::spawn(async move {
        loop {
            let next_outgoing = ws_outgoing_receiver.recv().boxed();
            let next_incoming = websocket.next();
            match select(next_outgoing, next_incoming).await {
                Either::Left((out_msg, _)) => {
                    if let Some(Ok(m_text)) = out_msg.map(|x| serde_json::to_string(&x)) {
                        // If sending fails, pass error through broker and global channel
                        match websocket.send(Message::Text(m_text)).await {
                            Ok(_) => {}
                            Err(_) => {
                                let error = ProtocolError(
                                    "failed to send message on WS connection, likely disconnected",
                                );
                                let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                                break;
                            }
                        }
                    } else {
                        let error =
                            ProtocolError("outgoing channel died or errored, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
                Either::Right((in_msg, _)) => {
                    if let Some(Ok(Ok(resp))) = in_msg.map(|x| x.map(|x| x.into_text())) {
                        if let Ok(resp_copy1) = serde_json::from_str(&resp) {
                            // Similarly, let the client timeout on incoming response if this fails and handle that
                            // todo: this is a hack since the graphql library has not implemented clone() for responses
                            // but we need another copy of the response to send to the broker
                            // this could actually be more efficient and have broker handle all messages?
                            let _ignore =
                                message_broker_link.send(BrokerAction::Message(Ok(resp_copy1)));
                        } else {
                            // if response fails to destructure, ignore and continue broker loop
                        }
                    } else {
                        let error =
                            ProtocolError("incoming WS channel failed, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
            };
        }
    });
}
// this will recieve messages to send out over websockets on one channel, and pass incoming ws messages
// back up to client on anoter channel.
this will recieve messages to send out over websockets on one channel, and pass incoming ws messages
back up to client on anoter channel.
this will recieve messages to send out over websockets on one channel and pass incoming ws messages back up to client on anoter channel .
pub fn spawn_sender_loop(
    mut websocket: WebSocket,
    mut ws_outgoing_receiver: UnboundedReceiver<AbsintheWSRequest>,
    message_broker_link: UnboundedSender<BrokerAction>,
) {
    tokio::spawn(async move {
        loop {
            let next_outgoing = ws_outgoing_receiver.recv().boxed();
            let next_incoming = websocket.next();
            match select(next_outgoing, next_incoming).await {
                Either::Left((out_msg, _)) => {
                    if let Some(Ok(m_text)) = out_msg.map(|x| serde_json::to_string(&x)) {
                        // If sending fails, pass error through broker and global channel
                        match websocket.send(Message::Text(m_text)).await {
                            Ok(_) => {}
                            Err(_) => {
                                let error = ProtocolError(
                                    "failed to send message on WS connection, likely disconnected",
                                );
                                let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                                break;
                            }
                        }
                    } else {
                        let error =
                            ProtocolError("outgoing channel died or errored, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
                Either::Right((in_msg, _)) => {
                    if let Some(Ok(Ok(resp))) = in_msg.map(|x| x.map(|x| x.into_text())) {
                        if let Ok(resp_copy1) = serde_json::from_str(&resp) {
                            // Similarly, let the client timeout on incoming response if this fails and handle that
                            // todo: this is a hack since the graphql library has not implemented clone() for responses
                            // but we need another copy of the response to send to the broker
                            // this could actually be more efficient and have broker handle all messages?
                            let _ignore =
                                message_broker_link.send(BrokerAction::Message(Ok(resp_copy1)));
                        } else {
                            // if response fails to destructure, ignore and continue broker loop
                        }
                    } else {
                        let error =
                            ProtocolError("incoming WS channel failed, likely disconnected");
                        let _ = message_broker_link.send(BrokerAction::Message(Err(error)));
                        break;
                    }
                }
            };
        }
    });
}
pub fn spawn_sender_loop ( mut websocket : WebSocket , mut ws_outgoing_receiver : UnboundedReceiver < AbsintheWSRequest > , message_broker_link : UnboundedSender < BrokerAction > , ) { tokio :: spawn ( async move { loop { let next_outgoing = ws_outgoing_receiver . recv ( ) . boxed ( ) ; let next_incoming = websocket . next ( ) ; match select ( next_outgoing , next_incoming ) . await { Either :: Left ( ( out_msg , _ ) ) => { if let Some ( Ok ( m_text ) ) = out_msg . map ( | x | serde_json :: to_string ( & x ) ) { // If sending fails, pass error through broker and global channel match websocket . send ( Message :: Text ( m_text ) ) . await { Ok ( _ ) => { } Err ( _ ) => { let error = ProtocolError ( " " , ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) ; break ; } } } else { let error = ProtocolError ( " " ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) ; break ; } } Either :: Right ( ( in_msg , _ ) ) => { if let Some ( Ok ( Ok ( resp ) ) ) = in_msg . map ( | x | x . map ( | x | x . into_text ( ) ) ) { if let Ok ( resp_copy1 ) = serde_json :: from_str ( & resp ) { // Similarly, let the client timeout on incoming response if this fails and handle that // todo: this is a hack since the graphql library has not implemented clone() for responses // but we need another copy of the response to send to the broker // this could actually be more efficient and have broker handle all messages? let _ignore = message_broker_link . send ( BrokerAction :: Message ( Ok ( resp_copy1 ) ) ) ; } else { // if response fails to destructure, ignore and continue broker loop } } else { let error = ProtocolError ( " " ) ; let _ = message_broker_link . send ( BrokerAction :: Message ( Err ( error ) ) ) ; break ; } } } ; } } ) ; }
this will recieve messages to send out over websockets on one channel, and pass incoming ws messages
back up to client on anoter channel.
this will recieve messages to send out over websockets on one channel and pass incoming ws messages back up to client on anoter channel .
websocket
WebSocket
ws_outgoing_receiver
UnboundedReceiver<AbsintheWSRequest>
message_broker_link
UnboundedSender<BrokerAction>
websocket
WebSocket
ws_outgoing_receiver
UnboundedReceiver<AbsintheWSRequest>
message_broker_link
UnboundedSender<BrokerAction>