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;
 }

 /**
  * 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;
 }

 /**
  * Traffic shaping.
  *
  * @param resource $in       The input stream.
  * @param resource $out      The ouput stream.
  * @param int      $consumed The amount of consumed bytes.
  * @param bool     $closing  If the stream is closing.
  *
  * @return int The processing state.
  *
  * @SuppressWarnings("unused")
  * @SuppressWarnings("short")
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     try {
         while ($bucket = stream_bucket_make_writeable($in)) {
             $chunks = str_split($bucket->data, $this->tokenBucket->getCapacity());
             foreach ($chunks as $chunk) {
                 $tokens = strlen($chunk);
                 $this->tokenConsumer->consume($tokens);
                 $consumed += $tokens;
             }
             stream_bucket_append($out, $bucket);
         }
         return PSFS_PASS_ON;
     } catch (StorageException $e) {
         trigger_error($e->getMessage(), E_USER_ERROR);
         return PSFS_ERR_FATAL;
     } catch (\LengthException $e) {
         /*
          * This case would be a logic error, as the stream chunk is already
          * splitted to the bucket's capacity.
          */
         trigger_error($e->getMessage(), E_USER_ERROR);
         return PSFS_ERR_FATAL;
     }
 }

 /**
  * The main filter method.
  * Implemented according to \php_user_filter class. Will loop over all stream buckets, buffer them and perform
  * the needed actions.
  *
  * @param resource $in       Incoming bucket brigade we need to filter
  * @param resource $out      Outgoing bucket brigade with already filtered content
  * @param integer  $consumed The count of altered characters as buckets pass the filter
  * @param boolean  $closing  Is the stream about to close?
  *
  * @throws \TechDivision\PBC\Exceptions\GeneratorException
  *
  * @return integer
  *
  * @link http://www.php.net/manual/en/php-user-filter.filter.php
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     // Get our buckets from the stream
     while ($bucket = stream_bucket_make_writeable($in)) {
         // Get the tokens
         $tokens = token_get_all($bucket->data);
         // Go through the tokens and check what we found
         $tokensCount = count($tokens);
         for ($i = 0; $i < $tokensCount; $i++) {
             // Did we find a function? If so check if we know that thing and insert the code of its preconditions.
             if (is_array($tokens[$i]) && $tokens[$i][0] === T_FUNCTION && is_array($tokens[$i + 2])) {
                 // Get the name of the function
                 $functionName = $tokens[$i + 2][1];
                 // Check if we got the function in our list, if not continue
                 $functionDefinition = $this->params->get($functionName);
                 if (!$functionDefinition instanceof FunctionDefinition) {
                     continue;
                 } else {
                     // Get the code for the assertions
                     $code = $this->generateCode($functionDefinition->getAllPreconditions());
                     // Insert the code
                     $bucket->data = str_replace(PBC_PRECONDITION_PLACEHOLDER . $functionDefinition->getName() . PBC_PLACEHOLDER_CLOSE, $code, $bucket->data);
                     // "Destroy" code and function definition
                     $code = null;
                     $functionDefinition = null;
                 }
             }
         }
         // Tell them how much we already processed, and stuff it back into the output
         $consumed += $bucket->datalen;
         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 Stream Through Buckets
  *
  * @param resource $in userfilter.bucket brigade
  *                         pointer to a group of buckets objects containing the data to be filtered
  * @param resource $out userfilter.bucket brigade
  *                         pointer to another group of buckets for storing the converted data
  * @param int $consumed counter passed by reference that must be incremented by the length
  *                         of converted data
  * @param boolean $closing flag that is set to TRUE if we are in the last cycle and the stream is
  *                           about to close
  * @return int
  */
 function filter($in, $out, &$consumed, $closing)
 {
     // $in and $out are opaque "bucket brigade" objects which consist of a
     // sequence of opaque "buckets", which contain the actual stream data.
     // The only way to use these objects is the stream_bucket_* functions.
     // Unfortunately, there doesn't seem to be any way to access a bucket
     // without turning it into a string using stream_bucket_make_writeable(),
     // even if you want to pass the bucket along unmodified.
     // Each call to this pops a bucket from the bucket brigade and
     // converts it into an object with two properties: datalen and data.
     // This same object interface is accepted by stream_bucket_append().
     while ($bucket = stream_bucket_make_writeable($in)) {
         $outbuffer = '';
         $offset = 0;
         // Loop through the string.  For efficiency, we don't advance a character
         // at a time but try to zoom ahead to where we think the next chunk
         // boundary should be.
         // Since the stream filter divides the data into buckets arbitrarily,
         // we have to maintain state ($this->chunkremaining) across filter() calls.
         while ($offset < $bucket->datalen) {
             if ($this->chunkremaining === 0) {
                 // start of new chunk, or the start of the transfer
                 $firstline = strpos($bucket->data, "\r\n", $offset);
                 $chunkline = substr($bucket->data, $offset, $firstline - $offset);
                 $chunklen = current(explode(';', $chunkline, 2));
                 // ignore MIME-like extensions
                 $chunklen = trim($chunklen);
                 if (!ctype_xdigit($chunklen)) {
                     // There should have been a chunk length specifier here, but since
                     // there are non-hex digits something must have gone wrong.
                     return PSFS_ERR_FATAL;
                 }
                 $this->chunkremaining = hexdec($chunklen);
                 // $firstline already includes $offset in it
                 $offset = $firstline + 2;
                 // +2 is CRLF
                 if ($this->chunkremaining === 0) {
                     //end of the transfer
                     break;
                 }
                 // ignore possible trailing headers
             }
             // get as much data as available in a single go...
             $nibble = substr($bucket->data, $offset, $this->chunkremaining);
             $nibblesize = strlen($nibble);
             $offset += $nibblesize;
             // ...but recognize we may not have got all of it
             if ($nibblesize === $this->chunkremaining) {
                 $offset += 2;
             }
             // skip over trailing CRLF
             $this->chunkremaining -= $nibblesize;
             $outbuffer .= $nibble;
         }
         $consumed += $bucket->datalen;
         $bucket->data = $outbuffer;
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 /**
  * The main filter method.
  * Implemented according to \php_user_filter class. Will loop over all stream buckets, buffer them and perform
  * the needed actions.
  *
  * @param resource $in       Incoming bucket brigade we need to filter
  * @param resource $out      Outgoing bucket brigade with already filtered content
  * @param integer  $consumed The count of altered characters as buckets pass the filter
  * @param boolean  $closing  Is the stream about to close?
  *
  * @throws \Exception
  * @throws \PHPParser_Error
  *
  * @return integer
  *
  * @link http://www.php.net/manual/en/php-user-filter.filter.php
  *
  * TODO The buffering does not work that well, maybe we should implement universal buffering within parent class!
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     // Get our buckets from the stream
     $buffer = '';
     while ($bucket = stream_bucket_make_writeable($in)) {
         $buffer .= $bucket->data;
         // Tell them how much we already processed, and stuff it back into the output
         $consumed += $bucket->datalen;
         // Save a bucket for later reuse
         $bigBucket = $bucket;
     }
     // Beautify all the buckets!
     $parser = new \PHPParser_Parser(new \PHPParser_Lexer());
     $prettyPrinter = new \PHPParser_PrettyPrinter_Default();
     try {
         // parse
         $stmts = $parser->parse($buffer);
         $data = '<?php ' . $prettyPrinter->prettyPrint($stmts);
     } catch (PHPParser_Error $e) {
         throw $e;
     }
     // Refill the bucket with the beautified data
     // Do not forget to set the length!
     $bigBucket->data = $data;
     $bigBucket->datalen = strlen($data);
     // Only append our big bucket
     stream_bucket_append($out, $bigBucket);
     return PSFS_PASS_ON;
 }

 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;
 }

 /**
  * The main filter method.
  * Implemented according to \php_user_filter class. Will loop over all stream buckets, buffer them and perform
  * the needed actions.
  *
  * @param resource $in       Incoming bucket brigade we need to filter
  * @param resource $out      Outgoing bucket brigade with already filtered content
  * @param integer  $consumed The count of altered characters as buckets pass the filter
  * @param boolean  $closing  Is the stream about to close?
  *
  * @throws \AppserverIo\Doppelgaenger\Exceptions\GeneratorException
  *
  * @return integer
  *
  * @link http://www.php.net/manual/en/php-user-filter.filter.php
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     // get all the introductions of a structure definition
     $introductions = $this->params;
     // Get our buckets from the stream
     $interfaceHook = '';
     $keywordNeeded = true;
     while ($bucket = stream_bucket_make_writeable($in)) {
         // Has to be done only once at the beginning of the definition
         if (empty($interfaceHook) && $introductions->count() > 0) {
             // Get the tokens
             $tokens = token_get_all($bucket->data);
             // Go through the tokens and check what we found
             $tokensCount = count($tokens);
             for ($i = 0; $i < $tokensCount; $i++) {
                 // We need something to hook into, right after class header seems fine
                 if (is_array($tokens[$i]) && $tokens[$i][0] === T_CLASS && $tokens[$i - 1][0] !== T_PAAMAYIM_NEKUDOTAYIM) {
                     for ($j = $i; $j < $tokensCount; $j++) {
                         // If we got the opening bracket we can break
                         if ($tokens[$j] === '{' || $tokens[$j][0] === T_CURLY_OPEN) {
                             break;
                         }
                         if (is_array($tokens[$j])) {
                             // we have to check if there already are interfaces
                             if ($tokens[$j][0] === T_IMPLEMENTS) {
                                 $keywordNeeded = false;
                             }
                             $interfaceHook .= $tokens[$j][1];
                         } else {
                             $interfaceHook .= $tokens[$j];
                         }
                     }
                     // build up the injected code and make the injection
                     if ($keywordNeeded) {
                         $implementsCode = ' implements ';
                     } else {
                         $implementsCode = ', ';
                     }
                     $useCode = '';
                     $interfaces = array();
                     foreach ($introductions as $introduction) {
                         $interfaces[] = $introduction->getInterface();
                         // build up code for the trait usage
                         $useCode .= 'use ' . $introduction->getImplementation() . ';
                         ';
                     }
                     $implementsCode .= implode(', ', $interfaces);
                     // add the "use" code
                     $bucket->data = str_replace($interfaceHook . '{', $interfaceHook . '{' . $useCode, $bucket->data);
                     // add the "implements" code
                     $bucket->data = str_replace($interfaceHook, $interfaceHook . $implementsCode, $bucket->data);
                 }
             }
         }
         // Tell them how much we already processed, and stuff it back into the output
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
     }
     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;
 }

 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $bucket->data = preg_replace(self::$pattern, '', $bucket->data);
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $consumed += $bucket->datalen;
         $bucket->data = call_user_func($this->params["func"], $this, $bucket->data, $closing);
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $bucket->data = preg_replace("/[\r\n]+\$/", "\r\n", $bucket->data);
         $consumed += $bucket->datalen;
         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)) {
         $bucket->data = mb_convert_encoding($bucket->data, 'UTF-8', $this->charset);
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 public function filter($in, $out, &$consumed, $closing)
 {
     while ($res = stream_bucket_make_writeable($in)) {
         $res->data = @mb_convert_encoding($res->data, $this->encoding_to, $this->encoding_from);
         $consumed += $res->datalen;
         stream_bucket_append($out, $res);
     }
     return PSFS_PASS_ON;
 }

function filter_impl($in, $out)
{
    while ($bucket = stream_bucket_make_writeable($in)) {
        $bucket->data = strtoupper($bucket->data) . $bucket->data;
        $consumed += $bucket->datalen;
        $bucket->datalen *= 2;
        stream_bucket_append($out, $bucket);
    }
}

 /**
  * @see stream_filter_register()
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $consumed += $bucket->datalen;
         $this->params->crc32 = $this->_crc32Combine($this->params->crc32, crc32($bucket->data), $bucket->datalen);
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 /**
  * Collect intercepted data
  *
  * @param resource $in
  * @param resource $out
  * @param int $consumed
  * @param bool $closing
  * @return int
  *
  * @SuppressWarnings(PHPMD.UnusedFormalParameter)
  * @SuppressWarnings(PHPMD.ShortVariable)
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         self::$_collectedData .= $bucket->data;
         $consumed += $bucket->datalen;
         stream_bucket_append($out, $bucket);
     }
     return PSFS_PASS_ON;
 }

 /**
  * Called when applying the filter
  *
  * @param resource $in  Resource pointing to a bucket brigade which contains one or more bucket objects containing
  *                      data to be filtered
  * @param resource $out Resource pointing to a second bucket brigade into which your modified buckets should be
  *                      placed.
  * @param integer $consumed Consumed, which must always be declared by reference, should be incremented by the length
  *                          of the data which your filter reads in and alters. In most cases this means you will
  *                          increment consumed by $bucket->datalen for each $bucket.
  * @param bool $closing If the stream is in the process of closing (and therefore this is the last pass through the
  *                      filterchain), the closing parameter will be set to TRUE.
  * @return int
  */
 public function filter($in, $out, &$consumed, $closing)
 {
     while ($bucket = stream_bucket_make_writeable($in)) {
         $bucket->data = trim(preg_replace('/>\\s+</', '><', $bucket->data));
         $consumed += $bucket->datalen;
         stream_bucket_prepend($out, $bucket);
     }
     return PSFS_PASS_ON;
 }