support of additional properties

README.md

@@ -161,6 +161,32 @@ Amqp::consume('queue-name', function ($message, $resolver) {
 ]);
 ```
 
+## Fanout example
+
+### Publishing a message
+
+```php
+\Amqp::publish('', 'message' , [
+    'exchange_type' => 'fanout',
+    'exchange' => 'amq.fanout',
+]);
+```
+
+### Consuming messages
+
+```php
+\Amqp::consume('', function ($message, $resolver) {
+    var_dump($message->body);
+    $resolver->acknowledge($message);
+}, [
+    'exchange' => 'amq.fanout',
+    'exchange_type' => 'fanout',
+    'queue_force_declare' => true,
+    'queue_exclusive' => true,
+    'persistent' => true// required if you want to listen forever
+]);
+```
+
 ## Credits
 
 * Some concepts were used from https://github.com/mookofe/tail

src/Consumer.php

@@ -2,6 +2,7 @@
 
 use Illuminate\Config\Repository;
 use Closure;
+use PhpAmqpLib\Exception\AMQPTimeoutException;
 
 /**
  * @author Björn Schmitt <[email protected]>
@@ -45,10 +46,10 @@ public function consume($queue, Closure $closure)
             $this->getChannel()->basic_consume(
                 $queue,
                 $this->getProperty('consumer_tag'),
-                false,
-                false,
-                false,
-                false,
+                $this->getProperty('consumer_no_local'),
+                $this->getProperty('consumer_no_ack'),
+                $this->getProperty('consumer_exclusive'),
+                $this->getProperty('consumer_nowait'),
                 function ($message) use ($closure, $object) {
                     $closure($message, $object);
                 }

src/Context.php

@@ -14,7 +14,41 @@ abstract class Context
     /**
      * @var array
      */
-    protected $properties = [];
+    protected $properties = [
+        'host' => 'localhost',
+        'port' => 5672,
+        'username' => 'guest',
+        'password' => 'guest',
+        'vhost' => '/',
+        'persistent' => false,
+
+        'exchange' => 'amq.topic',
+        'exchange_type' => 'topic',
+        'exchange_passive' => false,
+        'exchange_durable' => true,
+        'exchange_auto_delete' => false,
+        'exchange_internal' => false,
+        'exchange_nowait' => false,
+        'exchange_properties' => [],
+
+        'consumer_tag' => '',
+        'consumer_no_local' => false,
+        'consumer_no_ack' => false,
+        'consumer_exclusive' => false,
+        'consumer_nowait' => false,
+
+        'queue_force_declare' => false,
+        'queue_passive' => false,
+        'queue_durable' => true,
+        'queue_exclusive' => false,
+        'queue_auto_delete' => false,
+        'queue_nowait' => false,
+
+        'queue_properties' => [],
+        'connect_options' => [],
+        'ssl_options' => [],
+        'timeout' => 0,
+    ];
 
     /**
      * Context constructor.
@@ -33,7 +67,7 @@ protected function extractProperties(Repository $config)
     {
         if ($config->has(self::REPOSITORY_KEY)) {
             $data = $config->get(self::REPOSITORY_KEY);
-            $this->properties = $data['properties'][$data['use']];
+            $this->mergeProperties($data['properties'][$data['use']]);
         }
     }
 

src/Request.php

@@ -68,39 +68,40 @@ public function setup()
         $this->channel->exchange_declare(
             $exchange,
             $this->getProperty('exchange_type'),
-            false,
-            true,
-            false,
-            false,
-            false,
+            $this->getProperty('exchange_passive'),
+            $this->getProperty('exchange_durable'),
+            $this->getProperty('exchange_auto_delete'),
+            $this->getProperty('exchange_internal'),
+            $this->getProperty('exchange_nowait'),
             $this->getProperty('exchange_properties')
         );
 
         $queue = $this->getProperty('queue');
 
-        if (!empty($queue)) {
+        if (!empty($queue) || $this->getProperty('queue_force_declare')) {
 
             /*
                 name: $queue
                 passive: false
                 durable: true // the queue will survive server restarts
-                exclusive: false // the queue can be accessed in other channels
+                exclusive: false // queue is deleted when connection closes
                 auto_delete: false //the queue won't be deleted once the channel is closed.
                 nowait: false // Doesn't wait on replies for certain things.
                 parameters: array // Extra data, like high availability params
             */
 
+            /** @var ['queue name', 'message count',] queueInfo */
             $this->queueInfo = $this->channel->queue_declare(
                 $queue,
-                false,
-                true,
-                false,
-                false,
-                false,
+                $this->getProperty('queue_passive'),
+                $this->getProperty('queue_durable'),
+                $this->getProperty('queue_exclusive'),
+                $this->getProperty('queue_auto_delete'),
+                $this->getProperty('queue_nowait'),
                 $this->getProperty('queue_properties')
             );
 
-            $this->channel->queue_bind($queue, $exchange, $this->getProperty('routing'));
+            $this->channel->queue_bind($queue ?: $this->queueInfo[0], $exchange, $this->getProperty('routing'));
 
         }