#353 - added logger rabbitmq

Author: alexbumbacea

DependencyInjection/Configuration.php

@@ -88,6 +88,7 @@ protected function addProducers(ArrayNodeDefinition $node)
                             ->scalarNode('connection')->defaultValue('default')->end()
                             ->scalarNode('auto_setup_fabric')->defaultTrue()->end()
                             ->scalarNode('class')->defaultValue('%old_sound_rabbit_mq.producer.class%')->end()
+                            ->scalarNode('audit')->defaultFalse()->end()
                         ->end()
                     ->end()
                 ->end()
@@ -143,6 +144,7 @@ protected function addConsumers(ArrayNodeDefinition $node)
                                     ->booleanNode('global')->defaultFalse()->end()
                                 ->end()
                             ->end()
+                            ->scalarNode('audit')->defaultFalse()->end()
                         ->end()
                     ->end()
                 ->end()
@@ -174,6 +176,7 @@ protected function addMultipleConsumers(ArrayNodeDefinition $node)
                             ->end()
                         ->end()
                         ->scalarNode('queues_provider')->defaultNull()->end()
+                        ->scalarNode('audit')->defaultFalse()->end()
                     ->end()
                 ->end()
             ->end()
@@ -204,6 +207,7 @@ protected function addDynamicConsumers(ArrayNodeDefinition $node)
                                 ->end()
                             ->end()
                             ->scalarNode('queue_options_provider')->isRequired()->end()
+                            ->scalarNode('audit')->defaultFalse()->end()
                         ->end()
                     ->end()
                 ->end()
@@ -275,6 +279,7 @@ protected function addRpcServers(ArrayNodeDefinition $node)
                                 ->end()
                             ->end()
                             ->scalarNode('serializer')->defaultValue('serialize')->end()
+                            ->scalarNode('audit')->defaultFalse()->end()
                         ->end()
                     ->end()
                 ->end()

DependencyInjection/OldSoundRabbitMqExtension.php

@@ -150,6 +150,10 @@ protected function loadProducers()
                     $definition->addMethodCall('disableAutoSetupFabric');
                 }
 
+                if ($producer['audit']) {
+                    $this->injectLogger($definition);
+                }
+
                 $this->container->setDefinition(sprintf('old_sound_rabbit_mq.%s_producer', $key), $definition);
             }
         } else {
@@ -191,6 +195,10 @@ protected function loadConsumers()
                 $this->injectLoggedChannel($definition, $key, $consumer['connection']);
             }
 
+            if ($consumer['audit']) {
+                $this->injectLogger($definition);
+            }
+
             $name = sprintf('old_sound_rabbit_mq.%s_consumer', $key);
             $this->container->setDefinition($name, $definition);
             $this->addDequeuerAwareCall($consumer['callback'], $name);
@@ -250,6 +258,10 @@ protected function loadMultipleConsumers()
                 $this->injectLoggedChannel($definition, $key, $consumer['connection']);
             }
 
+            if ($consumer['audit']) {
+                $this->injectLogger($definition);
+            }
+
             $name = sprintf('old_sound_rabbit_mq.%s_multiple', $key);
             $this->container->setDefinition($name, $definition);
             if ($consumer['queues_provider']) {
@@ -305,6 +317,10 @@ protected function loadDynamicConsumers()
                 $this->injectLoggedChannel($definition, $key, $consumer['connection']);
             }
 
+            if ($consumer['audit']) {
+                $this->injectLogger($definition);
+            }
+
             $name = sprintf('old_sound_rabbit_mq.%s_dynamic', $key);
             $this->container->setDefinition($name, $definition);
             $this->addDequeuerAwareCall($consumer['callback'], $name);
@@ -481,4 +497,11 @@ protected function addDequeuerAwareCall($callback, $name)
             $callbackDefinition->addMethodCall('setDequeuer', array(new Reference($name)));
         }
     }
+
+    private function injectLogger(Definition $definition)
+    {
+        $definition->addTag('monolog.logger', [
+            'channel' => 'phpamqplib'
+        ]);
+    }
 }

README.md

@@ -361,6 +361,23 @@ This seems to be quite a lot of work for just sending messages, let's recap to h
 
 And that's it!
 
+### Audit / Logging ###
+
+This was a requirement to have a traceability of messages received/published.
+In order to enable this you'll need to add "audit" config to consumers or publishers.
+
+```yaml
+consumers:
+    upload_picture:
+        connection:       default
+        exchange_options: {name: 'upload-picture', type: direct}
+        queue_options:    {name: 'upload-picture'}
+        callback:         upload_picture_service
+        audit: true
+```
+
+If you would like you can also treat logging from queues with different handlers in monolog, by referencing channel "phpamqplib"
+
 ### RPC or Reply/Response ###
 
 So far we just have sent messages to consumers, but what if we want to get a reply from them? To achieve this we have to implement RPC calls into our application. This bundle makes it pretty easy to achieve such things with Symfony2.

RabbitMq/BaseAmqp.php

@@ -4,6 +4,8 @@
 use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Connection\AMQPLazyConnection;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
 
 abstract class BaseAmqp
 {
@@ -16,6 +18,11 @@ abstract class BaseAmqp
     protected $autoSetupFabric = true;
     protected $basicProperties = array('content_type' => 'text/plain', 'delivery_mode' => 2);
 
+    /**
+     * @var LoggerInterface
+     */
+    protected $logger;
+    
     protected $exchangeOptions = array(
         'passive' => false,
         'durable' => true,
@@ -53,6 +60,8 @@ public function __construct(AbstractConnection $conn, AMQPChannel $ch = null, $c
         }
 
         $this->consumerTag = empty($consumerTag) ? sprintf("PHPPROCESS_%s_%s", gethostname(), getmypid()) : $consumerTag;
+
+        $this->logger = new NullLogger();
     }
 
     public function __destruct()
@@ -187,4 +196,12 @@ public function setupFabric()
     public function disableAutoSetupFabric() {
         $this->autoSetupFabric = false;
     }
+
+    /**
+     * @param LoggerInterface $logger
+     */
+    public function setLogger($logger)
+    {
+        $this->logger = $logger;
+    }
 }

RabbitMq/Consumer.php

@@ -67,9 +67,36 @@ public function delete()
 
     public function processMessage(AMQPMessage $msg)
     {
-        $processFlag = call_user_func($this->callback, $msg);
+        try {
+            $processFlag = call_user_func($this->callback, $msg);
+            $this->handleProcessMessage($msg, $processFlag);
+            $this->logger->debug('Queue message processed', array(
+                'amqp' => array(
+                    'queue' => $this->queueOptions['name'],
+                    'message' => $msg,
+                    'return_code' => $processFlag
+                )
+            ));
+        } catch (\Exception $e) {
+            $this->logger->error($e->getMessage(), array(
+                'amqp' => array(
+                    'queue' => $this->queueOptions['name'],
+                    'message' => $msg,
+                    'stacktrace' => $e->getTraceAsString()
+                )
+            ));
+            throw $e;
+        } catch (\Error $e) {
+            $this->logger->error($e->getMessage(), array(
+                'amqp' => array(
+                    'queue' => $this->queueOptions['name'],
+                    'message' => $msg,
+                    'stacktrace' => $e->getTraceAsString()
+                )
+            ));
+            throw $e;
+        }
 
-        $this->handleProcessMessage($msg, $processFlag);
     }
 
     protected function handleProcessMessage(AMQPMessage $msg, $processFlag)

RabbitMq/Producer.php

@@ -2,7 +2,6 @@
 
 namespace OldSound\RabbitMqBundle\RabbitMq;
 
-use OldSound\RabbitMqBundle\RabbitMq\BaseAmqp;
 use PhpAmqpLib\Message\AMQPMessage;
 use PhpAmqpLib\Wire\AMQPTable;
 
@@ -49,11 +48,19 @@ public function publish($msgBody, $routingKey = '', $additionalProperties = arra
 
         $msg = new AMQPMessage((string) $msgBody, array_merge($this->getBasicProperties(), $additionalProperties));
 
-        if(!empty($headers)){
+        if (!empty($headers)) {
             $headersTable = new AMQPTable($headers);
             $msg->set('application_headers', $headersTable);
         }
 
-        $this->getChannel()->basic_publish($msg, $this->exchangeOptions['name'], (string) $routingKey);
+        $this->getChannel()->basic_publish($msg, $this->exchangeOptions['name'], (string)$routingKey);
+        $this->logger->debug('AMQP message published', array(
+            'amqp' => array(
+                'body' => $msgBody,
+                'routingkeys' => $routingKey,
+                'properties' => $additionalProperties,
+                'headers' => $headers
+            )
+        ));
     }
 }

Tests/DependencyInjection/Fixtures/config_with_audit.yml

@@ -0,0 +1,22 @@
+old_sound_rabbit_mq:
+
+    enable_collector: true
+
+    connections:
+        default:
+
+    producers:
+        default_producer:
+            exchange_options:
+                name:       default_exchange
+                type:       direct
+
+    consumers:
+        default_consumer:
+            exchange_options:
+                name:       default_exchange
+                type:       direct
+            queue_options:
+                name:       default_queue
+            callback:       default.callback
+            audit:      true

Tests/DependencyInjection/OldSoundRabbitMqExtensionTest.php

@@ -805,6 +805,15 @@ public function testProducerWithoutExplicitExchangeOptionsConnectsToAMQPDefault(
         $this->assertEquals(true, $options[0]['passive']);
     }
 
+    public function testProducersWithLogger()
+    {
+        $container = $this->getContainer('config_with_audit.yml');
+        $this->assertTrue(
+            $container->getDefinition('old_sound_rabbit_mq.default_consumer_consumer')->hasTag('monolog.logger'),
+            'service should be marked for logger'
+        );
+    }
+
     private function getContainer($file, $debug = false)
     {
         $container = new ContainerBuilder(new ParameterBag(array('kernel.debug' => $debug)));

composer.json

@@ -15,7 +15,8 @@
         "symfony/config":               "~2.3 || ~3.0",
         "symfony/yaml":                 "~2.3 || ~3.0",
         "symfony/console":              "~2.3 || ~3.0",
-        "php-amqplib/php-amqplib":      "~2.6"
+        "php-amqplib/php-amqplib":      "~2.6",
+        "psr/log": "~1.0"
     },
     "require-dev": {
         "symfony/serializer":           "~2.3 || ~3.0",