fn spawn_iterator_server(
        &mut self,
        iterator: ServerEnd<ArchiveIteratorMarker>,
    ) -> Result<fasync::Task<Result<(), Error>>, StreamError> {
        let stream_result = match self.start_log_stream() {
            Ok(r) => r,
            Err(e) => return Err(e),
        };
        let task = fasync::Task::spawn(async move {
            let mut iter_stream = iterator.into_stream()?;
            let mut result_stream = stream_result;

            while let Some(request) = iter_stream.next().await {
                match request? {
                    ArchiveIteratorRequest::GetNext { responder } => {
                        let logs = select! {
                            result = result_stream.select_next_some() => {
                                match result {
                                    Err(err) => {
                                        warn!(?err, "Data read error");
                                        responder.send(&mut Err(
                                            ArchiveIteratorError::DataReadFailed))?;
                                        continue;
                                    }
                                    Ok(log) => log,
                                }
                            }

                            complete => {
                                responder.send(&mut Ok(vec![]))?;
                                break;
                            }
                        };

                        let send_inline = logs_data_can_be_sent_inline(&logs);
                        let data = serde_json::to_string(&logs)?;
                        if send_inline {
                            // Short data => send the data within the message.
                            let response = vec![ArchiveIteratorEntry {
                                diagnostics_data: Some(DiagnosticsData::Inline(InlineData {
                                    data,
                                    truncated_chars: 0,
                                })),
                                ..ArchiveIteratorEntry::EMPTY
                            }];
                            responder.send(&mut Ok(response))?;
                        } else {
                            // Long data => create a socket, send the socket in the message, then
                            // write all the data on the other side of the socket.
                            let sock_opts = fuchsia_zircon::SocketOpts::STREAM;
                            let (socket, tx_socket) = fuchsia_zircon::Socket::create(sock_opts)?;
                            let mut tx_socket = fasync::Socket::from_socket(tx_socket)?;
                            let response = vec![ArchiveIteratorEntry {
                                diagnostics_data: Some(DiagnosticsData::Socket(socket)),
                                ..ArchiveIteratorEntry::EMPTY
                            }];
                            responder.send(&mut Ok(response))?;
                            tx_socket.write_all(data.as_bytes()).await?;
                        }
                    }
                }
            }

            Ok::<(), Error>(())
        });
        Ok(task)
    }
/// Provides an implementation of an ArchiveIterator server. Intended to be used by clients who
/// wish to spawn an ArchiveIterator server given an implementation of `start_log_stream`..
Provides an implementation of an ArchiveIterator server. Intended to be used by clients who
wish to spawn an ArchiveIterator server given an implementation of `start_log_stream`
Provides an implementation of an ArchiveIterator server . Intended to be used by clients who wish to spawn an ArchiveIterator server given an implementation of ` start_log_stream `
fn spawn_iterator_server(
        &mut self,
        iterator: ServerEnd<ArchiveIteratorMarker>,
    ) -> Result<fasync::Task<Result<(), Error>>, StreamError> {
        let stream_result = match self.start_log_stream() {
            Ok(r) => r,
            Err(e) => return Err(e),
        };
        let task = fasync::Task::spawn(async move {
            let mut iter_stream = iterator.into_stream()?;
            let mut result_stream = stream_result;
            while let Some(request) = iter_stream.next().await {
                match request? {
                    ArchiveIteratorRequest::GetNext { responder } => {
                        let logs = select! {
                            result = result_stream.select_next_some() => {
                                match result {
                                    Err(err) => {
                                        warn!(?err, "Data read error");
                                        responder.send(&mut Err(
                                            ArchiveIteratorError::DataReadFailed))?;
                                        continue;
                                    }
                                    Ok(log) => log,
                                }
                            }
                            complete => {
                                responder.send(&mut Ok(vec![]))?;
                                break;
                            }
                        };
                        let send_inline = logs_data_can_be_sent_inline(&logs);
                        let data = serde_json::to_string(&logs)?;
                        if send_inline {
                            // Short data => send the data within the message.
                            let response = vec![ArchiveIteratorEntry {
                                diagnostics_data: Some(DiagnosticsData::Inline(InlineData {
                                    data,
                                    truncated_chars: 0,
                                })),
                                ..ArchiveIteratorEntry::EMPTY
                            }];
                            responder.send(&mut Ok(response))?;
                        } else {
                            // Long data => create a socket, send the socket in the message, then
                            // write all the data on the other side of the socket.
                            let sock_opts = fuchsia_zircon::SocketOpts::STREAM;
                            let (socket, tx_socket) = fuchsia_zircon::Socket::create(sock_opts)?;
                            let mut tx_socket = fasync::Socket::from_socket(tx_socket)?;
                            let response = vec![ArchiveIteratorEntry {
                                diagnostics_data: Some(DiagnosticsData::Socket(socket)),
                                ..ArchiveIteratorEntry::EMPTY
                            }];
                            responder.send(&mut Ok(response))?;
                            tx_socket.write_all(data.as_bytes()).await?;
                        }
                    }
                }
            }
            Ok::<(), Error>(())
        });
        Ok(task)
    }
fn spawn_iterator_server ( & mut self , iterator : ServerEnd < ArchiveIteratorMarker > , ) -> Result < fasync :: Task < Result < ( ) , Error > > , StreamError > { let stream_result = match self . start_log_stream ( ) { Ok ( r ) => r , Err ( e ) => return Err ( e ) , } ; let task = fasync :: Task :: spawn ( async move { let mut iter_stream = iterator . into_stream ( ) ? ; let mut result_stream = stream_result ; while let Some ( request ) = iter_stream . next ( ) . await { match request ? { ArchiveIteratorRequest :: GetNext { responder } => { let logs = select ! { result result_stream select_next_some ( ) { match result { Err ( err ) { warn ( err " " ) responder send ( mut Err ( ArchiveIteratorError DataReadFailed ) ) continue } Ok ( log ) log } } complete { responder send ( mut Ok ( vec [ ] ) ) break } } ; let send_inline = logs_data_can_be_sent_inline ( & logs ) ; let data = serde_json :: to_string ( & logs ) ? ; if send_inline { // Short data => send the data within the message. let response = vec ! [ ArchiveIteratorEntry { diagnostics_data Some ( DiagnosticsData Inline ( InlineData { data truncated_chars 0 } ) ) ArchiveIteratorEntry EMPTY } ] ; responder . send ( & mut Ok ( response ) ) ? ; } else { // Long data => create a socket, send the socket in the message, then // write all the data on the other side of the socket. let sock_opts = fuchsia_zircon :: SocketOpts :: STREAM ; let ( socket , tx_socket ) = fuchsia_zircon :: Socket :: create ( sock_opts ) ? ; let mut tx_socket = fasync :: Socket :: from_socket ( tx_socket ) ? ; let response = vec ! [ ArchiveIteratorEntry { diagnostics_data Some ( DiagnosticsData Socket ( socket ) ) ArchiveIteratorEntry EMPTY } ] ; responder . send ( & mut Ok ( response ) ) ? ; tx_socket . write_all ( data . as_bytes ( ) ) . await ? ; } } } } Ok :: < ( ) , Error > ( ( ) ) } ) ; Ok ( task ) }
Provides an implementation of an ArchiveIterator server.
Provides an implementation of an ArchiveIterator server .
&mut self
iterator
ServerEnd<ArchiveIteratorMarker>
&mut self
iterator
ServerEnd<ArchiveIteratorMarker>