Fix snsqs non-SNS SQS message headers

.github/workflows/ci.yml

@@ -120,7 +120,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        php: ['7.3', '8.0'] # same as in the container
+        php: ['7.4', '8.0'] # same as in the container
         symfony_version: ['4.4.*', '5.2.*']
         dependencies: ['--prefer-lowest', '--prefer-dist']
         rdkafka_action: ['exclude-group', 'group']
@@ -159,5 +159,7 @@ jobs:
       - run: sed -i 's/525568/16777471/' vendor/kwn/php-rdkafka-stubs/stubs/constants.php
 
       - run: bin/dev -b
+        env:
+          PHP_VERSION: ${{ matrix.php }}
 
       - run: bin/test.sh --${{ matrix.rdkafka_action }}=rdkafka

composer.json

@@ -25,7 +25,7 @@
     "queue-interop/amqp-interop": "^0.8.2",
     "queue-interop/queue-interop": "^0.8.1",
     "bunny/bunny": "^0.4|^0.5",
-    "php-amqplib/php-amqplib": "^2.12.1",
+    "php-amqplib/php-amqplib": "^3.0",
     "doctrine/dbal": "^2.12|^3.1",
     "ramsey/uuid": "^3.5|^4",
     "psr/log": "^1.1",
@@ -42,15 +42,16 @@
     "php-http/client-common": "^2.2.1",
     "richardfullmer/rabbitmq-management-api": "^2.1.1",
     "predis/predis": "^1.1",
-    "thruway/client": "^0.5",
+    "thruway/client": "^0.5.5",
     "thruway/pawl-transport": "^0.5.1",
     "influxdb/influxdb-php": "^1.14",
     "datadog/php-datadogstatsd": "^1.3",
     "guzzlehttp/guzzle": "^7.0.1",
     "php-http/discovery": "^1.13",
     "voryx/thruway-common": "^1.0.1",
-    "react/dns": "^1.0",
-    "react/event-loop": "^1.0"
+    "react/dns": "^1.4",
+    "react/event-loop": "^1.2",
+    "react/promise": "^2.8"
   },
   "require-dev": {
     "ext-pcntl": "*",

docker-compose.yml

@@ -3,11 +3,11 @@ version: '2'
 services:
   dev:
     # when image publishing gets sorted:
-    # image: enqueue/dev:${PHP_VERSION}
+#    image: enqueue/dev:${PHP_VERSION:-7.4}
     build:
       context: docker
       args:
-        PHP_VERSION: "${PHP_VERSION:-7.3}"
+        PHP_VERSION: "${PHP_VERSION:-7.4}"
     depends_on:
       - rabbitmq
       - mysql

docker/Dockerfile

@@ -1,4 +1,4 @@
-ARG PHP_VERSION=7.3
+ARG PHP_VERSION=7.4
 FROM makasim/nginx-php-fpm:${PHP_VERSION}-all-exts
 
 ARG PHP_VERSION

pkg/enqueue-bundle/composer.json

@@ -35,7 +35,7 @@
         "enqueue/test": "0.10.x-dev",
         "enqueue/async-event-dispatcher": "0.10.x-dev",
         "enqueue/async-command": "0.10.x-dev",
-        "php-amqplib/php-amqplib": "^2.12.1",
+        "php-amqplib/php-amqplib": "^3.0",
         "doctrine/doctrine-bundle": "^2.0",
         "doctrine/mongodb-odm-bundle": "^3.5|^4.3",
         "alcaeus/mongo-php-adapter": "^1.0",

pkg/monitoring/composer.json

@@ -15,7 +15,7 @@
         "enqueue/test": "0.10.x-dev",
         "influxdb/influxdb-php": "^1.14",
         "datadog/php-datadogstatsd": "^1.3",
-        "thruway/client": "^0.5",
+        "thruway/client": "^0.5.5",
         "thruway/pawl-transport": "^0.5",
         "voryx/thruway-common": "^1.0.1"
     },

pkg/snsqs/SnsQsConsumer.php

@@ -135,6 +135,7 @@ private function convertMessage(SqsMessage $sqsMessage): SnsQsMessage
         }
 
         $message->setBody($sqsMessage->getBody());
+        $message->setHeaders($sqsMessage->getHeaders());
         $message->setProperties($sqsMessage->getProperties());
 
         return $message;

pkg/snsqs/Tests/SnsQsConsumerTest.php

@@ -0,0 +1,72 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\SnsQs\Tests;
+
+use Enqueue\SnsQs\SnsQsConsumer;
+use Enqueue\SnsQs\SnsQsContext;
+use Enqueue\SnsQs\SnsQsMessage;
+use Enqueue\SnsQs\SnsQsQueue;
+use Enqueue\Sqs\SqsConsumer;
+use Enqueue\Sqs\SqsMessage;
+use PHPUnit\Framework\TestCase;
+
+final class SnsQsConsumerTest extends TestCase
+{
+    public function testReceivesSnsMessage(): void
+    {
+        $context = $this->createMock(SnsQsContext::class);
+        $context->expects($this->once())
+            ->method('createMessage')
+            ->willReturn(new SnsQsMessage());
+
+        $sqsConsumer = $this->createMock(SqsConsumer::class);
+        $sqsConsumer->expects($this->once())
+            ->method('receive')
+            ->willReturn(new SqsMessage(json_encode([
+                'Type' => 'Notification',
+                'TopicArn' => 'arn:aws:sns:us-east-2:12345:topic-name',
+                'Message' => 'The Body',
+                'MessageAttributes' => [
+                    'Headers' => [
+                        'Type' => 'String',
+                        'Value' => '[{"headerKey":"headerVal"},{"propKey": "propVal"}]',
+                    ],
+                ],
+            ])));
+
+        $consumer = new SnsQsConsumer($context, $sqsConsumer, new SnsQsQueue('queue'));
+        $result = $consumer->receive();
+
+        $this->assertInstanceOf(SnsQsMessage::class, $result);
+        $this->assertSame('The Body', $result->getBody());
+        $this->assertSame(['headerKey' => 'headerVal'], $result->getHeaders());
+        $this->assertSame(['propKey' => 'propVal'], $result->getProperties());
+    }
+
+    public function testReceivesSqsMessage(): void
+    {
+        $context = $this->createMock(SnsQsContext::class);
+        $context->expects($this->once())
+            ->method('createMessage')
+            ->willReturn(new SnsQsMessage());
+
+        $sqsConsumer = $this->createMock(SqsConsumer::class);
+        $sqsConsumer->expects($this->once())
+            ->method('receive')
+            ->willReturn(new SqsMessage(
+                'The Body',
+                ['propKey' => 'propVal'],
+                ['headerKey' => 'headerVal'],
+            ));
+
+        $consumer = new SnsQsConsumer($context, $sqsConsumer, new SnsQsQueue('queue'));
+        $result = $consumer->receive();
+
+        $this->assertInstanceOf(SnsQsMessage::class, $result);
+        $this->assertSame('The Body', $result->getBody());
+        $this->assertSame(['headerKey' => 'headerVal'], $result->getHeaders());
+        $this->assertSame(['propKey' => 'propVal'], $result->getProperties());
+    }
+}

pkg/wamp/composer.json

@@ -9,11 +9,12 @@
         "php": "^7.3|^8.0",
         "queue-interop/queue-interop": "^0.8.1",
         "enqueue/dsn": "^0.10.8",
-        "thruway/client": "^0.5",
+        "thruway/client": "^0.5.5",
         "thruway/pawl-transport": "^0.5.1",
         "voryx/thruway-common": "^1.0.1",
-        "react/dns": "^1.0",
-        "react/event-loop": "^1.0"
+        "react/dns": "^1.4",
+        "react/event-loop": "^1.2",
+        "react/promise": "^2.8"
     },
     "require-dev": {
         "phpunit/phpunit": "^9.5",
@@ -34,10 +35,13 @@
             "/Tests/"
         ]
     },
-    "minimum-stability": "dev",
+    "minimum-stability": "beta",
     "extra": {
         "branch-alias": {
             "dev-master": "0.10.x-dev"
         }
+    },
+    "config": {
+        "prefer-stable": true
     }
 }