original_string: public function consume(AbstractConsumer $consumer, $auto_ack = false)
      {
          if (!$consumer->active()) {
              // prevent dirty consumer been listening on queue
              return;
          }
  
          $outside_error = null;
  
          try {
              $consumer->begin($this);
  
              $this->queue->consume(
                          function (AMQPEnvelope $envelope /*, AMQPQueue $queue*/) use ($consumer) {
                              $delivery = $this->builder->build($envelope);
                              $this->feed($delivery, $consumer);
  
                              return $consumer->active();
                          }, $auto_ack ? AMQP_AUTOACK : AMQP_NOPARAM
              );
  
          } catch (Exception $e) {
              $outside_error = $e;
          }
  
          try {
              $this->queue->cancel();
          } catch (Exception $e) {
          }
  
          $consumer->end($this, $outside_error);
  
          if ($outside_error) {
              throw $outside_error;
          }
      }
code: public function consume(AbstractConsumer $consumer, $auto_ack = false)
      {
          if (!$consumer->active()) {
              // prevent dirty consumer been listening on queue
              return;
          }
  
          $outside_error = null;
  
          try {
              $consumer->begin($this);
  
              $this->queue->consume(
                          function (AMQPEnvelope $envelope /*, AMQPQueue $queue*/) use ($consumer) {
                              $delivery = $this->builder->build($envelope);
                              $this->feed($delivery, $consumer);
  
                              return $consumer->active();
                          }, $auto_ack ? AMQP_AUTOACK : AMQP_NOPARAM
              );
  
          } catch (Exception $e) {
              $outside_error = $e;
          }
  
          try {
              $this->queue->cancel();
          } catch (Exception $e) {
          }
  
          $consumer->end($this, $outside_error);
  
          if ($outside_error) {
              throw $outside_error;
          }
      }
code_tokens: public function consume ( AbstractConsumer $ consumer , $ auto_ack = false ) { if ( ! $ consumer -> active ( ) ) { // prevent dirty consumer been listening on queue return ; } $ outside_error = null ; try { $ consumer -> begin ( $ this ) ; $ this -> queue -> consume ( function ( AMQPEnvelope $ envelope /*, AMQPQueue $queue*/ ) use ( $ consumer ) { $ delivery = $ this -> builder -> build ( $ envelope ) ; $ this -> feed ( $ delivery , $ consumer ) ; return $ consumer -> active ( ) ; } , $ auto_ack ? AMQP_AUTOACK : AMQP_NOPARAM ) ; } catch ( Exception $ e ) { $ outside_error = $ e ; } try { $ this -> queue -> cancel ( ) ; } catch ( Exception $ e ) { } $ consumer -> end ( $ this , $ outside_error ) ; if ( $ outside_error ) { throw $ outside_error ; } }
docstring: Attach consumer to process payload from queue
  
  @param AbstractConsumer $consumer Consumer to process payload and handle possible errors
  @param bool             $auto_ack Should message been acknowledged upon receive
  
  @return mixed
  
  @throws SerializerException
  @throws Exception           Any exception from pre/post-consume handlers and from exception handler
docstring_tokens: Attach consumer to process payload from queue