public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         if (!empty($this->_previous)) {
             $bucket->data = $this->_previous . $bucket->data;
             $this->_previous = '';
         }
         if (!feof($this->stream) && preg_match('/(%[12]\\$|%[12]|%)$/', $bucket->data)) {
             $this->_previous .= $bucket->data;
             return PSFS_FEED_ME;
         }
         $consumed += $bucket->datalen;
         if (preg_match('/%([12])\\$s/', $bucket->data, $matches)) {
             if ($matches[1] == '1') {
                 $bucket->data = preg_replace('/%1\\$s/', $this->_sender, $bucket->data);
             } else {
                 $bucket->data = preg_replace('/%2\\$s/', $this->_recipient, $bucket->data);
             }
         }
         $bucket->datalen = strlen($bucket->data);
         stream_bucket_append($out, $bucket);
     }
     if (!empty($this->_previous)) {
         if ($closing) {
             $bucket = stream_bucket_new($this->stream, $this->_previous);
             $bucket->data = $this->_previous;
             $consumed += strlen($this->_previous);
             $this->_previous = '';
             stream_bucket_append($out, $bucket);
         }
     }
     return PSFS_PASS_ON;
 }

 function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $this->data .= $bucket->data;
         $this->bucket = $bucket;
         $consumed = 0;
     }
     if ($closing) {
         $consumed += strlen($this->data);
         if (isset($this->bucket)) {
             $bucket = $this->bucket;
         } else {
             $bucket = stream_bucket_new($this->stream, '');
         }
         $f = self::$registry[get_class($this)];
         AbstractStreamProcessor__lazyUriResolver::bind($this->stream, $f->uri);
         $bucket->data = $f->process($this->data);
         $bucket->datalen = strlen($bucket->data);
         stream_bucket_append($out, $bucket);
         $this->bucket = null;
         $this->data = '';
         return PSFS_PASS_ON;
     }
     return PSFS_FEED_ME;
 }

 /**
  * Filter some data.
  *
  * @param resource $in       Read from this input stream
  * @param resource $out      Write to this output stream
  * @param int      $consumed Count of bytes processed
  * @param bool     $closing  Is the input about to end?
  *
  * @return int PSFS_PASS_ON / PSFS_FEED_ME / PSFS_ERR_FATAL
  */
 public function filter($in, $out, &$consumed, $closing) : int
 {
     $return = PSFS_FEED_ME;
     // While input data is available, continue to read it.
     while ($bucket_in = stream_bucket_make_writeable($in)) {
         $this->data .= $bucket_in->data;
         $consumed += $bucket_in->datalen;
         // While we have complete GEDCOM records, process them.
         while (preg_match('/(.*?[\\r\\n]\\s*)(0.*)/s', $this->data, $match) === 1) {
             list(, $data, $this->data) = $match;
             // Send this record output.
             $data = $this->filterData($data);
             $bucket_out = stream_bucket_new($this->stream, $data);
             $return = PSFS_PASS_ON;
             stream_bucket_append($out, $bucket_out);
         }
     }
     // Process the final record.
     if ($closing && $this->data !== '') {
         $data = $this->filterData($this->data);
         $bucket_out = stream_bucket_new($this->stream, $data);
         $return = PSFS_PASS_ON;
         stream_bucket_append($out, $bucket_out);
     }
     return $return;
 }

 function filter($in, $out, &$consumed, $closing)
 {
     if (!$closing) {
         $bucket = stream_bucket_new($this->stream, "42");
     }
     return PSFS_ERR_FATAL;
 }

 public function filter($in, $out, &$consumed, $closing) : int
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         stream_bucket_append($out, stream_bucket_new($this->stream, str_rot13($bucket->data)));
     }
     return \PSFS_PASS_ON;
 }

 function filter($in, $out, &$consumed, $closing)
 {
     while ($ib = stream_bucket_make_writeable($in)) {
         $ob = stream_bucket_new($this->stream, strtoupper($ib->data));
         stream_bucket_append($out, $ob);
     }
     return PSFS_PASS_ON;
 }

 /**
  * @see stream_filter_register()
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     stream_bucket_append($out, stream_bucket_new($this->stream, '"'));
     while ($bucket = stream_bucket_make_writeable($in)) {
         $consumed += $bucket->datalen;
         $bucket->data = addcslashes($bucket->data, '"\\');
         stream_bucket_append($out, $bucket);
     }
     stream_bucket_append($out, stream_bucket_new($this->stream, '"'));
     return PSFS_PASS_ON;
 }

 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $this->buffer .= $bucket->data;
         $consumed += $bucket->datalen;
     }
     if ($closing) {
         $bucket = stream_bucket_new($this->stream, 'Closed: ' . $this->buffer);
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $lenbucket = stream_bucket_new($this->stream, dechex($bucket->datalen) . "\r\n");
         stream_bucket_append($out, $lenbucket);
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
         $lenbucket = stream_bucket_new($this->stream, "\r\n");
         stream_bucket_append($out, $lenbucket);
     }
     return PSFS_PASS_ON;
 }

 public function filter($in, $out, &$consumed, $closing)
 {
     // concatenate whole buffer from input brigade
     $data = '';
     while ($bucket = stream_bucket_make_writeable($in)) {
         $consumed += $bucket->datalen;
         $data .= $bucket->data;
     }
     // skip processing callback that already ended
     if ($this->closed) {
         return PSFS_FEED_ME;
     }
     // only invoke filter function if buffer is not empty
     // this may skip flushing a closing filter
     if ($data !== '') {
         try {
             $data = call_user_func($this->callback, $data);
         } catch (Exception $e) {
             // exception should mark filter as closed
             $this->onClose();
             trigger_error('Error invoking filter: ' . $e->getMessage(), E_USER_WARNING);
             return PSFS_ERR_FATAL;
         }
     }
     // mark filter as closed after processing closing chunk
     if ($closing) {
         $this->closed = true;
         // callback supports closing and is not already closed
         if ($this->supportsClose) {
             $this->supportsClose = false;
             // invoke without argument to signal end and append resulting buffer
             try {
                 $data .= call_user_func($this->callback);
             } catch (Exception $e) {
                 trigger_error('Error ending filter: ' . $e->getMessage(), E_USER_WARNING);
                 return PSFS_ERR_FATAL;
             }
         }
     }
     if ($data !== '') {
         // create a new bucket for writing the resulting buffer to the output brigade
         // reusing an existing bucket turned out to be bugged in some environments (ancient PHP versions and HHVM)
         $bucket = @stream_bucket_new($this->stream, $data);
         // legacy PHP versions (PHP < 5.4) do not support passing data from the event signal handler
         // because closing the stream invalidates the stream and its stream bucket brigade before
         // invoking the filter close handler.
         if ($bucket !== false) {
             stream_bucket_append($out, $bucket);
         }
     }
     return PSFS_PASS_ON;
 }

 /**
  * Filter implementation converts encoding before returning PSFS_PASS_ON.
  * 
  * @param resource $in
  * @param resource $out
  * @param int $consumed
  * @param bool $closing
  * @return int
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $converted = $this->converter->convert($bucket->data);
         $consumed += strlen($bucket->data);
         // $this->stream is undocumented.  It was found looking at HHVM's source code
         // for its convert.iconv.* implementation in ConvertIconFilter and explained
         // somewhat in this StackOverflow page: http://stackoverflow.com/a/31132646/335059
         // declaring a member variable called 'stream' breaks the PHP implementation (5.5.9
         // at least).
         stream_bucket_append($out, stream_bucket_new($this->stream, $converted));
     }
     return PSFS_PASS_ON;
 }

 public function filter($in, $out, &$consumed, $closing)
 {
     static $closed = 0;
     while ($bucket = stream_bucket_make_writeable($in)) {
         stream_bucket_append($out, stream_bucket_new($this->stream, $bucket->data));
     }
     if ($closing) {
         $closed++;
     }
     if ($closed > 0) {
         var_dump($closed++);
     }
     return PSFS_PASS_ON;
 }

 /**
  * {@inheritdoc}
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     while ($streamBucket = stream_bucket_make_writeable($in)) {
         $this->sourceCode .= $streamBucket->data;
     }
     if ($closing || feof($this->stream)) {
         $consumed = strlen($this->sourceCode);
         $processedCode = Engine::parse($this->sourceCode);
         $streamBucket = stream_bucket_new($this->stream, $processedCode);
         stream_bucket_append($out, $streamBucket);
         return PSFS_PASS_ON;
     }
     return PSFS_FEED_ME;
 }

 /**
  * @param resource $in
  * @param resource $out
  * @param int $consumed
  * @param bool $closing
  *
  * @return int
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     if ($closing) {
         /** @var resource|\stdClass $bucket */
         $bucket = stream_bucket_new($this->stream, $this->transform($this->data));
         $this->data = '';
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
         return PSFS_PASS_ON;
     } else {
         while ($bucket = stream_bucket_make_writeable($in)) {
             $this->data .= $bucket->data;
         }
         return PSFS_FEED_ME;
     }
 }

 function filter($in, $out, &$consumed, $closing)
 {
     $output = 0;
     while ($bucket = stream_bucket_make_writeable($in)) {
         $bucket->data = strtoupper($bucket->data);
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
         $output = 1;
     }
     if ($closing) {
         $bucket = stream_bucket_new($this->stream, "\n===close===\n");
         stream_bucket_append($out, $bucket);
         $output = 1;
     }
     return $output ? PSFS_PASS_ON : PSFS_FEED_ME;
 }

 /**
  * @see stream_filter_register()
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     if (!$this->_prepend) {
         stream_bucket_append($out, stream_bucket_new($this->stream, '"'));
         $this->_prepend = true;
     }
     while ($bucket = stream_bucket_make_writeable($in)) {
         $consumed += $bucket->datalen;
         $bucket->data = addcslashes($bucket->data, '"\\');
         stream_bucket_append($out, $bucket);
     }
     /* feof() call needed due to:
      * http://news.php.net/php.internals/80363 */
     if ($closing || feof($this->stream)) {
         stream_bucket_append($out, stream_bucket_new($this->stream, '"'));
     }
     return PSFS_PASS_ON;
 }

 public function filter($in, $out, &$consumed, $closing)
 {
     $type = $this->type;
     $ret = PSFS_FEED_ME;
     while ($bucket = stream_bucket_make_writeable($in)) {
         $consumed += $bucket->datalen;
         $this->buffer .= $bucket->data;
     }
     if ($type == 'enrypt') {
         $blockSize = $this->crypter->getEncryptBlockSize();
     } else {
         $blockSize = $this->crypter->getDecryptBlockSize();
     }
     if ($blockSize < 1) {
         // stream cipher or special cases, encrypt every bucket
         $blockSize = 1;
     }
     $len = strlen($this->buffer);
     // process filled blocks
     if ($blockSize > 0 and $len >= $blockSize) {
         $mod = $len % $blockSize;
         $data = $this->crypter->{$type}(substr($this->buffer, 0, $len - $mod));
         $this->buffer = substr($this->buffer, $len - $mod);
         $bucket = stream_bucket_new($this->temp, $data);
         if ($bucket === false) {
             return PSFS_ERR_FATAL;
         }
         stream_bucket_append($out, $bucket);
         $ret = PSFS_PASS_ON;
     }
     // process rest data if we are closing
     if ($closing and strlen($this->buffer) > 0) {
         $bucket = stream_bucket_new($this->temp, $this->crypter->{$type}($this->buffer));
         if ($bucket === false) {
             return PSFS_ERR_FATAL;
         }
         $this->buffer = '';
         stream_bucket_append($out, $bucket);
         $ret = PSFS_PASS_ON;
     }
     return $ret;
 }

 /**
  * Filter the input data through the transform.
  *
  * @param resource $input     The input bucket brigade.
  * @param resource $output    The output bucket brigade.
  * @param integer  &$consumed The number of bytes consumed.
  * @param boolean  $isEnd     True if the stream is closing.
  *
  * @return integer The result code.
  */
 public function filter($input, $output, &$consumed, $isEnd)
 {
     $bucket = stream_bucket_make_writeable($input);
     if ($isEnd && !$bucket) {
         $bucket = stream_bucket_new(STDIN, '');
     }
     $hasOutput = false;
     while ($bucket) {
         $this->buffer .= $bucket->data;
         $bufferSize = strlen($this->buffer);
         if (!$isEnd && $bufferSize < $this->bufferSize) {
             $bucket = stream_bucket_make_writeable($input);
             continue;
         }
         list($outputBuffer, $thisConsumed, $error) = $this->transform->transform($this->buffer, $this->context, $isEnd);
         $consumed += $thisConsumed;
         if ($bufferSize === $thisConsumed) {
             $this->buffer = '';
         } else {
             $this->buffer = substr($this->buffer, $thisConsumed);
         }
         if ('' !== $outputBuffer) {
             $bucket->data = $outputBuffer;
             stream_bucket_append($output, $bucket);
             $hasOutput = true;
         }
         if (null !== $error) {
             return PSFS_ERR_FATAL;
         }
         $bucket = stream_bucket_make_writeable($input);
     }
     if ($hasOutput || $isEnd) {
         return PSFS_PASS_ON;
     }
     return PSFS_FEED_ME;
 }

 /**
  * Create a new bucket for use on the current stream
  *
  * @param resource $stream
  * @param string   $buffer
  *
  * @return object
  */
 public function bucketNew($stream, string $buffer)
 {
     return stream_bucket_new($stream, $buffer);
 }

 /**
  * {@inheritdoc}
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $this->data .= $bucket->data;
     }
     if ($closing || feof($this->stream)) {
         $consumed = strlen($this->data);
         // $this->stream contains pointer to the source
         $metadata = new StreamMetaData($this->stream, $this->data);
         $this->transformCode($metadata);
         $bucket = stream_bucket_new($this->stream, $metadata->source);
         stream_bucket_append($out, $bucket);
         return PSFS_PASS_ON;
     }
     return PSFS_FEED_ME;
 }