Struct RequestResponseInner 

pub struct RequestResponseInner<S: Sender<K>, R: Receiver, Req: Request, RS: RecipientSource<Req, K>, DS: DataSource<Req>, K: SignatureKey + 'static> {
    pub(crate) config: RequestResponseConfig,
    pub sender: S,
    pub recipient_source: RS,
    pub(crate) data_source: DS,
    pub(crate) active_requests: Arc<RwLock<HashMap<RequestHash, Vec<Waiter>>>>,
    pub(crate) next_waiter_id: AtomicU64,
    pub(crate) phantom_data: PhantomData<(K, R, Req, DS)>,
}

The inner implementation for the request-response protocol

Fields

config: RequestResponseConfig

The configuration of the protocol

sender: S

The sender to use for the protocol

recipient_source: RS

The recipient source to use for the protocol

data_source: DS

The data source to use for the protocol

active_requests: Arc<RwLock<HashMap<RequestHash, Vec<Waiter>>>>

The map of currently active, outgoing requests

next_waiter_id: AtomicU64

The id to assign to the next registered waiter

phantom_data: PhantomData<(K, R, Req, DS)>

Phantom data to help with type inference

Implementations

impl<S: Sender<K>, R: Receiver, Req: Request, RS: RecipientSource<Req, K>, DS: DataSource<Req>, K: SignatureKey + 'static> RequestResponseInner<S, R, Req, RS, DS, K>

pub async fn request_indefinitely<F, Fut, O>( self: &Arc<Self>, public_key: &K, private_key: &K::PrivateKey, estimated_request_ttl: Duration, request: Req, response_validation_fn: F, ) -> Result<O, RequestError>

where F: Fn(&Req, Req::Response) -> Fut + Send + Sync + Clone, Fut: Future<Output = Result<O>> + Send, O: Send,

Request something from the protocol indefinitely until we get a response or there was a critical error (e.g. the request could not be signed)

Errors

• If the request was invalid
• If there was a critical error (e.g. the channel was closed)

pub async fn request<F, Fut, O>( self: &Arc<Self>, request_message: RequestMessage<Req, K>, timeout_duration: Duration, response_validation_fn: F, ) -> Result<O, RequestError>

where F: Fn(&Req, Req::Response) -> Fut + Send + Sync, Fut: Future<Output = Result<O>> + Send, O: Send,

Request something from the protocol and wait for the first response that passes validation. Concurrent requests for the same data (determined by Blake3 hash of the request) share incoming responses, but each caller validates them independently and all of them make requests until their own timeout is reached.

Errors

• If the request times out
• If the request we sign is invalid

pub(crate) fn register_waiter( self: &Arc<Self>, request_hash: RequestHash, ) -> ResponseReceiver

Register a waiter for responses to the request with the given hash

pub(crate) async fn spawn_batched_sender( self: &Arc<Self>, request_message: RequestMessage<Req, K>, message: Bytes, timeout_duration: Duration, ) -> Result<AbortOnDropHandle<()>, RequestError>

Spawn the task that repeatedly sends a batched request to [config.request_batch_size] recipients at a time until timeout_duration elapses or the returned handle is dropped

pub(crate) async fn receiving_task(self: Arc<Self>, receiver: R)

The task responsible for receiving messages from the receiver and handling them

pub(crate) fn handle_message( self: &Arc<Self>, message: &[u8], incoming_requests: &mut IncomingRequests<K>, ) -> Result<()>

Route a message by its framing, deserializing only what will be acted on

pub(crate) fn handle_request( self: &Arc<Self>, request_message: RequestMessage<Req, K>, incoming_requests: &mut IncomingRequests<K>, )

Handle a request sent to us

pub(crate) fn handle_response(&self, request_hash: RequestHash, body: &[u8])

Handle a response sent to us: fan its serialized body out to every waiter registered for the request hash. Each waiter decodes the body itself, so a response nobody is waiting for costs no more than this lookup