// Set up a WebSocket connection so that outgoing Prepare packets can be sent to it,
// incoming Prepare packets are buffered in a channel (until an IncomingService is added
// via the handle_incoming method), and ILP Fulfill and Reject packets will be
// sent back to the Future that sent the outgoing request originally.
Set up a WebSocket connection so that outgoing Prepare packets can be sent to it,
incoming Prepare packets are buffered in a channel (until an IncomingService is added
via the handle_incoming method), and ILP Fulfill and Reject packets will be
sent back to the Future that sent the outgoing request originally.
Set up a WebSocket connection so that outgoing Prepare packets can be sent to it incoming Prepare packets are buffered in a channel ( until an IncomingService is added via the handle_incoming method ) and ILP Fulfill and Reject packets will be sent back to the Future that sent the outgoing request originally .
pub(crate) fn add_connection(
        &self,
        account: A,
        ws_stream: impl Stream<Item = Message> + Sink<Message> + Send + 'static,
    ) {
        let account_id = account.id();
        // Set up a channel to forward outgoing packets to the WebSocket connection
        let (client_tx, client_rx) = unbounded();
        let (write, read) = ws_stream.split();
        let (close_connection, valve) = Valve::new();
        // tx -> rx -> write -> our peer
        // Responsible mainly for responding to Pings
        let write_to_ws = client_rx.map(Ok).forward(write).then(move |_| {
            async move {
                debug!(
                    "Finished forwarding to WebSocket stream for account: {}",
                    account_id
                );
                // When this is dropped, the read valve will close
                drop(close_connection);
                Ok::<(), ()>(())
            }
        });
        tokio::spawn(write_to_ws);
        // Process incoming messages depending on their type
        let pending_outgoing = self.pending_outgoing.clone();
        let incoming_sender = self.incoming_sender.clone();
        let client_tx_clone = client_tx.clone();
        let handle_message_fn = move |msg: Message| {
            handle_message(
                msg,
                client_tx_clone.clone(),
                account.clone(),
                pending_outgoing.clone(),
                incoming_sender.clone(),
            )
        };
        // Close connections trigger
        let read = valve.wrap(read); // close when `write_to_ws` calls `drop(connection)`
        let read = self.stream_valve.wrap(read);
        let read_from_ws = read.for_each(handle_message_fn).then(move |_| async move {
            debug!(
                "Finished reading from WebSocket stream for account: {}",
                account_id
            );
            Ok::<(), ()>(())
        });
        tokio::spawn(read_from_ws);
        // Send pings every PING_INTERVAL until the connection closes (when `drop(close_connection)` is called)
        // or the Service is dropped (which will implicitly drop `close_all_connections`, closing the stream_valve)
        let tx_clone = client_tx.clone();
        let ping_interval = time::interval(Duration::from_secs(PING_INTERVAL));
        let ping_stream = tokio_stream::wrappers::IntervalStream::new(ping_interval);
        let repeat_until_service_drops = self.stream_valve.wrap(ping_stream);
        let send_pings = valve.wrap(repeat_until_service_drops).for_each(move |_| {
            // For each tick send a ping
            if let Err(err) = tx_clone.unbounded_send(PING.clone()) {
                warn!(
                    "Error sending Ping on connection to account {}: {:?}",
                    account_id, err
                );
            }
            future::ready(())
        });
        tokio::spawn(send_pings);
        // Save the sender side of the channel so we have a way to forward outgoing requests to the WebSocket
        self.connections.write().insert(account_id, client_tx);
    }
pub ( crate ) fn add_connection ( & self , account : A , ws_stream : impl Stream < Item = Message > + Sink < Message > + Send + ' static , ) { let account_id = account . id ( ) ; // Set up a channel to forward outgoing packets to the WebSocket connection let ( client_tx , client_rx ) = unbounded ( ) ; let ( write , read ) = ws_stream . split ( ) ; let ( close_connection , valve ) = Valve :: new ( ) ; // tx -> rx -> write -> our peer // Responsible mainly for responding to Pings let write_to_ws = client_rx . map ( Ok ) . forward ( write ) . then ( move | _ | { async move { debug ! ( " " account_id ) ; // When this is dropped, the read valve will close drop ( close_connection ) ; Ok :: < ( ) , ( ) > ( ( ) ) } } ) ; tokio :: spawn ( write_to_ws ) ; // Process incoming messages depending on their type let pending_outgoing = self . pending_outgoing . clone ( ) ; let incoming_sender = self . incoming_sender . clone ( ) ; let client_tx_clone = client_tx . clone ( ) ; let handle_message_fn = move | msg : Message | { handle_message ( msg , client_tx_clone . clone ( ) , account . clone ( ) , pending_outgoing . clone ( ) , incoming_sender . clone ( ) , ) } ; // Close connections trigger let read = valve . wrap ( read ) ; // close when `write_to_ws` calls `drop(connection)` let read = self . stream_valve . wrap ( read ) ; let read_from_ws = read . for_each ( handle_message_fn ) . then ( move | _ | async move { debug ! ( " " account_id ) ; Ok :: < ( ) , ( ) > ( ( ) ) } ) ; tokio :: spawn ( read_from_ws ) ; // Send pings every PING_INTERVAL until the connection closes (when `drop(close_connection)` is called) // or the Service is dropped (which will implicitly drop `close_all_connections`, closing the stream_valve) let tx_clone = client_tx . clone ( ) ; let ping_interval = time :: interval ( Duration :: from_secs ( PING_INTERVAL ) ) ; let ping_stream = tokio_stream :: wrappers :: IntervalStream :: new ( ping_interval ) ; let repeat_until_service_drops = self . stream_valve . wrap ( ping_stream ) ; let send_pings = valve . wrap ( repeat_until_service_drops ) . for_each ( move | _ | { // For each tick send a ping if let Err ( err ) = tx_clone . unbounded_send ( PING . clone ( ) ) { warn ! ( " " account_id err ) ; } future :: ready ( ( ) ) } ) ; tokio :: spawn ( send_pings ) ; // Save the sender side of the channel so we have a way to forward outgoing requests to the WebSocket self . connections . write ( ) . insert ( account_id , client_tx ) ; }
Set up a WebSocket connection so that outgoing Prepare packets can be sent to it,
incoming Prepare packets are buffered in a channel (until an IncomingService is added
via the handle_incoming method), and ILP Fulfill and Reject packets will be
sent back to the Future that sent the outgoing request originally.
Set up a WebSocket connection so that outgoing Prepare packets can be sent to it incoming Prepare packets are buffered in a channel ( until an IncomingService is added via the handle_incoming method ) and ILP Fulfill and Reject packets will be sent back to the Future that sent the outgoing request originally .
&self
account
A
ws_stream
impl Stream<Item = Message> + Sink<Message> + Send + 'static