fixed receiver (#5)

Message/DomainEventMessage.php

@@ -37,7 +37,7 @@ public function __construct(string $name, string $type, string $id, array $paylo
         $this->type = $type;
         $this->id = $id;
         $this->payload = $payload;
-        $this->created = new \DateTimeImmutable();
+        $this->created = (new \DateTimeImmutable())->format('c');
     }
 
     public function getName(): string
@@ -60,7 +60,7 @@ public function getPayload(): array
         return $this->payload;
     }
 
-    public function getCreated(): \DateTimeImmutable
+    public function getCreated(): string
     {
         return $this->created;
     }
@@ -72,7 +72,7 @@ public function toArray(): array
             'type' => $this->type,
             'id' => $this->id,
             'payload' => $this->payload,
-            'created' => $this->created->format('c'),
+            'created' => $this->created,
         ];
     }
 }

Transport/RedisStreamReceiver.php

@@ -14,7 +14,7 @@
 namespace HandcraftedInTheAlps\Bundle\RedisTransportBundle\Transport;
 
 use Redis;
-use Symfony\Component\Messenger\Transport\ReceiverInterface;
+use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\Serializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 
@@ -57,9 +57,16 @@ public function __construct(Redis $redis, string $stream, string $group = null,
     public function receive(callable $handler): void
     {
         foreach ($this->read() as $key => $message) {
-            // TODO receive message
+            $content = (array) json_decode($message['content']);
 
-            $this->ack($key, $message);
+            $content = [
+                'body' => $content['body'],
+                'headers' => (array) $content['headers'],
+            ];
+
+            $handler($this->serializer->decode($content));
+
+            $this->ack($key);
         }
     }
 
@@ -112,7 +119,7 @@ private function read()
         }
     }
 
-    private function ack(string $key, array $message)
+    private function ack(string $key)
     {
         if ($this->group) {
             $this->redis->xAck($this->stream, $this->group, [$key]);

Transport/RedisStreamSender.php

@@ -47,7 +47,7 @@ public function send(Envelope $envelope): Envelope
     {
         $encodedMessage = $this->serializer->encode($envelope);
 
-        $this->redis->xAdd($this->stream, '*', $encodedMessage);
+        $this->redis->xAdd($this->stream, '*', ['content' => json_encode($encodedMessage)]);
 
         return $envelope;
     }