fn take_slice_impl(
        &mut self,
        subnet_id: SubnetId,
        begin: Option<&ExpectedIndices>,
        msg_limit: Option<usize>,
        byte_limit: Option<usize>,
    ) -> CertifiedSliceResult<Option<UnpackedStreamSlice>> {
        // Update the stream position in case we bail out early with no slice returned.
        begin.map(|begin| self.stream_positions.insert(subnet_id, begin.clone()));

        let (subnet_id, mut slice) = match self.slices.remove_entry(&subnet_id) {
            Some(entry) => entry,
            None => return Ok(None),
        };

        // GC first if explicit begin indices were provided.
        let original_message_count = slice.payload.len();
        if let Some(begin) = begin {
            slice = match slice.garbage_collect(begin)? {
                Some(slice) => slice,
                None => {
                    self.metrics
                        .observe_take_messages_gced(original_message_count);
                    return Ok(None);
                }
            };

            if let Some(actual_begin) = slice.payload.messages_begin() {
                if actual_begin != begin.message_index {
                    // Slice's `messages.begin` is past the requested stream index, bail out.
                    return Err(CertifiedSliceError::TakeBeforeSliceBegin);
                }
            }
        }

        let mut prefix_message_count = slice.payload.len();
        self.metrics
            .observe_take_messages_gced(original_message_count - prefix_message_count);
        let signals_end = slice.payload.header.signals_end();

        let (prefix, slice) = slice.take_prefix(msg_limit, byte_limit)?;

        // Put back the rest of the slice, if any.
        if let Some(slice) = slice {
            prefix_message_count -= slice.payload.len();
            self.slices.insert(subnet_id, slice);
        }

        if let Some(prefix) = prefix {
            // A prefix is being returned, update stream position accordingly.
            if let Some(stream_indices) = self.stream_positions.get_mut(&subnet_id) {
                stream_indices.message_index += StreamIndex::from(prefix_message_count as u64);
                stream_indices.signal_index = stream_indices.signal_index.max(signals_end);
            }
            Ok(Some(prefix))
        } else {
            Ok(None)
        }
    }
/// Helper function to allow easy instrumentation of `take_slice` results.
///
/// On success, returns a prefix respecting the given limits.
///
/// Returns `Err(InvalidPayload)` or `Err(WitnessPruningFailed)` if
/// `self.payload` is malformed. Returns `Err(TakeBeforeSliceBegin)` if
/// `begin`'s `message_index` is before the pooled slice's messages begin
/// index.
Helper function to allow easy instrumentation of `take_slice` results.
On success, returns a prefix respecting the given limits.


Helper function to allow easy instrumentation of ` take_slice ` results . On success returns a prefix respecting the given limits .
fn take_slice_impl(
        &mut self,
        subnet_id: SubnetId,
        begin: Option<&ExpectedIndices>,
        msg_limit: Option<usize>,
        byte_limit: Option<usize>,
    ) -> CertifiedSliceResult<Option<UnpackedStreamSlice>> {
        // Update the stream position in case we bail out early with no slice returned.
        begin.map(|begin| self.stream_positions.insert(subnet_id, begin.clone()));
        let (subnet_id, mut slice) = match self.slices.remove_entry(&subnet_id) {
            Some(entry) => entry,
            None => return Ok(None),
        };
        // GC first if explicit begin indices were provided.
        let original_message_count = slice.payload.len();
        if let Some(begin) = begin {
            slice = match slice.garbage_collect(begin)? {
                Some(slice) => slice,
                None => {
                    self.metrics
                        .observe_take_messages_gced(original_message_count);
                    return Ok(None);
                }
            };
            if let Some(actual_begin) = slice.payload.messages_begin() {
                if actual_begin != begin.message_index {
                    // Slice's `messages.begin` is past the requested stream index, bail out.
                    return Err(CertifiedSliceError::TakeBeforeSliceBegin);
                }
            }
        }
        let mut prefix_message_count = slice.payload.len();
        self.metrics
            .observe_take_messages_gced(original_message_count - prefix_message_count);
        let signals_end = slice.payload.header.signals_end();
        let (prefix, slice) = slice.take_prefix(msg_limit, byte_limit)?;
        // Put back the rest of the slice, if any.
        if let Some(slice) = slice {
            prefix_message_count -= slice.payload.len();
            self.slices.insert(subnet_id, slice);
        }
        if let Some(prefix) = prefix {
            // A prefix is being returned, update stream position accordingly.
            if let Some(stream_indices) = self.stream_positions.get_mut(&subnet_id) {
                stream_indices.message_index += StreamIndex::from(prefix_message_count as u64);
                stream_indices.signal_index = stream_indices.signal_index.max(signals_end);
            }
            Ok(Some(prefix))
        } else {
            Ok(None)
        }
    }
fn take_slice_impl ( & mut self , subnet_id : SubnetId , begin : Option < & ExpectedIndices > , msg_limit : Option < usize > , byte_limit : Option < usize > , ) -> CertifiedSliceResult < Option < UnpackedStreamSlice > > { // Update the stream position in case we bail out early with no slice returned. begin . map ( | begin | self . stream_positions . insert ( subnet_id , begin . clone ( ) ) ) ; let ( subnet_id , mut slice ) = match self . slices . remove_entry ( & subnet_id ) { Some ( entry ) => entry , None => return Ok ( None ) , } ; // GC first if explicit begin indices were provided. let original_message_count = slice . payload . len ( ) ; if let Some ( begin ) = begin { slice = match slice . garbage_collect ( begin ) ? { Some ( slice ) => slice , None => { self . metrics . observe_take_messages_gced ( original_message_count ) ; return Ok ( None ) ; } } ; if let Some ( actual_begin ) = slice . payload . messages_begin ( ) { if actual_begin != begin . message_index { // Slice's `messages.begin` is past the requested stream index, bail out. return Err ( CertifiedSliceError :: TakeBeforeSliceBegin ) ; } } } let mut prefix_message_count = slice . payload . len ( ) ; self . metrics . observe_take_messages_gced ( original_message_count - prefix_message_count ) ; let signals_end = slice . payload . header . signals_end ( ) ; let ( prefix , slice ) = slice . take_prefix ( msg_limit , byte_limit ) ? ; // Put back the rest of the slice, if any. if let Some ( slice ) = slice { prefix_message_count -= slice . payload . len ( ) ; self . slices . insert ( subnet_id , slice ) ; } if let Some ( prefix ) = prefix { // A prefix is being returned, update stream position accordingly. if let Some ( stream_indices ) = self . stream_positions . get_mut ( & subnet_id ) { stream_indices . message_index += StreamIndex :: from ( prefix_message_count as u64 ) ; stream_indices . signal_index = stream_indices . signal_index . max ( signals_end ) ; } Ok ( Some ( prefix ) ) } else { Ok ( None ) } }
Helper function to allow easy instrumentation of `take_slice` results.
Helper function to allow easy instrumentation of ` take_slice ` results .
&mut self
subnet_id
SubnetId
begin
Option<&ExpectedIndices>
msg_limit
Option<usize>
byte_limit
Option<usize>
&mut self
subnet_id
SubnetId
begin
Option<&ExpectedIndices>
msg_limit
Option<usize>
byte_limit
Option<usize>