Repository files navigation

Event-driven readable and writable streams for non-blocking I/O inReactPHP.

Development version: This branch contains the code for the upcoming v3release. For the code of the current stable v1 release, check out the1.x branch.

The upcoming v3 release will be the way forward for this package. However,we will still actively support v1 for those not yet on the latest version.See alsoinstallation instructions for more details.

In order to make theEventLoopeasier to use, this component introduces the powerful concept of "streams".Streams allow you to efficiently process huge amounts of data (such as a multiGigabyte file download) in small chunks without having to store everything inmemory at once.They are very similar to the streams found in PHP itself,but have an interface more suited for async, non-blocking I/O.

Table of contents

• Stream usage
  • ReadableStreamInterface
    • data event
    • end event
    • error event
    • close event
    • isReadable()
    • pause()
    • resume()
    • pipe()
    • close()
  • WritableStreamInterface
    • drain event
    • pipe event
    • error event
    • close event
    • isWritable()
    • write()
    • end()
    • close()
  • DuplexStreamInterface
• Creating streams
  • ReadableResourceStream
  • WritableResourceStream
  • DuplexResourceStream
  • ThroughStream
  • CompositeStream
• Usage
• Install
• Tests
• License
• More

ReactPHP uses the concept of "streams" throughout its ecosystem to provide aconsistent higher-level abstraction for processing streams of arbitrary datacontents and size.While a stream itself is a quite low-level concept, it can be used as a powerfulabstraction to build higher-level components and protocols on top.

If you're new to this concept, it helps to think of them as a water pipe:You can consume water from a source or you can produce water and forward (pipe)it to any destination (sink).

Similarly, streams can either be

• readable (such asSTDIN terminal input) or
• writable (such asSTDOUT terminal output) or
• duplex (both readableand writable, such as a TCP/IP connection)

Accordingly, this package defines the following three interfaces

• ReadableStreamInterface
• WritableStreamInterface
• DuplexStreamInterface

TheReadableStreamInterface is responsible for providing an interface forread-only streams and the readable side of duplex streams.

Besides defining a few methods, this interface also implements theEventEmitterInterface which allows you to react to certain events.

The event callback functions MUST be a validcallable that obeys strictparameter definitions and MUST accept event parameters exactly as documented.The event callback functions MUST NOT throw anException.The return value of the event callback functions will be ignored and has noeffect, so for performance reasons you're recommended to not return anyexcessive data structures.

Every implementation of this interface MUST follow these event semantics inorder to be considered a well-behaving stream.

Note that higher-level implementations of this interface may choose todefine additional events with dedicated semantics not defined as part ofthis low-level stream specification. Conformance with these event semanticsis out of scope for this interface, so you may also have to refer to thedocumentation of such a higher-level implementation.

Thedata event will be emitted whenever some data was read/receivedfrom this source stream.The event receives a single mixed argument for incoming data.

$stream->on('data',function (mixed$data):void {echo$data;});

This event MAY be emitted any number of times, which may be zero times ifthis stream does not send any data at all.It SHOULD not be emitted after anend orclose event.

The given$data argument may be of mixed type, but it's usuallyrecommended it SHOULD be astring value or MAY use a type that allowsrepresentation as astring for maximum compatibility.

Many common streams (such as a TCP/IP connection or a file-based stream)will emit the raw (binary) payload data that is received over the wire aschunks ofstring values.

Due to the stream-based nature of this, the sender may send any numberof chunks with varying sizes. There are no guarantees that these chunkswill be received with the exact same framing the sender intended to send.In other words, many lower-level protocols (such as TCP/IP) transfer thedata in chunks that may be anywhere between single-byte values to severaldozens of kilobytes. You may want to apply a higher-level protocol tothese low-level data chunks in order to achieve proper message framing.

Theend event will be emitted once the source stream has successfullyreached the end of the stream (EOF).

$stream->on('end',function ():void {echo'END';});

This event SHOULD be emitted once or never at all, depending on whethera successful end was detected.It SHOULD NOT be emitted after a previousend orclose event.It MUST NOT be emitted if the stream closes due to a non-successfulend, such as after a previouserror event.

After the stream is ended, it MUST switch to non-readable mode,see alsoisReadable().

This event will only be emitted if theend was reached successfully,not if the stream was interrupted by an unrecoverable error or explicitlyclosed. Not all streams know this concept of a "successful end".Many use-cases involve detecting when the stream closes (terminates)instead, in this case you should use theclose event.After the stream emits anend event, it SHOULD usually be followed by aclose event.

Many common streams (such as a TCP/IP connection or a file-based stream)will emit this event if either the remote side closes the connection ora file handle was successfully read until reaching its end (EOF).

Note that this event should not be confused with theend() method.This event defines a successful endreading from a source stream, whiletheend() method defineswriting a successful end to a destinationstream.

Theerror event will be emitted once a fatal error occurs, usually whiletrying to read from this stream.The event receives a singleException argument for the error instance.

$server->on('error',function (Exception$e):void {echo'Error:' .$e->getMessage() .PHP_EOL;});

This event SHOULD be emitted once the stream detects a fatal error, suchas a fatal transmission error or after an unexpecteddata or prematureend event.It SHOULD NOT be emitted after a previouserror,end orclose event.It MUST NOT be emitted if this is not a fatal error condition, such asa temporary network issue that did not cause any data to be lost.

After the stream errors, it MUST close the stream and SHOULD thus befollowed by aclose event and then switch to non-readable mode, seealsoclose() andisReadable().

Many common streams (such as a TCP/IP connection or a file-based stream)only deal with data transmission and do not make assumption about databoundaries (such as unexpecteddata or prematureend events).In other words, many lower-level protocols (such as TCP/IP) may chooseto only emit this for a fatal transmission error once and will thenclose (terminate) the stream in response.

If this stream is aDuplexStreamInterface, you should also noticehow the writable side of the stream also implements anerror event.In other words, an error may occur while either reading or writing thestream which should result in the same error processing.

Theclose event will be emitted once the stream closes (terminates).

$stream->on('close',function ():void {echo'CLOSED';});

This event SHOULD be emitted once or never at all, depending on whetherthe stream ever terminates.It SHOULD NOT be emitted after a previousclose event.

After the stream is closed, it MUST switch to non-readable mode,see alsoisReadable().

Unlike theend event, this event SHOULD be emitted whenever the streamcloses, irrespective of whether this happens implicitly due to anunrecoverable error or explicitly when either side closes the stream.If you only want to detect asuccessful end, you should use theendevent instead.

Many common streams (such as a TCP/IP connection or a file-based stream)will likely choose to emit this event after reading asuccessfulendevent or after a fatal transmissionerror event.

If this stream is aDuplexStreamInterface, you should also noticehow the writable side of the stream also implements aclose event.In other words, after receiving this event, the stream MUST switch intonon-writable AND non-readable mode, see alsoisWritable().Note that this event should not be confused with theend event.

TheisReadable(): bool method can be used tocheck whether this stream is in a readable state (not closed already).

This method can be used to check if the stream still accepts incomingdata events or if it is ended or closed already.Once the stream is non-readable, no furtherdata orend events SHOULDbe emitted.

assert($stream->isReadable() ===false);$stream->on('data',assertNeverCalled());$stream->on('end',assertNeverCalled());

A successfully opened stream always MUST start in readable mode.

Once the stream ends or closes, it MUST switch to non-readable mode.This can happen any time, explicitly throughclose() orimplicitly due to a remote close or an unrecoverable transmission error.Once a stream has switched to non-readable mode, it MUST NOT transitionback to readable mode.

If this stream is aDuplexStreamInterface, you should also noticehow the writable side of the stream also implements anisWritable()method. Unless this is a half-open duplex stream, they SHOULD usuallyhave the same return value.

Thepause(): void method can be used topause reading incoming data events.

Removes the data source file descriptor from the event loop. Thisallows you to throttle incoming data.

Unless otherwise noted, a successfully opened stream SHOULD NOT startin paused state.

Once the stream is paused, no futherdata orend events SHOULDbe emitted.

$stream->pause();$stream->on('data',assertShouldNeverCalled());$stream->on('end',assertShouldNeverCalled());

This method is advisory-only, though generally not recommended, thestream MAY continue emittingdata events.

You can continue processing events by callingresume() again.

Note that both methods can be called any number of times, in particularcallingpause() more than once SHOULD NOT have any effect.

See alsoresume().

Theresume(): void method can be used toresume reading incoming data events.

Re-attach the data source after a previouspause().

$stream->pause();Loop::addTimer(1.0,function ()use ($stream):void {$stream->resume();});

Note that both methods can be called any number of times, in particularcallingresume() without a priorpause() SHOULD NOT have any effect.

See alsopause().

Thepipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface method can be used topipe all the data from this readable source into the given writable destination.

Automatically sends all incoming data to the destination.Automatically throttles the source based on what the destination can handle.

$source->pipe($dest);

Similarly, you can also pipe an instance implementingDuplexStreamInterfaceinto itself in order to write back all the data that is received.This may be a useful feature for a TCP/IP echo service:

$connection->pipe($connection);

This method returns the destination stream as-is, which can be used toset up chains of piped streams:

$source->pipe($decodeGzip)->pipe($filterBadWords)->pipe($dest);

By default, this will callend() on the destination stream once thesource stream emits anend event. This can be disabled like this:

$source->pipe($dest, ['end' =>false]);

Note that this only applies to theend event.If anerror or explicitclose event happens on the source stream,you'll have to manually close the destination stream:

$source->pipe($dest);$source->on('close',function ()use ($dest):void {$dest->end('BYE!');});

If the source stream is not readable (closed state), then this is a NO-OP.

$source->close();$source->pipe($dest);// NO-OP

If the destinantion stream is not writable (closed state), then this will simplythrottle (pause) the source stream:

$dest->close();$source->pipe($dest);// calls $source->pause()

Similarly, if the destination stream is closed while the pipe is stillactive, it will also throttle (pause) the source stream:

$source->pipe($dest);$dest->close();// calls $source->pause()

Once the pipe is set up successfully, the destination stream MUST emitapipe event with this source stream an event argument.

Theclose(): void method can be used toclose the stream (forcefully).

This method can be used to (forcefully) close the stream.

$stream->close();

Once the stream is closed, it SHOULD emit aclose event.Note that this event SHOULD NOT be emitted more than once, in particularif this method is called multiple times.

After calling this method, the stream MUST switch into a non-readablemode, see alsoisReadable().This means that no furtherdata orend events SHOULD be emitted.

$stream->close();assert($stream->isReadable() ===false);$stream->on('data',assertNeverCalled());$stream->on('end',assertNeverCalled());

If this stream is aDuplexStreamInterface, you should also noticehow the writable side of the stream also implements aclose() method.In other words, after calling this method, the stream MUST switch intonon-writable AND non-readable mode, see alsoisWritable().Note that this method should not be confused with theend() method.

TheWritableStreamInterface is responsible for providing an interface forwrite-only streams and the writable side of duplex streams.

Besides defining a few methods, this interface also implements theEventEmitterInterface which allows you to react to certain events.

The event callback functions MUST be a validcallable that obeys strictparameter definitions and MUST accept event parameters exactly as documented.The event callback functions MUST NOT throw anException.The return value of the event callback functions will be ignored and has noeffect, so for performance reasons you're recommended to not return anyexcessive data structures.

Every implementation of this interface MUST follow these event semantics inorder to be considered a well-behaving stream.

Note that higher-level implementations of this interface may choose todefine additional events with dedicated semantics not defined as part ofthis low-level stream specification. Conformance with these event semanticsis out of scope for this interface, so you may also have to refer to thedocumentation of such a higher-level implementation.

Thedrain event will be emitted whenever the write buffer became fullpreviously and is now ready to accept more data.

$stream->on('drain',function ()use ($stream):void {echo'Stream is now ready to accept more data';});

This event SHOULD be emitted once every time the buffer became fullpreviously and is now ready to accept more data.In other words, this event MAY be emitted any number of times, which maybe zero times if the buffer never became full in the first place.This event SHOULD NOT be emitted if the buffer has not become fullpreviously.

This event is mostly used internally, see alsowrite() for more details.

Thepipe event will be emitted whenever a readable stream ispipe()dinto this stream.The event receives a singleReadableStreamInterface argument for thesource stream.

$stream->on('pipe',function (ReadableStreamInterface$source)use ($stream):void {echo'Now receiving piped data';// explicitly close target if source emits an error$source->on('error',function ()use ($stream):void {$stream->close();    });});$source->pipe($stream);

This event MUST be emitted once for each readable stream that issuccessfully piped into this destination stream.In other words, this event MAY be emitted any number of times, which maybe zero times if no stream is ever piped into this stream.This event MUST NOT be emitted if either the source is not readable(closed already) or this destination is not writable (closed already).

This event is mostly used internally, see alsopipe() for more details.

Theerror event will be emitted once a fatal error occurs, usually whiletrying to write to this stream.The event receives a singleException argument for the error instance.

$stream->on('error',function (Exception$e):void {echo'Error:' .$e->getMessage() .PHP_EOL;});

This event SHOULD be emitted once the stream detects a fatal error, suchas a fatal transmission error.It SHOULD NOT be emitted after a previouserror orclose event.It MUST NOT be emitted if this is not a fatal error condition, such asa temporary network issue that did not cause any data to be lost.

After the stream errors, it MUST close the stream and SHOULD thus befollowed by aclose event and then switch to non-writable mode, seealsoclose() andisWritable().

Many common streams (such as a TCP/IP connection or a file-based stream)only deal with data transmission and may chooseto only emit this for a fatal transmission error once and will thenclose (terminate) the stream in response.

If this stream is aDuplexStreamInterface, you should also noticehow the readable side of the stream also implements anerror event.In other words, an error may occur while either reading or writing thestream which should result in the same error processing.

Theclose event will be emitted once the stream closes (terminates).

$stream->on('close',function ():void {echo'CLOSED';});

This event SHOULD be emitted once or never at all, depending on whetherthe stream ever terminates.It SHOULD NOT be emitted after a previousclose event.

After the stream is closed, it MUST switch to non-writable mode,see alsoisWritable().

This event SHOULD be emitted whenever the stream closes, irrespective ofwhether this happens implicitly due to an unrecoverable error orexplicitly when either side closes the stream.

Many common streams (such as a TCP/IP connection or a file-based stream)will likely choose to emit this event after flushing the buffer fromtheend() method, after receiving asuccessfulend event or aftera fatal transmissionerror event.

If this stream is aDuplexStreamInterface, you should also noticehow the readable side of the stream also implements aclose event.In other words, after receiving this event, the stream MUST switch intonon-writable AND non-readable mode, see alsoisReadable().Note that this event should not be confused with theend event.

TheisWritable(): bool method can be used tocheck whether this stream is in a writable state (not closed already).

This method can be used to check if the stream still accepts writingany data or if it is ended or closed already.Writing any data to a non-writable stream is a NO-OP:

assert($stream->isWritable() ===false);$stream->write('end');// NO-OP$stream->end('end');// NO-OP

A successfully opened stream always MUST start in writable mode.

Once the stream ends or closes, it MUST switch to non-writable mode.This can happen any time, explicitly throughend() orclose() orimplicitly due to a remote close or an unrecoverable transmission error.Once a stream has switched to non-writable mode, it MUST NOT transitionback to writable mode.

If this stream is aDuplexStreamInterface, you should also noticehow the readable side of the stream also implements anisReadable()method. Unless this is a half-open duplex stream, they SHOULD usuallyhave the same return value.

Thewrite(mixed $data): bool method can be used towrite some data into the stream.

A successful write MUST be confirmed with a booleantrue, which meansthat either the data was written (flushed) immediately or is buffered andscheduled for a future write. Note that this interface gives you nocontrol over explicitly flushing the buffered data, as finding theappropriate time for this is beyond the scope of this interface and leftup to the implementation of this interface.

Many common streams (such as a TCP/IP connection or file-based stream)may choose to buffer all given data and schedule a future flush by usingan underlying EventLoop to check when the resource is actually writable.

If a stream cannot handle writing (or flushing) the data, it SHOULD emitanerror event and MAYclose() the stream if it can not recover fromthis error.

If the internal buffer is full after adding$data, thenwrite()SHOULD returnfalse, indicating that the caller should stop sendingdata until the buffer drains.The stream SHOULD send adrain event once the buffer is ready to acceptmore data.

Similarly, if the stream is not writable (already in a closed state)it MUST NOT process the given$data and SHOULD returnfalse,indicating that the caller should stop sending data.

The given$data argument MAY be of mixed type, but it's usuallyrecommended it SHOULD be astring value or MAY use a type that allowsrepresentation as astring for maximum compatibility.

Many common streams (such as a TCP/IP connection or a file-based stream)will only accept the raw (binary) payload data that is transferred overthe wire as chunks ofstring values.

Due to the stream-based nature of this, the sender may send any numberof chunks with varying sizes. There are no guarantees that these chunkswill be received with the exact same framing the sender intended to send.In other words, many lower-level protocols (such as TCP/IP) transfer thedata in chunks that may be anywhere between single-byte values to severaldozens of kilobytes. You may want to apply a higher-level protocol tothese low-level data chunks in order to achieve proper message framing.

Theend(mixed $data = null): void method can be used tosuccessfully end the stream (after optionally sending some final data).

This method can be used to successfully end the stream, i.e. closethe stream after sending out all data that is currently buffered.

$stream->write('hello');$stream->write('world');$stream->end();

If there's no data currently buffered and nothing to be flushed, thenthis method MAYclose() the stream immediately.

If there's still data in the buffer that needs to be flushed first, thenthis method SHOULD try to write out this data and only thenclose()the stream.Once the stream is closed, it SHOULD emit aclose event.

Note that this interface gives you no control over explicitly flushingthe buffered data, as finding the appropriate time for this is beyond thescope of this interface and left up to the implementation of thisinterface.

Many common streams (such as a TCP/IP connection or file-based stream)may choose to buffer all given data and schedule a future flush by usingan underlying EventLoop to check when the resource is actually writable.

You can optionally pass some final data that is written to the streambefore ending the stream. If a non-null value is given as$data, thenthis method will behave just like callingwrite($data) before endingwith no data.

// shorter version$stream->end('bye');// same as longer version$stream->write('bye');$stream->end();

After calling this method, the stream MUST switch into a non-writablemode, see alsoisWritable().This means that no further writes are possible, so any additionalwrite() orend() calls have no effect.

$stream->end();assert($stream->isWritable() ===false);$stream->write('nope');// NO-OP$stream->end();// NO-OP

If this stream is aDuplexStreamInterface, calling this method SHOULDalso end its readable side, unless the stream supports half-open mode.In other words, after calling this method, these streams SHOULD switchinto non-writable AND non-readable mode, see alsoisReadable().This implies that in this case, the stream SHOULD NOT emit anydataorend events anymore.Streams MAY choose to use thepause() method logic for this, butspecial care may have to be taken to ensure a following call to theresume() method SHOULD NOT continue emitting readable events.

Note that this method should not be confused with theclose() method.

Theclose(): void method can be used toclose the stream (forcefully).

This method can be used to forcefully close the stream, i.e. closethe stream without waiting for any buffered data to be flushed.If there's still data in the buffer, this data SHOULD be discarded.

$stream->close();

Once the stream is closed, it SHOULD emit aclose event.Note that this event SHOULD NOT be emitted more than once, in particularif this method is called multiple times.

After calling this method, the stream MUST switch into a non-writablemode, see alsoisWritable().This means that no further writes are possible, so any additionalwrite() orend() calls have no effect.

$stream->close();assert($stream->isWritable() ===false);$stream->write('nope');// NO-OP$stream->end();// NO-OP

Note that this method should not be confused with theend() method.Unlike theend() method, this method does not take care of any existingbuffers and simply discards any buffer contents.Likewise, this method may also be called after callingend() on astream in order to stop waiting for the stream to flush its final data.

$stream->end();Loop::addTimer(1.0,function ()use ($stream):void {$stream->close();});

If this stream is aDuplexStreamInterface, you should also noticehow the readable side of the stream also implements aclose() method.In other words, after calling this method, the stream MUST switch intonon-writable AND non-readable mode, see alsoisReadable().

TheDuplexStreamInterface is responsible for providing an interface forduplex streams (both readable and writable).

It builds on top of the existing interfaces for readable and writable streamsand follows the exact same method and event semantics.If you're new to this concept, you should look into theReadableStreamInterface andWritableStreamInterface first.

Besides defining a few methods, this interface also implements theEventEmitterInterface which allows you to react to the same events definedon theReadbleStreamInterface andWritableStreamInterface.

The event callback functions MUST be a validcallable that obeys strictparameter definitions and MUST accept event parameters exactly as documented.The event callback functions MUST NOT throw anException.The return value of the event callback functions will be ignored and has noeffect, so for performance reasons you're recommended to not return anyexcessive data structures.

Every implementation of this interface MUST follow these event semantics inorder to be considered a well-behaving stream.

Note that higher-level implementations of this interface may choose todefine additional events with dedicated semantics not defined as part ofthis low-level stream specification. Conformance with these event semanticsis out of scope for this interface, so you may also have to refer to thedocumentation of such a higher-level implementation.

See alsoReadableStreamInterface andWritableStreamInterface for more details.

ReactPHP uses the concept of "streams" throughout its ecosystem, so thatmany higher-level consumers of this package only deal withstream usage.This implies that stream instances are most often created within somehigher-level components and many consumers never actually have to deal withcreating a stream instance.

• Usereact/socketif you want to accept incoming or establish outgoing plaintext TCP/IP orsecure TLS socket connection streams.
• Usereact/httpif you want to receive an incoming HTTP request body streams.
• Usereact/child-processif you want to communicate with child processes via process pipes such asSTDIN, STDOUT, STDERR etc.
• Use experimentalreact/filesystemif you want to read from / write to the filesystem.
• See also the last chapter formore real-world applications.

However, if you are writing a lower-level component or want to create a streaminstance from a stream resource, then the following chapter is for you.

Note that the following examples usefopen() andstream_socket_client()for illustration purposes only.These functions SHOULD NOT be used in a truly async program because each callmay take several seconds to complete and would block the EventLoop otherwise.Additionally, thefopen() call will return a file handle on some platformswhich may or may not be supported by all EventLoop implementations.As an alternative, you may want to use higher-level libraries listed above.

TheReadableResourceStream is a concrete implementation of theReadableStreamInterface for PHP's stream resources.

This can be used to represent a read-only resource like a file stream opened inreadable mode or a stream such asSTDIN:

$stream =newReadableResourceStream(STDIN);$stream->on('data',function (string$chunk):void {echo$chunk;});$stream->on('end',function ():void {echo'END';});

See alsoReadableStreamInterface for more details.

The first parameter given to the constructor MUST be a valid stream resourcethat is opened in reading mode (e.g.fopen() moder).Otherwise, it will throw anInvalidArgumentException:

// throws InvalidArgumentException$stream =newReadableResourceStream(false);

See also theDuplexResourceStream for read-and-writestream resources otherwise.

Internally, this class tries to enable non-blocking mode on the stream resourcewhich may not be supported for all stream resources.Most notably, this is not supported by pipes on Windows (STDIN etc.).If this fails, it will throw aRuntimeException:

// throws RuntimeException on Windows$stream =newReadableResourceStream(STDIN);

Once the constructor is called with a valid stream resource, this class willtake care of the underlying stream resource.You SHOULD only use its public API and SHOULD NOT interfere with the underlyingstream resource manually.

This class takes an optionalLoopInterface|null $loop parameter that can be used topass the event loop instance to use for this object. You can use anull valuehere in order to use thedefault loop.This value SHOULD NOT be given unless you're sure you want to explicitly use agiven event loop instance.

This class takes an optionalint|null $readChunkSize parameter that controlsthe maximum buffer size in bytes to read at once from the stream.You can use anull value here in order to apply its default value.This value SHOULD NOT be changed unless you know what you're doing.This can be a positive number which means that up to X bytes will be readat once from the underlying stream resource. Note that the actual numberof bytes read may be lower if the stream resource has less than X bytescurrently available.This can be-1 which means "read everything available" from theunderlying stream resource.This should read until the stream resource is not readable anymore(i.e. underlying buffer drained), note that this does not neccessarilymean it reached EOF.

$stream =newReadableResourceStream(STDIN,null,8192);

PHP bug warning: If the PHP process has explicitly been started without aSTDIN stream, then trying to read fromSTDIN may return data fromanother stream resource. This does not happen if you start this with an emptystream likephp test.php < /dev/null instead ofphp test.php <&-.See#81 for more details.

Changelog: As of v1.2.0 the$loop parameter can be omitted (or skipped with anull value) to use thedefault loop.

TheWritableResourceStream is a concrete implementation of theWritableStreamInterface for PHP's stream resources.

This can be used to represent a write-only resource like a file stream opened inwritable mode or a stream such asSTDOUT orSTDERR:

$stream =newWritableResourceStream(STDOUT);$stream->write('hello!');$stream->end();

See alsoWritableStreamInterface for more details.

The first parameter given to the constructor MUST be a valid stream resourcethat is opened for writing.Otherwise, it will throw anInvalidArgumentException:

// throws InvalidArgumentException$stream =newWritableResourceStream(false);

See also theDuplexResourceStream for read-and-writestream resources otherwise.

Internally, this class tries to enable non-blocking mode on the stream resourcewhich may not be supported for all stream resources.Most notably, this is not supported by pipes on Windows (STDOUT, STDERR etc.).If this fails, it will throw aRuntimeException:

// throws RuntimeException on Windows$stream =newWritableResourceStream(STDOUT);

Once the constructor is called with a valid stream resource, this class willtake care of the underlying stream resource.You SHOULD only use its public API and SHOULD NOT interfere with the underlyingstream resource manually.

Anywrite() calls to this class will not be performed instantly, but willbe performed asynchronously, once the EventLoop reports the stream resource isready to accept data.For this, it uses an in-memory buffer string to collect all outstanding writes.This buffer has a soft-limit applied which defines how much data it is willingto accept before the caller SHOULD stop sending further data.

This class takes an optionalLoopInterface|null $loop parameter that can be used topass the event loop instance to use for this object. You can use anull valuehere in order to use thedefault loop.This value SHOULD NOT be given unless you're sure you want to explicitly use agiven event loop instance.

This class takes an optionalint|null $writeBufferSoftLimit parameter that controlsthis maximum buffer size in bytes.You can use anull value here in order to apply its default value.This value SHOULD NOT be changed unless you know what you're doing.

$stream =newWritableResourceStream(STDOUT,null,8192);

This class takes an optionalint|null $writeChunkSize parameter that controlsthis maximum buffer size in bytes to write at once to the stream.You can use anull value here in order to apply its default value.This value SHOULD NOT be changed unless you know what you're doing.This can be a positive number which means that up to X bytes will be writtenat once to the underlying stream resource. Note that the actual numberof bytes written may be lower if the stream resource has less than X bytescurrently available.This can be-1 which means "write everything available" to theunderlying stream resource.

$stream =newWritableResourceStream(STDOUT,null,null,8192);

See alsowrite() for more details.

Changelog: As of v1.2.0 the$loop parameter can be omitted (or skipped with anull value) to use thedefault loop.

TheDuplexResourceStream is a concrete implementation of theDuplexStreamInterface for PHP's stream resources.

This can be used to represent a read-and-write resource like a file stream openedin read and write mode mode or a stream such as a TCP/IP connection:

$conn =stream_socket_client('tcp://google.com:80');$stream =newDuplexResourceStream($conn);$stream->write('hello!');$stream->end();

See alsoDuplexStreamInterface for more details.

The first parameter given to the constructor MUST be a valid stream resourcethat is opened for readingand writing.Otherwise, it will throw anInvalidArgumentException:

// throws InvalidArgumentException$stream =newDuplexResourceStream(false);

See also theReadableResourceStream for read-onlyand theWritableResourceStream for write-onlystream resources otherwise.

Internally, this class tries to enable non-blocking mode on the stream resourcewhich may not be supported for all stream resources.Most notably, this is not supported by pipes on Windows (STDOUT, STDERR etc.).If this fails, it will throw aRuntimeException:

// throws RuntimeException on Windows$stream =newDuplexResourceStream(STDOUT);

Once the constructor is called with a valid stream resource, this class willtake care of the underlying stream resource.You SHOULD only use its public API and SHOULD NOT interfere with the underlyingstream resource manually.

This class takes an optionalLoopInterface|null $loop parameter that can be used topass the event loop instance to use for this object. You can use anull valuehere in order to use thedefault loop.This value SHOULD NOT be given unless you're sure you want to explicitly use agiven event loop instance.

This class takes an optionalint|null $readChunkSize parameter that controlsthe maximum buffer size in bytes to read at once from the stream.You can use anull value here in order to apply its default value.This value SHOULD NOT be changed unless you know what you're doing.This can be a positive number which means that up to X bytes will be readat once from the underlying stream resource. Note that the actual numberof bytes read may be lower if the stream resource has less than X bytescurrently available.This can be-1 which means "read everything available" from theunderlying stream resource.This should read until the stream resource is not readable anymore(i.e. underlying buffer drained), note that this does not neccessarilymean it reached EOF.

$conn =stream_socket_client('tcp://google.com:80');$stream =newDuplexResourceStream($conn,null,8192);

Anywrite() calls to this class will not be performed instantly, but willbe performed asynchronously, once the EventLoop reports the stream resource isready to accept data.For this, it uses an in-memory buffer string to collect all outstanding writes.This buffer has a soft-limit applied which defines how much data it is willingto accept before the caller SHOULD stop sending further data.

This class takes another optionalWritableStreamInterface|null $buffer parameterthat controls this write behavior of this stream.You can use anull value here in order to apply its default value.This value SHOULD NOT be changed unless you know what you're doing.

If you want to change the write buffer soft limit, you can pass an instance ofWritableResourceStream like this:

$conn =stream_socket_client('tcp://google.com:80');$buffer =newWritableResourceStream($conn,null,8192);$stream =newDuplexResourceStream($conn,null,null,$buffer);

See alsoWritableResourceStream for more details.

Changelog: As of v1.2.0 the$loop parameter can be omitted (or skipped with anull value) to use thedefault loop.

TheThroughStream implements theDuplexStreamInterface and will simply pass any datayou write to it through to its readable end.

$through =newThroughStream();$through->on('data',$this->expectCallableOnceWith('hello'));$through->write('hello');

Similarly, theend() method will end the stream and emit anend event and thenclose() the stream.Theclose() method will close the stream and emit aclose event.Accordingly, this is can also be used in apipe() context like this:

$through =newThroughStream();$source->pipe($through)->pipe($dest);

Optionally, its constructor accepts any callable function which will then beused tofilter any data written to it. This function receives a single dataargument as passed to the writable side and must return the data as it will bepassed to its readable end:

$through =newThroughStream('strtoupper');$source->pipe($through)->pipe($dest);

Note that this class makes no assumptions about any data types. This can beused to convert data, for example for transforming any structured data intoa newline-delimited JSON (NDJSON) stream like this:

$through =newThroughStream(function (mixed$data):string {returnjson_encode($data) .PHP_EOL;});$through->on('data',$this->expectCallableOnceWith("[2, true]\n"));$through->write([2,true]);

The callback function is allowed to throw anException. In this case,the stream will emit anerror event and thenclose() the stream.

$through =newThroughStream(function (mixed$data):string {if (!is_string($data)) {thrownew \UnexpectedValueException('Only strings allowed');    }return$data;});$through->on('error',$this->expectCallableOnce()));$through->on('close',$this->expectCallableOnce()));$through->on('data',$this->expectCallableNever()));$through->write(2);

TheCompositeStream implements theDuplexStreamInterface and can be used to create asingle duplex stream from two individual streams implementingReadableStreamInterface andWritableStreamInterface respectively.

This is useful for some APIs which may require a singleDuplexStreamInterface or simply because it's oftenmore convenient to work with a single stream instance like this:

$stdin =newReadableResourceStream(STDIN);$stdout =newWritableResourceStream(STDOUT);$stdio =newCompositeStream($stdin,$stdout);$stdio->on('data',function (string$chunk)use ($stdio):void {$stdio->write('You said:' .$chunk);});

This is a well-behaving stream which forwards all stream events from theunderlying streams and forwards all streams calls to the underlying streams.

If youwrite() to the duplex stream, it will simplywrite() to thewritable side and return its status.

If youend() the duplex stream, it willend() the writable side and willpause() the readable side.

If youclose() the duplex stream, both input streams will be closed.If either of the two input streams emits aclose event, the duplex streamwill also close.If either of the two input streams is already closed while constructing theduplex stream, it willclose() the other side and return a closed stream.

The following example can be used to pipe the contents of a source file intoa destination file without having to ever read the whole file into memory:

$source =newReact\Stream\ReadableResourceStream(fopen('source.txt','r'));$dest =newReact\Stream\WritableResourceStream(fopen('destination.txt','w'));$source->pipe($dest);

Note that this example usesfopen() for illustration purposes only.This should not be used in a truly async program because the filesystem isinherently blocking and each call could potentially take several seconds.See alsocreating streams for more sophisticatedexamples.

The recommended way to install this library isthrough Composer.New to Composer?

Once released, this project will followSemVer.At the moment, this will install the latest development version:

composer require react/stream:^3@dev

See also theCHANGELOG for details about version upgrades.

This project aims to run on any platform and thus does not require any PHPextensions and supports running on PHP 7.1 through current PHP 8+.It'shighly recommended to use the latest supported PHP version for this project.

To run the test suite, you first need to clone this repo and then install alldependenciesthrough Composer:

composer install

To run the test suite, go to the project root and run:

vendor/bin/phpunit

The test suite also contains a number of functional integration tests that relyon a stable internet connection.If you do not want to run these, they can simply be skipped like this:

vendor/bin/phpunit --exclude-group internet

On top of this, we use PHPStan on max level to ensure type safety across the project:

vendor/bin/phpstan

MIT, seeLICENSE file.

• Seecreating streams for more information on how streamsare created in real-world applications.
• See ourusers wiki and thedependents on Packagistfor a list of packages that use streams in real-world applications.