Subversion Repositories php-qbpwcf

<?php

namespace React\Stream;

use Evenement\EventEmitter;

use InvalidArgumentException;

/**

 * The `ThroughStream` implements the

 * [`DuplexStreamInterface`](#duplexstreaminterface) and will simply pass any data

 * you write to it through to its readable end.

 *

 * ```php

 * $through = new ThroughStream();

 * $through->on('data', $this->expectCallableOnceWith('hello'));

 *

 * $through->write('hello');

 * ```

 *

 * Similarly, the [`end()` method](#end) will end the stream and emit an

 * [`end` event](#end-event) and then [`close()`](#close-1) the stream.

 * The [`close()` method](#close-1) will close the stream and emit a

 * [`close` event](#close-event).

 * Accordingly, this is can also be used in a [`pipe()`](#pipe) context like this:

 *

 * ```php

 * $through = new ThroughStream();

 * $source->pipe($through)->pipe($dest);

 * ```

 *

 * Optionally, its constructor accepts any callable function which will then be

 * used to *filter* any data written to it. This function receives a single data

 * argument as passed to the writable side and must return the data as it will be

 * passed to its readable end:

 *

 * ```php

 * $through = new ThroughStream('strtoupper');

 * $source->pipe($through)->pipe($dest);

 * ```

 *

 * Note that this class makes no assumptions about any data types. This can be

 * used to convert data, for example for transforming any structured data into

 * a newline-delimited JSON (NDJSON) stream like this:

 *

 * ```php

 * $through = new ThroughStream(function ($data) {

 *     return json_encode($data) . PHP_EOL;

 * });

 * $through->on('data', $this->expectCallableOnceWith("[2, true]\n"));

 *

 * $through->write(array(2, true));

 * ```

 *

 * The callback function is allowed to throw an `Exception`. In this case,

 * the stream will emit an `error` event and then [`close()`](#close-1) the stream.

 *

 * ```php

 * $through = new ThroughStream(function ($data) {

 *     if (!is_string($data)) {

 *         throw new \UnexpectedValueException('Only strings allowed');

 *     }

 *     return $data;

 * });

 * $through->on('error', $this->expectCallableOnce()));

 * $through->on('close', $this->expectCallableOnce()));

 * $through->on('data', $this->expectCallableNever()));

 *

 * $through->write(2);

 * ```

 *

 * @see WritableStreamInterface::write()

 * @see WritableStreamInterface::end()

 * @see DuplexStreamInterface::close()

 * @see WritableStreamInterface::pipe()

 */

final class ThroughStream extends EventEmitter implements DuplexStreamInterface

{

    private $readable = true;

    private $writable = true;

    private $closed = false;

    private $paused = false;

    private $drain = false;

    private $callback;

    public function __construct($callback = null)

    {

        if ($callback !== null && !\is_callable($callback)) {

            throw new InvalidArgumentException('Invalid transformation callback given');

        }

        $this->callback = $callback;

    }

    public function pause()

    {

        $this->paused = true;

    }

    public function resume()

    {

        if ($this->drain) {

            $this->drain = false;

            $this->emit('drain');

        }

        $this->paused = false;

    }

    public function pipe(WritableStreamInterface $dest, array $options = array())

    {

        return Util::pipe($this, $dest, $options);

    }

    public function isReadable()

    {

        return $this->readable;

    }

    public function isWritable()

    {

        return $this->writable;

    }

    public function write($data)

    {

        if (!$this->writable) {

            return false;

        }

        if ($this->callback !== null) {

            try {

                $data = \call_user_func($this->callback, $data);

            } catch (\Exception $e) {

                $this->emit('error', array($e));

                $this->close();

                return false;

            }

        }

        $this->emit('data', array($data));

        if ($this->paused) {

            $this->drain = true;

            return false;

        }

        return true;

    }

    public function end($data = null)

    {

        if (!$this->writable) {

            return;

        }

        if (null !== $data) {

            $this->write($data);

            // return if write() already caused the stream to close

            if (!$this->writable) {

                return;

            }

        }

        $this->readable = false;

        $this->writable = false;

        $this->paused = true;

        $this->drain = false;

        $this->emit('end');

        $this->close();

    }

    public function close()

    {

        if ($this->closed) {

            return;

        }

        $this->readable = false;

        $this->writable = false;

        $this->closed = true;

        $this->paused = true;

        $this->drain = false;

        $this->callback = null;

        $this->emit('close');

        $this->removeAllListeners();

    }

}