Struct BroadcastOutputStream 

pub struct BroadcastOutputStream { /* private fields */ }

The output stream from a process. Either representing stdout or stderr.

This is the broadcast variant, allowing for multiple simultaneous consumers with the downside of inducing memory allocations not required when only one consumer is listening. For that case, prefer using the crate::output_stream::single_subscriber::SingleSubscriberOutputStream.

Implementations

impl BroadcastOutputStream

pub fn from_stream<S: AsyncRead + Unpin + Send + 'static>( stream: S, stream_name: &'static str, options: FromStreamOptions, ) -> BroadcastOutputStream

Creates a new broadcast output stream from an async read stream.

impl BroadcastOutputStream

pub fn inspect_chunks( &self, f: impl FnMut(Chunk) -> Next + Send + 'static, ) -> Inspector

Inspects chunks of output from the stream without storing them.

The provided closure is called for each chunk of data. Return Next::Continue to keep processing or Next::Break to stop.

pub fn inspect_lines( &self, f: impl FnMut(Cow<'_, str>) -> Next + Send + 'static, options: LineParsingOptions, ) -> Inspector

Inspects lines of output from the stream without storing them.

The provided closure is called for each line. Return Next::Continue to keep processing or Next::Break to stop.

pub fn inspect_lines_async<Fut>( &self, f: impl FnMut(Cow<'_, str>) -> Fut + Send + 'static, options: LineParsingOptions, ) -> Inspector

where Fut: Future<Output = Next> + Send,

Inspects lines of output from the stream without storing them, using an async closure.

The provided async closure is called for each line. Return Next::Continue to keep processing or Next::Break to stop.

impl BroadcastOutputStream

pub fn collect_chunks<S: Sink>( &self, into: S, collect: impl FnMut(Chunk, &mut S) + Send + 'static, ) -> Collector<S>

Collects chunks from the stream into a sink.

The provided closure is called for each chunk, with mutable access to the sink.

pub fn collect_chunks_async<S, C>(&self, into: S, collect: C) -> Collector<S>

where S: Sink, C: AsyncChunkCollector<S>,

Collects chunks from the stream into a sink using an async collector.

The provided async collector is called for each chunk, with mutable access to the sink.

Example

use tokio_process_tools::{AsyncChunkCollector, Chunk, Next, Process};

struct ExtendChunks;

impl AsyncChunkCollector<Vec<u8>> for ExtendChunks {
    async fn collect<'a>(&'a mut self, chunk: Chunk, bytes: &'a mut Vec<u8>) -> Next {
        bytes.extend_from_slice(chunk.as_ref());
        Next::Continue
    }
}

let process = Process::new(tokio::process::Command::new("some-command"))
    .spawn_broadcast()?;
let collector = process.stdout().collect_chunks_async(Vec::new(), ExtendChunks);

pub fn collect_lines<S: Sink>( &self, into: S, collect: impl FnMut(Cow<'_, str>, &mut S) -> Next + Send + 'static, options: LineParsingOptions, ) -> Collector<S>

Collects lines from the stream into a sink.

The provided closure is called for each line, with mutable access to the sink. Return Next::Continue to keep processing or Next::Break to stop.

pub fn collect_lines_async<S, C>( &self, into: S, collect: C, options: LineParsingOptions, ) -> Collector<S>

where S: Sink, C: AsyncLineCollector<S>,

Collects lines from the stream into a sink using an async collector.

The provided async collector is called for each line, with mutable access to the sink. Return Next::Continue to keep processing or Next::Break to stop.

Example

use std::borrow::Cow;
use tokio_process_tools::{AsyncLineCollector, LineParsingOptions, Next, Process};

struct PushLines;

impl AsyncLineCollector<Vec<String>> for PushLines {
    async fn collect<'a>(
        &'a mut self,
        line: Cow<'a, str>,
        lines: &'a mut Vec<String>,
    ) -> Next {
        lines.push(line.into_owned());
        Next::Continue
    }
}

let process = Process::new(tokio::process::Command::new("some-command"))
    .spawn_broadcast()?;
let collector = process.stdout().collect_lines_async(
    Vec::new(),
    PushLines,
    LineParsingOptions::default(),
);

pub fn collect_chunks_into_vec(&self) -> Collector<Vec<u8>>

Convenience method to collect all chunks into a Vec<u8>.

pub fn collect_lines_into_vec( &self, options: LineParsingOptions, ) -> Collector<Vec<String>>

Convenience method to collect all lines into a Vec<String>.

pub fn collect_chunks_into_write<W: Sink + AsyncWriteExt + Unpin>( &self, write: W, ) -> Collector<W>

Collects chunks into an async writer.

pub fn collect_lines_into_write<W: Sink + AsyncWriteExt + Unpin>( &self, write: W, options: LineParsingOptions, mode: LineWriteMode, ) -> Collector<W>

Collects lines into an async writer.

Parsed lines no longer include their trailing newline byte, so mode controls whether a \n delimiter should be reintroduced for each emitted line.

pub fn collect_chunks_into_write_mapped<W: Sink + AsyncWriteExt + Unpin, B: AsRef<[u8]> + Send>( &self, write: W, mapper: impl Fn(Chunk) -> B + Send + Sync + Copy + 'static, ) -> Collector<W>

Collects chunks into an async writer after mapping them with the provided function.

pub fn collect_lines_into_write_mapped<W: Sink + AsyncWriteExt + Unpin, B: AsRef<[u8]> + Send>( &self, write: W, mapper: impl Fn(Cow<'_, str>) -> B + Send + Sync + Copy + 'static, options: LineParsingOptions, mode: LineWriteMode, ) -> Collector<W>

Collects lines into an async writer after mapping them with the provided function.

mode applies after mapper: choose LineWriteMode::AsIs when the mapped output already contains delimiters, or LineWriteMode::AppendLf to append \n after each mapped line.

impl BroadcastOutputStream

pub async fn wait_for_line( &self, predicate: impl Fn(Cow<'_, str>) -> bool + Send + Sync + 'static, options: LineParsingOptions, ) -> WaitForLineResult

Waits for a line that matches the given predicate.

Returns WaitForLineResult::Matched if a matching line is found, or WaitForLineResult::StreamClosed if the stream ends first. This method never returns WaitForLineResult::Timeout; use BroadcastOutputStream::wait_for_line_with_timeout if you need a bounded wait.

Broadcast delivery is lossy under pressure. If this waiter lags and misses chunks, it discards any partial line in progress and resynchronizes at the next newline instead of matching across a gap.

pub async fn wait_for_line_with_timeout( &self, predicate: impl Fn(Cow<'_, str>) -> bool + Send + Sync + 'static, options: LineParsingOptions, timeout: Duration, ) -> WaitForLineResult

Waits for a line that matches the given predicate, with a timeout.

Returns WaitForLineResult::Matched if a matching line is found, WaitForLineResult::StreamClosed if the stream ends first, or WaitForLineResult::Timeout if the timeout expires first.

This is the only line-wait variant on this type that can return WaitForLineResult::Timeout.

Broadcast delivery is lossy under pressure. If this waiter lags and misses chunks, it discards any partial line in progress and resynchronizes at the next newline instead of matching across a gap.

Trait Implementations

impl Debug for BroadcastOutputStream

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for BroadcastOutputStream

fn drop(&mut self)

Executes the destructor for this type.

impl OutputStream for BroadcastOutputStream

fn chunk_size(&self) -> NumBytes

The maximum size of every chunk read by the backing stream_reader.

fn channel_capacity(&self) -> usize

The number of chunks held by the underlying async channel.

fn name(&self) -> &'static str

Type of stream. Can be “stdout” or “stderr”. But we do not guarantee this output. It should only be used for logging/debugging.