fn induct_loopback_stream(&self, mut state: ReplicatedState) -> ReplicatedState {
        let loopback_stream = state.get_stream(&self.subnet_id);

        // All done if the loopback stream does not exist or is empty.
        if loopback_stream.is_none() || loopback_stream.unwrap().messages().is_empty() {
            return state;
        }
        let loopback_stream = loopback_stream.unwrap();

        // Precondition: ensure that the loopback stream has had signals for all
        // earlier messages.
        assert!(
            loopback_stream.signals_end() == loopback_stream.messages_begin(),
            "Expecting loopback stream signals to end ({}) where messages begin ({})",
            loopback_stream.signals_end(),
            loopback_stream.messages_begin()
        );

        // Remember the original messages end index, we may append reject responses.
        let loopback_stream_messages_end = loopback_stream.messages_end();

        // Wrap it within a StreamSlice.
        let loopback_stream_slice: StreamSlice = loopback_stream.clone().into();

        // Induct all messages. This will add signals to the loopback stream.
        let mut stream_slices = BTreeMap::new();
        stream_slices.insert(self.subnet_id, loopback_stream_slice);
        state = self.induct_stream_slices(state, stream_slices);

        let mut streams = state.take_streams();
        // We know for sure that the loopback stream exists, so it is safe to unwrap.
        let loopback_stream = streams.get_mut(&self.subnet_id).unwrap();
        // Garbage collect all initial messages.
        self.discard_messages_before(loopback_stream, loopback_stream_messages_end);
        state.put_streams(streams);

        state
    }
/// Inducts and consumes all messages from the subnet's loopback stream.
///
/// After the call completes, the loopback stream may only contain reject
/// responses, with all initial messages garbage collected and signals
/// generated for them.
Inducts and consumes all messages from the subnet's loopback stream.
After the call completes, the loopback stream may only contain reject
responses, with all initial messages garbage collected and signals
generated for them.
Inducts and consumes all messages from the subnet ' s loopback stream . After the call completes the loopback stream may only contain reject responses with all initial messages garbage collected and signals generated for them .
fn induct_loopback_stream(&self, mut state: ReplicatedState) -> ReplicatedState {
        let loopback_stream = state.get_stream(&self.subnet_id);
        // All done if the loopback stream does not exist or is empty.
        if loopback_stream.is_none() || loopback_stream.unwrap().messages().is_empty() {
            return state;
        }
        let loopback_stream = loopback_stream.unwrap();
        // Precondition: ensure that the loopback stream has had signals for all
        // earlier messages.
        assert!(
            loopback_stream.signals_end() == loopback_stream.messages_begin(),
            "Expecting loopback stream signals to end ({}) where messages begin ({})",
            loopback_stream.signals_end(),
            loopback_stream.messages_begin()
        );
        // Remember the original messages end index, we may append reject responses.
        let loopback_stream_messages_end = loopback_stream.messages_end();
        // Wrap it within a StreamSlice.
        let loopback_stream_slice: StreamSlice = loopback_stream.clone().into();
        // Induct all messages. This will add signals to the loopback stream.
        let mut stream_slices = BTreeMap::new();
        stream_slices.insert(self.subnet_id, loopback_stream_slice);
        state = self.induct_stream_slices(state, stream_slices);
        let mut streams = state.take_streams();
        // We know for sure that the loopback stream exists, so it is safe to unwrap.
        let loopback_stream = streams.get_mut(&self.subnet_id).unwrap();
        // Garbage collect all initial messages.
        self.discard_messages_before(loopback_stream, loopback_stream_messages_end);
        state.put_streams(streams);
        state
    }
fn induct_loopback_stream ( & self , mut state : ReplicatedState ) -> ReplicatedState { let loopback_stream = state . get_stream ( & self . subnet_id ) ; // All done if the loopback stream does not exist or is empty. if loopback_stream . is_none ( ) || loopback_stream . unwrap ( ) . messages ( ) . is_empty ( ) { return state ; } let loopback_stream = loopback_stream . unwrap ( ) ; // Precondition: ensure that the loopback stream has had signals for all // earlier messages. assert ! ( loopback_stream signals_end ( ) loopback_stream messages_begin ( ) " " loopback_stream signals_end ( ) loopback_stream messages_begin ( ) ) ; // Remember the original messages end index, we may append reject responses. let loopback_stream_messages_end = loopback_stream . messages_end ( ) ; // Wrap it within a StreamSlice. let loopback_stream_slice : StreamSlice = loopback_stream . clone ( ) . into ( ) ; // Induct all messages. This will add signals to the loopback stream. let mut stream_slices = BTreeMap :: new ( ) ; stream_slices . insert ( self . subnet_id , loopback_stream_slice ) ; state = self . induct_stream_slices ( state , stream_slices ) ; let mut streams = state . take_streams ( ) ; // We know for sure that the loopback stream exists, so it is safe to unwrap. let loopback_stream = streams . get_mut ( & self . subnet_id ) . unwrap ( ) ; // Garbage collect all initial messages. self . discard_messages_before ( loopback_stream , loopback_stream_messages_end ) ; state . put_streams ( streams ) ; state }
Inducts and consumes all messages from the subnet's loopback stream.
Inducts and consumes all messages from the subnet ' s loopback stream .
&self
state
ReplicatedState
&self
state
ReplicatedState