Struct StreamReactor 

pub(crate) struct StreamReactor {
    hopnum: Option<HopNum>,
    hop: CircHopOutbound,
    time_provider: DynTimeProvider,
    unique_id: UniqId,
    cell_rx: Receiver<StreamMsg>,
    bwd_tx: Sender<ReadyStreamMsg>,
    incoming: Arc<Mutex<Option<IncomingStreamRequestHandler>>>,
    inner: Arc<dyn StreamHandler>,
    memquota: CircuitAccount,
}

The stream reactor for a given hop.

Drives the application streams.

This reactor accepts StreamMsgs from the forward reactor over its Self::cell_rx MPSC channel, and delivers them to the corresponding stream entries in the stream map.

The local streams are polled from the main loop, and any ready messages are sent to the backward reactor over the bwd_tx MPSC channel for packaging and delivery.

Shuts downs down if an error occurs, or if the sending end of the cell_rx MPSC channel, i.e. the forward reactor, closes.

Fields

hopnum: Option<HopNum>

The hop this stream reactor is for.

This is None for relays.

hop: CircHopOutbound

The state of this circuit hop.

time_provider: DynTimeProvider

The time provider.

unique_id: UniqId

An identifier for logging about this reactor’s circuit.

cell_rx: Receiver<StreamMsg>

Receiver for Tor stream data that need to be delivered to a Tor stream.

The sender is in ForwardReactor, which will forward all cells carrying Tor stream data to us.

This serves a dual purpose:

• it enables the ForwardReactor to deliver Tor stream data received from the client
• it lets the StreamReactor know if the ForwardReactor has shut down: we select! on this MPSC channel in the main loop, so if the ForwardReactor shuts down, we will get EOS upon calling .next())

bwd_tx: Sender<ReadyStreamMsg>

Sender for sending Tor stream data to BackwardReactor.

incoming: Arc<Mutex<Option<IncomingStreamRequestHandler>>>

A handler for incoming streams.

Set to None if incoming streams are not allowed on this circuit.

This handler is shared with the HopMgr of this reactor, which can install a new handler at runtime (for example, in response to a CtrlMsg). The ability to update the handler after the reactor is launched is needed for onion services, where the incoming stream request handler only gets installed after the virtual hop is created.

inner: Arc<dyn StreamHandler>

A handler for customizing the stream reactor behavior.

memquota: CircuitAccount

Memory quota account

Implementations

impl StreamReactor

pub(crate) async fn run(self) -> Result<()>

Launch the reactor, and run until the circuit closes or we encounter an error.

Once this method returns, the circuit is dead and cannot be used again.

impl StreamReactor

pub(crate) fn new<R: Runtime>( runtime: R, hopnum: Option<HopNum>, hop: CircHopOutbound, unique_id: UniqId, cell_rx: Receiver<StreamMsg>, bwd_tx: Sender<ReadyStreamMsg>, inner: Arc<dyn StreamHandler>, incoming: Arc<Mutex<Option<IncomingStreamRequestHandler>>>, memquota: CircuitAccount, ) -> Self

Create a new StreamReactor.

async fn run_once(&mut self) -> StdResult<(), ReactorError>

Helper for run.

Polls the stream map for messages that need to be delivered to the other endpoint, and the cells_rx MPSC stream for stream messages received from the ForwardReactor that need to be delivered to the application streams.

async fn handle_reactor_cmd( &mut self, msg: StreamMsg, ) -> StdResult<(), ReactorError>

Handle a stream message sent to us by the forward reactor.

Delivers the message to its corresponding application stream.

fn handle_msg( &mut self, streamid: StreamId, msg: UnparsedRelayMsg, cell_counts_toward_windows: bool, ) -> StdResult<Option<AnyRelayMsgOuter>, ReactorError>

Handle a RELAY message that has a non-zero stream ID.

A returned message is one that we need to send back to the client.

fn handle_incoming_stream_request( &mut self, sid: StreamId, msg: UnparsedRelayMsg, ) -> StdResult<Option<AnyRelayMsgOuter>, ReactorError>

A helper for handling incoming stream requests.

Accepts the specified incoming stream request, by adding a new entry to our stream map.

Returns the cell we need to send back to the client, if an error occurred and the stream cannot be opened.

Returns None if everything went well (the CONNECTED response only comes if the external consumer of our Stream of incoming Tor streams is able to actually establish the connection to the address specified in the BEGIN).

Any error returned from this function will shut down the reactor.

fn should_reject_incoming<'a>( handler: &mut IncomingStreamRequestHandler, sid: StreamId, request: &IncomingStreamRequest, view: &CircHopSyncView<'a>, ) -> StdResult<Option<AnyRelayMsgOuter>, ReactorError>

Check if we should reject this incoming stream request or not.

Returns a cell we need to send back to the client if we must reject the request, or None if we are allowed to accept it. ` Any error returned from this function will shut down the reactor.

async fn handle_stream_event( &mut self, event: StreamEvent, ) -> StdResult<(), ReactorError>

Handle a StreamEvent.

async fn send_msg_to_bwd( &mut self, msg: AnyRelayMsgOuter, ) -> StdResult<(), ReactorError>

Wrap msg in ReadyStreamMsg, and send it to the backward reactor.