/// Garbage collects the messages of an outgoing `Stream` based on the
/// signals in an incoming stream slice.
///
/// Panics if any of the incoming slices' `signals_end` refers to a
/// nonexistent (already garbage collected or future) message.
Garbage collects the messages of an outgoing `Stream` based on the
signals in an incoming stream slice.

Panics if any of the incoming slices' `signals_end` refers to a
nonexistent (already garbage collected or future) message.
Garbage collects the messages of an outgoing ` Stream ` based on the signals in an incoming stream slice . Panics if any of the incoming slices ' ` signals_end ` refers to a nonexistent ( already garbage collected or future ) message .
fn garbage_collect_messages(
        &self,
        stream: StreamHandle,
        remote_subnet: SubnetId,
        stream_slice: &StreamSlice,
    ) {
        assert!(
            stream.messages_begin() <= stream_slice.header().signals_end
                && stream_slice.header().signals_end <= stream.messages_end(),
            "Invalid signals in stream slice from subnet {}: signals_end {}, messages [{}, {})",
            remote_subnet,
            stream_slice.header().signals_end,
            stream.messages_begin(),
            stream.messages_end(),
        );
        {
            // Observe the enqueuing duration of all garbage collected messages.
            let mut time_in_stream_metrics = self.time_in_stream_metrics.lock().unwrap();
            time_in_stream_metrics.observe_message_durations(
                remote_subnet,
                stream.messages_begin()..stream_slice.header().signals_end,
            );
        }
        // Remove the consumed messages from our outgoing stream.
        self.discard_messages_before(stream, stream_slice.header().signals_end);
    }
fn garbage_collect_messages ( & self , stream : StreamHandle , remote_subnet : SubnetId , stream_slice : & StreamSlice , ) { assert ! ( stream messages_begin ( ) stream_slice header ( ) signals_end stream_slice header ( ) signals_end stream messages_end ( ) " " remote_subnet stream_slice header ( ) signals_end stream messages_begin ( ) stream messages_end ( ) ) ; { // Observe the enqueuing duration of all garbage collected messages. let mut time_in_stream_metrics = self . time_in_stream_metrics . lock ( ) . unwrap ( ) ; time_in_stream_metrics . observe_message_durations ( remote_subnet , stream . messages_begin ( ) .. stream_slice . header ( ) . signals_end , ) ; } // Remove the consumed messages from our outgoing stream. self . discard_messages_before ( stream , stream_slice . header ( ) . signals_end ) ; }
Garbage collects the messages of an outgoing `Stream` based on the
signals in an incoming stream slice.
Garbage collects the messages of an outgoing ` Stream ` based on the signals in an incoming stream slice .
&self
stream
StreamHandle
remote_subnet
SubnetId
stream_slice
&StreamSlice