/// Convert this BtpOutgoingService into a bidirectional BtpService by adding a handler for incoming requests.
/// This will automatically pull all incoming Prepare packets from the channel buffer and call the IncomingService with them.
Convert this BtpOutgoingService into a bidirectional BtpService by adding a handler for incoming requests.
This will automatically pull all incoming Prepare packets from the channel buffer and call the IncomingService with them.
Convert this BtpOutgoingService into a bidirectional BtpService by adding a handler for incoming requests . This will automatically pull all incoming Prepare packets from the channel buffer and call the IncomingService with them .
pub async fn handle_incoming<I>(self, incoming_handler: I) -> BtpService<I, O, A>
    where
        I: IncomingService<A> + Clone + Send + 'static,
    {
        // Any connections that were added to the BtpOutgoingService will just buffer
        // the incoming Prepare packets they get in self.pending_incoming
        // Now that we're adding an incoming handler, this will spawn a task to read
        // all Prepare packets from the buffer, handle them, and send the responses back
        let connections_clone = self.connections.clone();
        let mut handle_pending_incoming = self
            .pending_incoming
            .lock()
            .take()
            .expect("handle_incoming can only be called once");
        let handle_pending_incoming_fut = async move {
            while let Some((account, request_id, prepare)) = handle_pending_incoming.next().await {
                let account_id = account.id();
                let connections_clone = connections_clone.clone();
                let request = IncomingRequest {
                    from: account,
                    prepare,
                };
                trace!(
                    "Handling incoming request {} from account: {} (id: {})",
                    request_id,
                    request.from.username(),
                    request.from.id()
                );
                let mut handler = incoming_handler.clone();
                let packet = match handler.handle_request(request).await {
                    Ok(fulfill) => Packet::Fulfill(fulfill),
                    Err(reject) => Packet::Reject(reject),
                };
                if let Some(connection) = connections_clone.clone().read().get(&account_id) {
                    let message = ilp_packet_to_ws_message(request_id, packet);
                    let _ = connection.unbounded_send(message).map_err(move |err| {
                        error!(
                            "Error sending response to account: {} {:?}",
                            account_id, err
                        )
                    });
                } else {
                    error!(
                        "Error sending response to account: {}, connection was closed. {:?}",
                        account_id, packet
                    );
                }
            }
            trace!("Finished reading from pending_incoming buffer");
            Ok::<(), ()>(())
        };
        tokio::spawn(handle_pending_incoming_fut);
        BtpService {
            outgoing: self,
            incoming_handler_type: PhantomData,
        }
    }
pub async fn handle_incoming < I > ( self , incoming_handler : I ) -> BtpService < I , O , A > where I : IncomingService < A > + Clone + Send + ' static , { // Any connections that were added to the BtpOutgoingService will just buffer // the incoming Prepare packets they get in self.pending_incoming // Now that we're adding an incoming handler, this will spawn a task to read // all Prepare packets from the buffer, handle them, and send the responses back let connections_clone = self . connections . clone ( ) ; let mut handle_pending_incoming = self . pending_incoming . lock ( ) . take ( ) . expect ( " " ) ; let handle_pending_incoming_fut = async move { while let Some ( ( account , request_id , prepare ) ) = handle_pending_incoming . next ( ) . await { let account_id = account . id ( ) ; let connections_clone = connections_clone . clone ( ) ; let request = IncomingRequest { from : account , prepare , } ; trace ! ( " " request_id request from username ( ) request from id ( ) ) ; let mut handler = incoming_handler . clone ( ) ; let packet = match handler . handle_request ( request ) . await { Ok ( fulfill ) => Packet :: Fulfill ( fulfill ) , Err ( reject ) => Packet :: Reject ( reject ) , } ; if let Some ( connection ) = connections_clone . clone ( ) . read ( ) . get ( & account_id ) { let message = ilp_packet_to_ws_message ( request_id , packet ) ; let _ = connection . unbounded_send ( message ) . map_err ( move | err | { error ! ( " " account_id err ) } ) ; } else { error ! ( " " account_id packet ) ; } } trace ! ( " " ) ; Ok :: < ( ) , ( ) > ( ( ) ) } ; tokio :: spawn ( handle_pending_incoming_fut ) ; BtpService { outgoing : self , incoming_handler_type : PhantomData , } }
Convert this BtpOutgoingService into a bidirectional BtpService by adding a handler for incoming requests.
Convert this BtpOutgoingService into a bidirectional BtpService by adding a handler for incoming requests .
self
incoming_handler
I