text: <?php
  namespace GuzzleHttp\Stream;
  
  /**
   * Represents an asynchronous read-only stream that supports a drain event and
   * pumping data from a source stream.
   *
   * The AsyncReadStream can be used as a completely asynchronous stream, meaning
   * the data you can read from the stream will immediately return only
   * the data that is currently buffered.
   *
   * AsyncReadStream can also be used in a "blocking" manner if a "pump" function
   * is provided. When a caller requests more bytes than are available in the
   * buffer, then the pump function is used to block until the requested number
   * of bytes are available or the remote source stream has errored, closed, or
   * timed-out. This behavior isn't strictly "blocking" because the pump function
   * can send other transfers while waiting on the desired buffer size to be
   * ready for reading (e.g., continue to tick an event loop).
   *
   * @unstable This class is subject to change.
   */
  class AsyncReadStream implements StreamInterface
  {
      use StreamDecoratorTrait;
  
      /** @var callable|null Fn used to notify writers the buffer has drained */
      private $drain;
  
      /** @var callable|null Fn used to block for more data */
      private $pump;
  
      /** @var int|null Highwater mark of the underlying buffer */
      private $hwm;
  
      /** @var bool Whether or not drain needs to be called at some point */
      private $needsDrain;
  
      /** @var int The expected size of the remote source */
      private $size;
  
      /**
       * In order to utilize high water marks to tell writers to slow down, the
       * provided stream must answer to the "hwm" stream metadata variable,
       * providing the high water mark. If no "hwm" metadata value is available,
       * then the "drain" functionality is not utilized.
       *
       * This class accepts an associative array of configuration options.
       *
       * - drain: (callable) Function to invoke when the stream has drained,
       *   meaning the buffer is now writable again because the size of the
       *   buffer is at an acceptable level (e.g., below the high water mark).
       *   The function accepts a single argument, the buffer stream object that
       *   has drained.
       * - pump: (callable) A function that accepts the number of bytes to read
       *   from the source stream. This function will block until all of the data
       *   that was requested has been read, EOF of the source stream, or the
       *   source stream is closed.
       * - size: (int) The expected size in bytes of the data that will be read
       *   (if known up-front).
       *
       * @param StreamInterface $buffer   Buffer that contains the data that has
       *                                  been read by the event loop.
       * @param array           $config   Associative array of options.
       *
       * @throws \InvalidArgumentException if the buffer is not readable and
       *                                   writable.
       */
      public function __construct(
          StreamInterface $buffer,
          array $config = []
      ) {
          if (!$buffer->isReadable() || !$buffer->isWritable()) {
              throw new \InvalidArgumentException(
                  'Buffer must be readable and writable'
              );
          }
  
          if (isset($config['size'])) {
              $this->size = $config['size'];
          }
  
          static $callables = ['pump', 'drain'];
          foreach ($callables as $check) {
              if (isset($config[$check])) {
                  if (!is_callable($config[$check])) {
                      throw new \InvalidArgumentException(
                          $check . ' must be callable'
                      );
                  }
                  $this->{$check} = $config[$check];
              }
          }
  
          $this->hwm = $buffer->getMetadata('hwm');
  
          // Cannot drain when there's no high water mark.
          if ($this->hwm === null) {
              $this->drain = null;
          }
  
          $this->stream = $buffer;
      }
  
      /**
       * Factory method used to create new async stream and an underlying buffer
       * if no buffer is provided.
       *
       * This function accepts the same options as AsyncReadStream::__construct,
       * but added the following key value pairs:
       *
       * - buffer: (StreamInterface) Buffer used to buffer data. If none is
       *   provided, a default buffer is created.
       * - hwm: (int) High water mark to use if a buffer is created on your
       *   behalf.
       * - max_buffer: (int) If provided, wraps the utilized buffer in a
       *   DroppingStream decorator to ensure that buffer does not exceed a given
       *   length. When exceeded, the stream will begin dropping data. Set the
       *   max_buffer to 0, to use a NullStream which does not store data.
       * - write: (callable) A function that is invoked when data is written
       *   to the underlying buffer. The function accepts the buffer as the first
       *   argument, and the data being written as the second. The function MUST
       *   return the number of bytes that were written or false to let writers
       *   know to slow down.
       * - drain: (callable) See constructor documentation.
       * - pump: (callable) See constructor documentation.
       *
       * @param array $options Associative array of options.
       *
       * @return array Returns an array containing the buffer used to buffer
       *               data, followed by the ready to use AsyncReadStream object.
       */
      public static function create(array $options = [])
      {
          $maxBuffer = isset($options['max_buffer'])
              ? $options['max_buffer']
              : null;
  
          if ($maxBuffer === 0) {
              $buffer = new NullStream();
          } elseif (isset($options['buffer'])) {
              $buffer = $options['buffer'];
          } else {
              $hwm = isset($options['hwm']) ? $options['hwm'] : 16384;
              $buffer = new BufferStream($hwm);
          }
  
          if ($maxBuffer > 0) {
              $buffer = new DroppingStream($buffer, $options['max_buffer']);
          }
  
          // Call the on_write callback if an on_write function was provided.
          if (isset($options['write'])) {
              $onWrite = $options['write'];
              $buffer = FnStream::decorate($buffer, [
                  'write' => function ($string) use ($buffer, $onWrite) {
                      $result = $buffer->write($string);
                      $onWrite($buffer, $string);
                      return $result;
                  }
              ]);
          }
  
          return [$buffer, new self($buffer, $options)];
      }
  
      public function getSize()
      {
          return $this->size;
      }
  
      public function isWritable()
      {
          return false;
      }
  
      public function write($string)
      {
          return false;
      }
  
      public function read($length)
      {
          if (!$this->needsDrain && $this->drain) {
              $this->needsDrain = $this->stream->getSize() >= $this->hwm;
          }
  
          $result = $this->stream->read($length);
  
          // If we need to drain, then drain when the buffer is empty.
          if ($this->needsDrain && $this->stream->getSize() === 0) {
              $this->needsDrain = false;
              $drainFn = $this->drain;
              $drainFn($this->stream);
          }
  
          $resultLen = strlen($result);
  
          // If a pump was provided, the buffer is still open, and not enough
          // data was given, then block until the data is provided.
          if ($this->pump && $resultLen < $length) {
              $pumpFn = $this->pump;
              $result .= $pumpFn($length - $resultLen);
          }
  
          return $result;
      }
  }
  