public function consume($message, array $headers = [])
    {
        try {
            $this->transactionManager->beginTransaction();
        } catch (BeginException $e) {
            throw new ConsumerException($e->getMessage(), $e->getCode(), $e);
        }

        try {
            $this->consumer->consume($message, $headers);
            $this->transactionManager->commit();
        } catch (\Exception $e) {
            $this->transactionManager->rollback();
            throw $e;
        }
    }
public function consume($message, array $headers = [])
    {
        try {
            $this->transactionManager->beginTransaction();
        } catch (BeginException $e) {
            throw new ConsumerException($e->getMessage(), $e->getCode(), $e);
        }

        try {
            $this->consumer->consume($message, $headers);
            $this->transactionManager->commit();
        } catch (\Exception $e) {
            $this->transactionManager->rollback();
            throw $e;
        }
    }
public function consume ( $ message , array $ headers = [ ] ) { try { $ this -> transactionManager -> beginTransaction ( ) ; } catch ( BeginException $ e ) { throw new ConsumerException ( $ e -> getMessage ( ) , $ e -> getCode ( ) , $ e ) ; } try { $ this -> consumer -> consume ( $ message , $ headers ) ; $ this -> transactionManager -> commit ( ) ; } catch ( \ Exception $ e ) { $ this -> transactionManager -> rollback ( ) ; throw $ e ; } }
Consumes a message.

@param string $message
@param array  $headers

@return null|string|void

@throws \Exception
Consumes a message .