/// Send an outgoing request to one of the open connections.
///
/// If there is no open connection for the Account specified in `request.to`, the
/// request will be passed through to the `next` handler.
Send an outgoing request to one of the open connections.
If there is no open connection for the Account specified in `request.to`, the
request will be passed through to the `next` handler.
Send an outgoing request to one of the open connections . If there is no open connection for the Account specified in ` request . to ` the request will be passed through to the ` next ` handler .
async fn send_request(&mut self, request: OutgoingRequest<A>) -> IlpResult {
        let account_id = request.to.id();
        let found = self.connections.read().get(&account_id).cloned();
        if let Some(connection) = found {
            let request_id = random::<u32>();
            let ilp_address = self.ilp_address.clone();
            // Clone the trigger so that the connections stay open until we've
            // gotten the response to our outgoing request
            let keep_connections_open = self.close_all_connections.clone();
            trace!(
                "Sending outgoing request {} to {} ({})",
                request_id,
                request.to.username(),
                account_id
            );
            // Connection is an unbounded sender which sends to the rx that
            // forwards to the sink which sends the data over
            match connection.unbounded_send(ilp_packet_to_ws_message(
                request_id,
                Packet::Prepare(request.prepare),
            )) {
                Ok(_) => {
                    let (sender, receiver) = oneshot::channel();
                    (*self.pending_outgoing.lock()).insert(request_id, sender);
                    // Wrap the receiver with a timeout to ensure we do not
                    // wait too long if the other party has disconnected
                    // FIXME: this causes the test case to take 30s
                    let result = tokio::time::timeout(SEND_MSG_TIMEOUT, receiver).await;
                    let result = match result {
                        Ok(packet) => packet,
                        Err(err) => {
                            error!("Request timed out. Did the peer disconnect? Err: {}", err);
                            // Assume that such a long timeout means that the peer closed their
                            // connection with us, so we'll remove the pending request and the websocket
                            (*self.pending_outgoing.lock()).remove(&request_id);
                            self.close_connection(&request.to.id());
                            return Err(RejectBuilder {
                                code: ErrorCode::R00_TRANSFER_TIMED_OUT,
                                message: &[],
                                triggered_by: Some(&ilp_address),
                                data: &[],
                            }
                            .build());
                        }
                    };
                    // Drop the trigger here since we've gotten the response
                    // and don't need to keep the connections open if this was the
                    // last thing we were waiting for
                    let _ = keep_connections_open;
                    match result {
                        // This can be either a reject or a fulfill packet
                        Ok(packet) => packet,
                        Err(err) => {
                            error!(
                                "Sending request {} to account {} failed: {:?}",
                                request_id, account_id, err
                            );
                            Err(RejectBuilder {
                                code: ErrorCode::T00_INTERNAL_ERROR,
                                message: &[],
                                triggered_by: Some(&ilp_address),
                                data: &[],
                            }
                            .build())
                        }
                    }
                }
                Err(send_error) => {
                    error!(
                        "Error sending websocket message for request {} to account {}: {:?}",
                        request_id, account_id, send_error
                    );
                    Err(RejectBuilder {
                        code: ErrorCode::T00_INTERNAL_ERROR,
                        message: &[],
                        triggered_by: Some(&ilp_address),
                        data: &[],
                    }
                    .build())
                }
            }
        } else {
            if request.to.get_ilp_over_btp_url().is_some()
                || request.to.get_ilp_over_btp_outgoing_token().is_some()
            {
                trace!(
                    "No open connection for account: {}, forwarding request to the next service",
                    request.to.username()
                );
            }
            self.next.send_request(request).await
        }
    }
async fn send_request ( & mut self , request : OutgoingRequest < A > ) -> IlpResult { let account_id = request . to . id ( ) ; let found = self . connections . read ( ) . get ( & account_id ) . cloned ( ) ; if let Some ( connection ) = found { let request_id = random :: < u32 > ( ) ; let ilp_address = self . ilp_address . clone ( ) ; // Clone the trigger so that the connections stay open until we've // gotten the response to our outgoing request let keep_connections_open = self . close_all_connections . clone ( ) ; trace ! ( " " request_id request to username ( ) account_id ) ; // Connection is an unbounded sender which sends to the rx that // forwards to the sink which sends the data over match connection . unbounded_send ( ilp_packet_to_ws_message ( request_id , Packet :: Prepare ( request . prepare ) , ) ) { Ok ( _ ) => { let ( sender , receiver ) = oneshot :: channel ( ) ; ( * self . pending_outgoing . lock ( ) ) . insert ( request_id , sender ) ; // Wrap the receiver with a timeout to ensure we do not // wait too long if the other party has disconnected // FIXME: this causes the test case to take 30s let result = tokio :: time :: timeout ( SEND_MSG_TIMEOUT , receiver ) . await ; let result = match result { Ok ( packet ) => packet , Err ( err ) => { error ! ( " " err ) ; // Assume that such a long timeout means that the peer closed their // connection with us, so we'll remove the pending request and the websocket ( * self . pending_outgoing . lock ( ) ) . remove ( & request_id ) ; self . close_connection ( & request . to . id ( ) ) ; return Err ( RejectBuilder { code : ErrorCode :: R00_TRANSFER_TIMED_OUT , message : & [ ] , triggered_by : Some ( & ilp_address ) , data : & [ ] , } . build ( ) ) ; } } ; // Drop the trigger here since we've gotten the response // and don't need to keep the connections open if this was the // last thing we were waiting for let _ = keep_connections_open ; match result { // This can be either a reject or a fulfill packet Ok ( packet ) => packet , Err ( err ) => { error ! ( " " request_id account_id err ) ; Err ( RejectBuilder { code : ErrorCode :: T00_INTERNAL_ERROR , message : & [ ] , triggered_by : Some ( & ilp_address ) , data : & [ ] , } . build ( ) ) } } } Err ( send_error ) => { error ! ( " " request_id account_id send_error ) ; Err ( RejectBuilder { code : ErrorCode :: T00_INTERNAL_ERROR , message : & [ ] , triggered_by : Some ( & ilp_address ) , data : & [ ] , } . build ( ) ) } } } else { if request . to . get_ilp_over_btp_url ( ) . is_some ( ) || request . to . get_ilp_over_btp_outgoing_token ( ) . is_some ( ) { trace ! ( " " request to username ( ) ) ; } self . next . send_request ( request ) . await } }
Send an outgoing request to one of the open connections.
Send an outgoing request to one of the open connections .
&mut self
request
OutgoingRequest<A>