Message serialization.

composer.json

@@ -11,8 +11,8 @@
     "ext-mongodb": "^1.3",
     "ext-rdkafka": "^3.0.3",
 
-    "queue-interop/amqp-interop": "^0.8",
-    "queue-interop/queue-interop": "^0.7|^0.8",
+    "queue-interop/amqp-interop": "dev-master",
+    "queue-interop/queue-interop": "dev-master",
     "bunny/bunny": "^0.2.4|^0.3|^0.4",
     "php-amqplib/php-amqplib": "^2.7",
     "doctrine/dbal": "^2.6",
@@ -55,6 +55,16 @@
     "kwn/php-rdkafka-stubs": "^1.0.2",
     "friendsofphp/php-cs-fixer": "^2"
   },
+  "repositories": [
+    {
+      "type": "vcs",
+      "url": "https://github.com/Sevavietl/queue-interop"
+    },
+    {
+      "type": "vcs",
+      "url": "https://github.com/Sevavietl/amqp-interop"
+    }
+  ],
   "autoload": {
     "psr-4": {
       "Enqueue\\AmqpBunny\\": "pkg/amqp-bunny/",

pkg/amqp-bunny/AmqpContext.php

@@ -71,9 +71,11 @@ public function __construct($bunnyChannel, array $config)
     }
 
     /**
+     * @param mixed $body
+     *
      * @return InteropAmqpMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         return new AmqpMessage($body, $properties, $headers);
     }

pkg/amqp-ext/AmqpContext.php

@@ -54,9 +54,11 @@ public function __construct($extChannel)
     }
 
     /**
+     * @param mixed $body
+     *
      * @return InteropAmqpMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         return new AmqpMessage($body, $properties, $headers);
     }

pkg/amqp-lib/AmqpContext.php

@@ -60,9 +60,11 @@ public function __construct(AbstractConnection $connection, array $config)
     }
 
     /**
+     * @param mixed $body
+     *
      * @return InteropAmqpMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         return new AmqpMessage($body, $properties, $headers);
     }

pkg/dbal/DbalContext.php

@@ -64,7 +64,7 @@ public function __construct($connection, array $config = [])
     /**
      * {@inheritdoc}
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         $message = new DbalMessage();
         $message->setBody($body);

pkg/dbal/DbalMessage.php

@@ -4,14 +4,12 @@
 
 namespace Enqueue\Dbal;
 
+use Interop\Queue\Impl\StringBodyOnlyTrait;
 use Interop\Queue\Message;
 
 class DbalMessage implements Message
 {
-    /**
-     * @var string
-     */
-    private $body;
+    use StringBodyOnlyTrait;
 
     /**
      * @var array
@@ -84,16 +82,6 @@ public function __construct(string $body = '', array $properties = [], array $he
         $this->redeliverAfter = null;
     }
 
-    public function setBody(string $body): void
-    {
-        $this->body = $body;
-    }
-
-    public function getBody(): string
-    {
-        return $this->body;
-    }
-
     public function setProperties(array $properties): void
     {
         $this->properties = $properties;

pkg/dbal/DbalProducer.php

@@ -62,8 +62,6 @@ public function send(Destination $destination, Message $message): void
             $message->setTimeToLive($this->timeToLive);
         }
 
-        $body = $message->getBody();
-
         $publishedAt = null !== $message->getPublishedAt() ?
             $message->getPublishedAt() :
             (int) (microtime(true) * 10000)
@@ -72,7 +70,7 @@ public function send(Destination $destination, Message $message): void
         $dbalMessage = [
             'id' => Uuid::uuid4(),
             'published_at' => $publishedAt,
-            'body' => $body,
+            'body' => $message->getBody(),
             'headers' => JSON::encode($message->getHeaders()),
             'properties' => JSON::encode($message->getProperties()),
             'priority' => -1 * $message->getPriority(),

pkg/dbal/Tests/DbalConsumerTest.php

@@ -217,11 +217,11 @@ private function createConectionMock()
 
 class InvalidMessage implements Message
 {
-    public function getBody(): string
+    public function getBody()
     {
     }
 
-    public function setBody(string $body): void
+    public function setBody($body): void
     {
     }
 

pkg/fs/FsContext.php

@@ -58,9 +58,11 @@ public function __construct(string $storeDir, int $preFetchCount, int $chmod, in
     }
 
     /**
+     * @param mixed $body
+     *
      * @return FsMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         return new FsMessage($body, $properties, $headers);
     }

pkg/fs/FsMessage.php

@@ -9,7 +9,7 @@
 class FsMessage implements Message, \JsonSerializable
 {
     /**
-     * @var string
+     * @var string|int|float|array|\JsonSerializable
      */
     private $body;
 
@@ -28,20 +28,20 @@ class FsMessage implements Message, \JsonSerializable
      */
     private $redelivered;
 
-    public function __construct(string $body = '', array $properties = [], array $headers = [])
+    public function __construct($body = '', array $properties = [], array $headers = [])
     {
         $this->body = $body;
         $this->properties = $properties;
         $this->headers = $headers;
         $this->redelivered = false;
     }
 
-    public function setBody(string $body): void
+    public function setBody($body): void
     {
         $this->body = $body;
     }
 
-    public function getBody(): string
+    public function getBody()
     {
         return $this->body;
     }

pkg/gearman/GearmanContext.php

@@ -40,9 +40,11 @@ public function __construct(array $config)
     }
 
     /**
+     * @param mixed $body
+     *
      * @return GearmanMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         return new GearmanMessage($body, $properties, $headers);
     }

pkg/gearman/GearmanMessage.php

@@ -9,7 +9,7 @@
 class GearmanMessage implements Message, \JsonSerializable
 {
     /**
-     * @var string
+     * @var string|int|float|array|\JsonSerializable
      */
     private $body;
 
@@ -33,20 +33,20 @@ class GearmanMessage implements Message, \JsonSerializable
      */
     private $job;
 
-    public function __construct(string $body = '', array $properties = [], array $headers = [])
+    public function __construct($body = '', array $properties = [], array $headers = [])
     {
         $this->body = $body;
         $this->properties = $properties;
         $this->headers = $headers;
         $this->redelivered = false;
     }
 
-    public function setBody(string $body): void
+    public function setBody($body): void
     {
         $this->body = $body;
     }
 
-    public function getBody(): string
+    public function getBody()
     {
         return $this->body;
     }

pkg/gps/GpsContext.php

@@ -61,9 +61,11 @@ public function __construct($client, array $options = [])
     }
 
     /**
+     * @param mixed $body
+     *
      * @return GpsMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         return new GpsMessage($body, $properties, $headers);
     }

pkg/gps/GpsMessage.php

@@ -10,7 +10,7 @@
 class GpsMessage implements Message, \JsonSerializable
 {
     /**
-     * @var string
+     * @var string|int|float|array|\JsonSerializable
      */
     private $body;
 
@@ -34,7 +34,7 @@ class GpsMessage implements Message, \JsonSerializable
      */
     private $nativeMessage;
 
-    public function __construct(string $body = '', array $properties = [], array $headers = [])
+    public function __construct($body = '', array $properties = [], array $headers = [])
     {
         $this->body = $body;
         $this->properties = $properties;
@@ -43,12 +43,12 @@ public function __construct(string $body = '', array $properties = [], array $he
         $this->redelivered = false;
     }
 
-    public function getBody(): string
+    public function getBody()
     {
         return $this->body;
     }
 
-    public function setBody(string $body): void
+    public function setBody($body): void
     {
         $this->body = $body;
     }

pkg/mongodb/MongodbContext.php

@@ -41,9 +41,11 @@ public function __construct($client, array $config = [])
     }
 
     /**
+     * @param mixed $body
+     *
      * @return MongodbMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         $message = new MongodbMessage();
         $message->setBody($body);

pkg/mongodb/MongodbMessage.php

@@ -4,19 +4,17 @@
 
 namespace Enqueue\Mongodb;
 
+use Interop\Queue\Impl\StringBodyOnlyTrait;
 use Interop\Queue\Message;
 
 class MongodbMessage implements Message
 {
-    /**
-     * @var string
-     */
-    private $id;
+    use StringBodyOnlyTrait;
 
     /**
      * @var string
      */
-    private $body;
+    private $id;
 
     /**
      * @var array
@@ -75,16 +73,6 @@ public function getId(): ?string
         return $this->id;
     }
 
-    public function setBody(string $body): void
-    {
-        $this->body = $body;
-    }
-
-    public function getBody(): string
-    {
-        return $this->body;
-    }
-
     public function setProperties(array $properties): void
     {
         $this->properties = $properties;

pkg/mongodb/MongodbProducer.php

@@ -57,16 +57,14 @@ public function send(Destination $destination, Message $message): void
             $message->setTimeToLive($this->timeToLive);
         }
 
-        $body = $message->getBody();
-
         $publishedAt = null !== $message->getPublishedAt() ?
             $message->getPublishedAt() :
             (int) (microtime(true) * 10000)
         ;
 
         $mongoMessage = [
             'published_at' => $publishedAt,
-            'body' => $body,
+            'body' => $message->getBody(),
             'headers' => JSON::encode($message->getHeaders()),
             'properties' => JSON::encode($message->getProperties()),
             'priority' => $message->getPriority(),

pkg/mongodb/Tests/MongodbConsumerTest.php

@@ -131,11 +131,11 @@ private function createContextMock()
 
 class InvalidMessage implements Message
 {
-    public function getBody(): string
+    public function getBody()
     {
     }
 
-    public function setBody(string $body): void
+    public function setBody($body): void
     {
     }
 

pkg/null/NullContext.php

@@ -16,9 +16,11 @@
 class NullContext implements Context
 {
     /**
+     * @param mixed $body
+     *
      * @return NullMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         $message = new NullMessage();
         $message->setBody($body);

pkg/null/NullMessage.php

@@ -37,7 +37,7 @@ public function __construct(string $body = '', array $properties = [], array $he
         $this->redelivered = false;
     }
 
-    public function setBody(string $body): void
+    public function setBody($body): void
     {
         $this->body = $body;
     }

pkg/pheanstalk/PheanstalkContext.php

@@ -31,9 +31,11 @@ public function __construct(Pheanstalk $pheanstalk)
     }
 
     /**
+     * @param mixed $body
+     *
      * @return PheanstalkMessage
      */
-    public function createMessage(string $body = '', array $properties = [], array $headers = []): Message
+    public function createMessage($body = '', array $properties = [], array $headers = []): Message
     {
         return new PheanstalkMessage($body, $properties, $headers);
     }