Streams

The subset of WHATWG Streams that media processing uses, with readable, writable and transform streams of objects.

Where the specification returns a promise, these methods return an asyncio.Future. The operation starts when the method is called, before the future is awaited. Byte streams and BYOB readers aren’t implemented. Streams are created and used from a running asyncio event loop.

class webrtc.streams.QueuingStrategy(high_water_mark=None, size=None)

How a stream measures its queue, given to the constructor of a stream.

Parameters:

high_water_mark (float, optional) – The total size of queued chunks at which the stream signals backpressure. None takes the default of the stream.
size (callable, optional) – Returns the size of a chunk. Each chunk counts as 1 if None.

class webrtc.streams.ReadableStream(underlying_source=None, strategy=None)

A source of chunks to read, iterable with async for.

See MDN: ReadableStream.

It belongs to the first event loop using it until that loop closes: from another loop, its methods and those of its readers, writers and controllers raise webrtc.InvalidStateError.

Parameters:

underlying_source (optional) – An object or a dict with optional start(controller), pull(controller) and cancel(reason) callables. Each can be a plain or a coroutine function, and they get a ReadableStreamDefaultController.
strategy (QueuingStrategy, optional) – How the queue is measured. The default high water mark is 1.

Raises:

webrtc.InvalidRangeError – If the high water mark of the strategy is negative or NaN.

static from_(async_iterable)

Creates a stream that reads the items of an iterable, asynchronous or not.

Named from in the specification. Items are fetched one at a time, as they’re read. Canceling the stream closes the iterator if it’s a generator.

See MDN: ReadableStream/from_static.

Parameters:: async_iterable (AsyncIterable[_R] | Iterable[_R]) – The iterable or asynchronous iterable.
Returns:: ReadableStream – The stream of its items.
Raises:: TypeError – If the object isn’t iterable.
Return type:: ReadableStream[_R]

property locked: bool

Whether a reader holds the lock of the stream.

See MDN: ReadableStream/locked.

Type:: bool

get_reader(options=None)

Creates a reader, which locks the stream until ReadableStreamDefaultReader.release_lock().

See MDN: ReadableStream/getReader.

Parameters:: options (ReadableStreamGetReaderOptions, optional) – The kind of reader. Only default readers exist.
Returns:: ReadableStreamDefaultReader – The reader.
Raises:: TypeError – If the stream is locked or the mode is 'byob'.
Return type:: ReadableStreamDefaultReader[_T]

cancel(reason=None)

Cancels the stream. Queued chunks are dropped and the cancel of the source is called.

See MDN: ReadableStream/cancel.

Parameters:: reason (optional) – Passed to the cancel of the source.
Returns:: asyncio.Future – Done once the source is canceled. Fails with TypeError if the stream is locked, or with the error of an errored stream or of the source’s cancel.
Return type:: Future[None]

pipe_to(destination, options=None)

Reads every chunk and writes it to a writable stream, respecting its backpressure.

Both streams are locked while the pipe runs. There’s no abort signal, so cancel the returned future to stop the pipe. The destination is then aborted unless StreamPipeOptions.prevent_abort is set.

See MDN: ReadableStream/pipeTo.

Parameters:

destination (WritableStream) – The stream to write to.
options (StreamPipeOptions, optional) – What the pipe leaves untouched when it stops.

Returns:

asyncio.Future – Done once every chunk is written and the destination is closed, unless closing is prevented. It fails with the error that stopped the pipe, or with TypeError if either stream is locked.

Return type:

Future[None]

pipe_through(transform, options=None)

Sends the chunks of the stream through a transform and returns the readable side of the transform.

See MDN: ReadableStream/pipeThrough.

Parameters:

transform (ReadableWritablePair) – The pair, or any object with writable and readable streams, like TransformStream.
options (StreamPipeOptions, optional) – The options of the pipe, as for pipe_to().

Returns:

ReadableStream – The readable side of the transform.

Raises:

TypeError – If this stream or the writable side is locked.

Return type:

ReadableStream[_O]

tee()

Splits the stream into two branches that each read every chunk, and locks the stream.

Chunks aren’t copied, so both branches get the same objects. The stream is canceled once both branches are, and both branches error if it errors.

See MDN: ReadableStream/tee.

Returns:: list of ReadableStream – The two branches.
Raises:: TypeError – If the stream is locked.
Return type:: list[ReadableStream[_T]]

values(options=None)

Returns an asynchronous iterator over the chunks, which locks the stream. async for uses it too.

Stopping early cancels the stream when the iterator is finalized. Wrap it in contextlib.aclosing() to cancel right away.

Parameters:: options (ReadableStreamIteratorOptions, optional) – Whether stopping early leaves the stream open.
Returns:: collections.abc.AsyncIterator – The chunks.
Raises:: TypeError – If the stream is locked.
Return type:: AsyncIterator[_T]

class webrtc.streams.ReadableStreamDefaultController(stream, source, strategy)

Given to an underlying source to enqueue chunks into its stream, close it or error it.

See MDN: ReadableStreamDefaultController.

property desired_size: float | None

The room left in the queue before backpressure.

It’s negative when the queue is over-full, 0 once the stream is closed and None once it errored.

See MDN: ReadableStreamDefaultController/desiredSize.

Type:: float, optional

enqueue(chunk=None)

Adds a chunk to the stream. If a read is pending, the chunk goes straight to it.

See MDN: ReadableStreamDefaultController/enqueue.

Parameters:

chunk (_T | None) – The chunk.

Raises:

TypeError – If the stream is closed or a close was requested.
webrtc.InvalidRangeError – If the size of the chunk isn’t a finite non-negative number. This error and any exception from the size function also error the stream.

close()

Closes the stream once the chunks already queued are read.

See MDN: ReadableStreamDefaultController/close.

Raises:: TypeError – If the stream is closed or a close was requested.

error(e=None)

Errors the stream and drops its queue. Pending and later reads fail with the error.

Does nothing if the stream isn’t readable.

See MDN: ReadableStreamDefaultController/error.

Parameters:: e (optional) – The error. A value that isn’t an exception is raised as a TypeError.

class webrtc.streams.ReadableStreamDefaultReader(stream)

Reads the chunks of a stream, locking it until release_lock().

See MDN: ReadableStreamDefaultReader.

Parameters:: stream (ReadableStream) – The stream to lock.
Raises:: TypeError – If the stream is locked.

property closed: Future[None]

Done once the stream closes. It fails if the stream errors or the lock is released.

See MDN: ReadableStreamDefaultReader/closed.

Type:: asyncio.Future

read()

Reads the next chunk.

See MDN: ReadableStreamDefaultReader/read.

Returns:: asyncio.Future – A ReadableStreamReadResult, set once a chunk is available or the stream closes. It fails with the error of the stream, or with TypeError once the reader is released.
Return type:: Future[ReadableStreamReadResult[_T]]

cancel(reason=None)

Cancels the stream, as ReadableStream.cancel() does, while keeping the lock.

See MDN: ReadableStreamDefaultReader/cancel.

Parameters:: reason (optional) – Passed to the cancel of the source.
Returns:: asyncio.Future – Done once the source is canceled. Fails with TypeError once the reader is released.
Return type:: Future[None]

release_lock()

Unlocks the stream. Pending reads and closed fail with TypeError.

It does nothing if the reader is already released.

See MDN: ReadableStreamDefaultReader/releaseLock.

class webrtc.streams.ReadableStreamGetReaderOptions(mode=None)

The options of ReadableStream.get_reader().

See MDN: ReadableStream/getReader.

Parameters:: mode (webrtc.ReadableStreamReaderMode, optional) – The kind of reader. None gives a default reader. 'byob' is refused because there are no byte streams.
Raises:: ValueError – If the mode isn’t a member of webrtc.ReadableStreamReaderMode.

class webrtc.streams.ReadableStreamIteratorOptions(prevent_cancel=False)

The options of ReadableStream.values().

Parameters:: prevent_cancel (bool, optional) – Keep the stream open when the iteration stops before the end.

class webrtc.streams.ReadableStreamReadResult(value=None, done=False)

The result of ReadableStreamDefaultReader.read().

See MDN: ReadableStreamDefaultReader/read.

Parameters:

value (optional) – The chunk read. It’s None when done is set.
done (bool, optional) – True once the stream is closed and every chunk was read.

class webrtc.streams.ReadableWritablePair(readable, writable)

A writable stream and the readable stream its output comes out of, for ReadableStream.pipe_through().

See MDN: ReadableStream/pipeThrough.

Parameters:

readable (ReadableStream) – The output side.
writable (WritableStream) – The input side.

class webrtc.streams.StreamPipeOptions(prevent_close=False, prevent_abort=False, prevent_cancel=False)

The options of ReadableStream.pipe_to() and ReadableStream.pipe_through().

There’s no signal. To stop a pipe, cancel the future that ReadableStream.pipe_to() returns.

See MDN: ReadableStream/pipeTo.

Parameters:

prevent_close (bool, optional) – Don’t close the destination when the source closes.
prevent_abort (bool, optional) – Don’t abort the destination when the source errors.
prevent_cancel (bool, optional) – Don’t cancel the source when the destination errors.

class webrtc.streams.TransformStream(transformer=None, writable_strategy=None, readable_strategy=None)

A writable side and a readable side. Chunks written to one are transformed and read from the other.

See MDN: TransformStream.

It belongs to the first event loop using it until that loop closes: from another loop, its methods and those of its readers, writers and controllers raise webrtc.InvalidStateError.

Parameters:

transformer (optional) – An object or a dict with optional start(controller), transform(chunk, controller) and flush(controller) callables. Each can be a plain or a coroutine function, and they get a TransformStreamDefaultController. Without transform, chunks pass through unchanged.
writable_strategy (QueuingStrategy, optional) – How the queue of the writable side is measured. The default high water mark is 1.
readable_strategy (QueuingStrategy, optional) – How the queue of the readable side is measured. The default high water mark is 0.

Raises:

webrtc.InvalidRangeError – If the high water mark of a strategy is negative or NaN.

property readable: ReadableStream[_O]

The side to read the output chunks from.

See MDN: TransformStream/readable.

Type:: ReadableStream

property writable: WritableStream[_T]

The side to write the input chunks to.

See MDN: TransformStream/writable.

Type:: WritableStream

class webrtc.streams.TransformStreamDefaultController(stream)

Given to a transformer to enqueue output chunks, error the stream or terminate it.

See MDN: TransformStreamDefaultController.

property desired_size: float | None

The room left in the queue of the readable side.

It matches ReadableStreamDefaultController.desired_size of that side.

See MDN: TransformStreamDefaultController/desiredSize.

Type:: float, optional

enqueue(chunk=None)

Adds an output chunk to the readable side.

See MDN: TransformStreamDefaultController/enqueue.

Parameters:

chunk (_O | None) – The chunk.

Raises:

TypeError – If the readable side is closed or closing.
webrtc.InvalidRangeError – If the size of the chunk isn’t a finite non-negative number.

error(reason=None)

Errors both sides of the stream.

See MDN: TransformStreamDefaultController/error.

Parameters:: reason (optional) – The error. A value that isn’t an exception is raised as a TypeError.

terminate()

Ends the stream. Readers get the end of the stream and writers get a TypeError.

See MDN: TransformStreamDefaultController/terminate.

class webrtc.streams.WritableStream(underlying_sink=None, strategy=None)

A destination to write chunks to, one at a time, in order.

See MDN: WritableStream.

It belongs to the first event loop using it until that loop closes: from another loop, its methods and those of its readers, writers and controllers raise webrtc.InvalidStateError.

Parameters:

underlying_sink (optional) – An object or a dict with optional start(controller), write(chunk, controller), close() and abort(reason) callables. Each can be a plain or a coroutine function, and they get a WritableStreamDefaultController.
strategy (QueuingStrategy, optional) – How the queue is measured. The default high water mark is 1.

Raises:

webrtc.InvalidRangeError – If the high water mark of the strategy is negative or NaN.

property locked: bool

Whether a writer holds the lock of the stream.

See MDN: WritableStream/locked.

Type:: bool

get_writer()

Creates a writer, which locks the stream until WritableStreamDefaultWriter.release_lock().

See MDN: WritableStream/getWriter.

Returns:: WritableStreamDefaultWriter – The writer.
Raises:: TypeError – If the stream is locked.
Return type:: WritableStreamDefaultWriter[_W_contra]

close()

Closes the stream after the chunks already written, then calls the close of the sink.

See MDN: WritableStream/close.

Returns:: asyncio.Future – Done once the sink is closed. It fails with TypeError if the stream is locked, closed or closing, or with the error that stops the stream.
Return type:: Future[None]

abort(reason=None)

Aborts the stream. Queued writes fail, the stream errors and the abort of the sink is called.

It doesn’t wait for a write in progress. Pending writes fail with reason if it’s an exception, and otherwise with an asyncio.CancelledError that holds it.

See MDN: WritableStream/abort.

Parameters:: reason (optional) – Passed to the abort of the sink.
Returns:: asyncio.Future – Done once the sink is aborted, or right away for a closed or errored stream. It fails with TypeError if the stream is locked, or with the sink’s abort error.
Return type:: Future[None]

class webrtc.streams.WritableStreamDefaultController(stream, sink, strategy)

Given to an underlying sink to error its stream.

See MDN: WritableStreamDefaultController.

error(e=None)

Errors the stream, so queued and later writes fail with the error.

It does nothing unless the stream is writable.

See MDN: WritableStreamDefaultController/error.

Parameters:: e (optional) – The error. A value that isn’t an exception is raised as a TypeError.

class webrtc.streams.WritableStreamDefaultWriter(stream)

Writes chunks to a stream, locking it until release_lock().

See MDN: WritableStreamDefaultWriter.

Parameters:: stream (WritableStream) – The stream to lock.
Raises:: TypeError – If the stream is locked.

property closed: Future[None]

Done once the stream closes. It fails if the stream errors or the lock is released.

See MDN: WritableStreamDefaultWriter/closed.

Type:: asyncio.Future

property ready: Future[None]

Done while the queue is below its high water mark.

Under backpressure it’s replaced by a pending future. It fails if the stream errors or the lock is released.

See MDN: WritableStreamDefaultWriter/ready.

Type:: asyncio.Future

property desired_size: float | None

The room left in the queue before backpressure.

It’s 0 once the stream is closed and None once it’s errored or erroring.

See MDN: WritableStreamDefaultWriter/desiredSize.

Raises:: TypeError – If the writer is released.
Type:: float, optional

write(chunk=None)

Queues a chunk for the write of the sink.

See MDN: WritableStreamDefaultWriter/write.

Parameters:: chunk (_W_contra | None) – The chunk.
Returns:: asyncio.Future – Done once the sink wrote the chunk. It fails with the error of the sink or the stream, or with TypeError if the stream is closing or the writer is released.
Return type:: Future[None]

close()

Closes the stream, as WritableStream.close() does, while keeping the lock.

See MDN: WritableStreamDefaultWriter/close.

Returns:: asyncio.Future – Done once the sink is closed. It fails with TypeError once the writer is released.
Return type:: Future[None]

abort(reason=None)

Aborts the stream, as WritableStream.abort() does, while keeping the lock.

See MDN: WritableStreamDefaultWriter/abort.

Parameters:: reason (optional) – Passed to the abort of the sink.
Returns:: asyncio.Future – Done once the sink is aborted. It fails with TypeError once the writer is released.
Return type:: Future[None]

release_lock()

Unlocks the stream. ready and closed fail with TypeError.

Writes already queued go on. It does nothing if the writer is already released.

See MDN: WritableStreamDefaultWriter/releaseLock.