Struct SingleSubscriberOutputStream 

pub struct SingleSubscriberOutputStream { /* private fields */ }

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

This is the single-subscriber variant, allowing for just one consumer. This has the upside of requiring as few memory allocations as possible. If multiple concurrent inspections are required, prefer using the output_stream::broadcast::BroadcastOutputSteam.

Implementations

impl SingleSubscriberOutputStream

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

Creates a new single subscriber output stream from an async read stream.

pub fn backpressure_control(&self) -> BackpressureControl

Returns the configured backpressure policy.

impl SingleSubscriberOutputStream

pub fn inspect_chunks( &self, f: impl Fn(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 SingleSubscriberOutputStream

pub fn collect_chunks<S: Sink>( &self, into: S, collect: impl Fn(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_single_subscriber()?;
let collector = process.stdout().collect_chunks_async(Vec::new(), ExtendChunks);

pub fn collect_lines<S: Sink>( &self, into: S, collect: impl Fn(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_single_subscriber()?;
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 SingleSubscriberOutputStream

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 SingleSubscriberOutputStream::wait_for_line_with_timeout if you need a bounded wait.

This method consumes the only receiver owned by the single-subscriber stream. After calling it, no other inspector or collector can be created for the same stream. Use the broadcast stream implementation if you need multiple consumers.

When chunks are dropped because BackpressureControl::DropLatestIncomingIfBufferFull is active, this waiter discards any partial line in progress and resynchronizes at the next newline instead of matching across the 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.

This method consumes the only receiver owned by the single-subscriber stream. After calling it, no other inspector or collector can be created for the same stream. Use the broadcast stream implementation if you need multiple consumers.

When chunks are dropped because BackpressureControl::DropLatestIncomingIfBufferFull is active, this waiter discards any partial line in progress and resynchronizes at the next newline instead of matching across the gap.

Trait Implementations

impl Debug for SingleSubscriberOutputStream

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

Formats the value using the given formatter.

impl Drop for SingleSubscriberOutputStream

fn drop(&mut self)

Executes the destructor for this type.

impl OutputStream for SingleSubscriberOutputStream

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.