phasync
Streams and files / UnbufferedStream

Class

UnbufferedStream

A PSR-7 stream for a body that is still being produced, read once and in order.

Introduction#

One coroutine appends with UnbufferedStream::append and ends with UnbufferedStream::end; another reads. read() waits while there is nothing to read and the stream is not ended, and append() waits while more than $bufferSize bytes are unread. The stream is not seekable or writable, and getSize() is null.

Both sides wait with phasync::awaitFlag(): after $deadlockTimeout seconds without the other side making progress, the waiting call throws a TimeoutException.

The coroutine that created the stream cannot convert it to a string: __toString() returns an error message instead of waiting for itself.

Class synopsis#

class UnbufferedStream implements StreamInterface, \Stringable {
    /* Methods */
    public UnbufferedStream::__construct(
        int $bufferSize = 65536,
        float $deadlockTimeout = 60.0
    )
    public UnbufferedStream::__toString(): string
    public UnbufferedStream::close(): void
    public UnbufferedStream::detach()
    public UnbufferedStream::getSize(): ?int
    public UnbufferedStream::tell(): int
    public UnbufferedStream::eof(): bool
    public UnbufferedStream::isSeekable(): bool
    public UnbufferedStream::seek($offset, $whence = SEEK_SET): void
    public UnbufferedStream::rewind(): void
    public UnbufferedStream::isWritable(): bool
    public UnbufferedStream::write($string): int
    public UnbufferedStream::isReadable(): bool
    public UnbufferedStream::read($length): string
    public UnbufferedStream::getContents(): string
    public UnbufferedStream::getMetadata($key = null)
    public UnbufferedStream::append(string $chunk): void
    public UnbufferedStream::end(): void
}

Methods#

MethodDescription
__constructCreates an empty stream, owned by the current coroutine.
__toStringReturns the rest of the stream, or a message starting with "Stream Error" when it cannot: it was closed or detached, or this is the coroutine that created the stream.
closeDiscards the unread content.
detachSeparates any underlying resources from the stream.
getSizeGet the size of the stream if known.
tellReturns the current position of the file read/write pointer
eofReturns true if the stream is at the end of the stream.
isSeekableReturns whether or not the stream is seekable.
seekSeek to a position in the stream.
rewindSeek to the beginning of the stream.
isWritableReturns whether or not the stream is writable.
writeWrite data to the stream.
isReadableReturns whether or not the stream is readable.
readRead data from the stream.
getContentsReturns the remaining contents in a string
getMetadataGet stream metadata as an associative array or retrieve a specific key.
appendAdds $chunk to the end of the stream.
endDeclares that nothing more will be appended, which lets a reader reach the end of the stream.

Examples#

Example 1
$stream = new UnbufferedStream();

phasync::go(function () use ($stream) {
    foreach (['one', 'two', 'three'] as $chunk) {
        $stream->append($chunk . "\n");
        phasync::sleep(0.1);
    }
    $stream->end();
});

// A server sends this body to the client, reading it from another coroutine
$response = $response->withBody($stream);

See also#