Bugfix in RdKafkaProducer:flush() method

bin/changelog

@@ -8,6 +8,6 @@ then
     exit 1
 fi
 
-docker-compose run -e CHANGELOG_GITHUB_TOKEN=${CHANGELOG_GITHUB_TOKEN:-""} --workdir="/mqdev" --rm generate-changelog github_changelog_generator --future-release "$1" --no-issues --unreleased-only --output "CHANGELOG_FUTURE.md"
+docker compose run -e CHANGELOG_GITHUB_TOKEN=${CHANGELOG_GITHUB_TOKEN:-""} --workdir="/mqdev" --rm generate-changelog github_changelog_generator --future-release "$1" --no-issues --unreleased-only --output "CHANGELOG_FUTURE.md"
 
-#git add CHANGELOG.md && git commit -m "Release $1" -S && git push origin "$CURRENT_BRANCH"
+#git add CHANGELOG.md && git commit -m "Release $1" -S && git push origin "$CURRENT_BRANCH"

bin/dev

@@ -6,22 +6,22 @@ set -e
 while getopts "bustefdp" OPTION; do
   case $OPTION in
     b)
-      docker-compose pull -q && docker-compose build
+      docker compose pull -q && docker compose build
       ;;
     u)
-      docker-compose up
+      docker compose up
       ;;
     s)
-      docker-compose stop
+      docker compose stop
       ;;
     e)
-      docker exec -it mqdev_dev_1 /bin/bash
+      docker exec -it enqueue-dev-dev-1 /bin/bash
       ;;
     f)
       ./bin/php-cs-fixer fix
       ;;
 
-    d) docker-compose run --workdir="/mqdev" --rm dev php pkg/enqueue-bundle/Tests/Functional/app/console.php config:dump-reference enqueue -vvv
+    d) docker compose run --workdir="/mqdev" --rm dev php pkg/enqueue-bundle/Tests/Functional/app/console.php config:dump-reference enqueue -vvv
       ;;
     \?)
       echo "Invalid option: -$OPTARG" >&2

bin/test.sh

@@ -3,4 +3,4 @@
 set -x
 set -e
 
-docker-compose run --workdir="/mqdev" --rm dev ./docker/bin/test.sh $@
+docker compose run --workdir="/mqdev" --rm dev ./docker/bin/test.sh $@

docker-compose.yml

@@ -1,5 +1,3 @@
-version: '2'
-
 services:
   dev:
     # when image publishing gets sorted:
@@ -38,16 +36,16 @@ services:
       - PREDIS_DSN=redis+predis://redis
       - PHPREDIS_DSN=redis+phpredis://redis
       - GPS_DSN=gps:?projectId=mqdev&emulatorHost=http://google-pubsub:8085
-      - SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4576&version=latest
-      - SNS_DSN=sns:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4575&version=latest
-      - SNSQS_DSN=snsqs:?key=key&secret=secret&region=us-east-1&sns_endpoint=http://localstack:4575&sqs_endpoint=http://localstack:4576&version=latest
+      - SQS_DSN=sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
+      - SNS_DSN=sns:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
+      - SNSQS_DSN=snsqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localstack:4566&version=latest
       - WAMP_DSN=wamp://thruway:9090
       - REDIS_HOST=redis
       - REDIS_PORT=6379
       - AWS_SQS_KEY=key
       - AWS_SQS_SECRET=secret
       - AWS_SQS_REGION=us-east-1
-      - AWS_SQS_ENDPOINT=http://localstack:4576
+      - AWS_SQS_ENDPOINT=http://localstack:4566
       - AWS_SQS_VERSION=latest
       - BEANSTALKD_DSN=beanstalk://beanstalkd:11300
       - GEARMAN_DSN=gearman://gearmand:4730
@@ -127,13 +125,19 @@ services:
       - '9090:9090'
 
   localstack:
-    image: 'localstack/localstack:0.8.10'
+    image: 'localstack/localstack:3.6.0'
     ports:
-      - '4576:4576'
-      - '4575:4575'
+      - '4566:4566'
     environment:
-      HOSTNAME_EXTERNAL: 'localstack'
+      LOCALSTACK_HOST: 'localstack'
       SERVICES: 'sqs,sns'
+      EAGER_SERVICE_LOADING: '1'
+    networks:
+      default:
+        aliases:
+          - 'localstack'
+          - 'sqs.us-east-1.localstack'
+          - 'snsqs.us-east-1.localstack'
 
   influxdb:
     image: 'influxdb:latest'

docker/bin/test.sh

@@ -27,6 +27,34 @@ function waitForService()
     printf "service is online %s:%s\n" $1 $2
 }
 
+function waitForServiceLocalStack()
+{
+    waitForService localstack 4566 $1
+
+    ATTEMPTS=0
+    RESPONSE=''
+    READY=0
+    until [ $READY = 1 ]; do
+        printf "check readiness for service localstack:4566\n"
+        RESPONSE=$(curl -s http://localstack:4566/_localstack/health)
+        if [[ $RESPONSE == *'"sns": "available"'* && $RESPONSE == *'"sqs": "available"'* ]]; then
+            READY=1
+        fi
+        ((ATTEMPTS++))
+        if [ $ATTEMPTS -ge $2 ]; then
+            printf "service is not ready localstack:4566\n"
+            exit 1
+        fi
+        if [ "$FORCE_EXIT" = true ]; then
+            exit;
+        fi
+
+        sleep 1
+    done
+
+    printf "service is ready localstack:4566\n"
+}
+
 trap "FORCE_EXIT=true" SIGTERM SIGINT
 
 waitForService rabbitmq 5672 50
@@ -39,7 +67,7 @@ waitForService gearmand 4730 50
 waitForService kafka 9092 50
 waitForService mongo 27017 50
 waitForService thruway 9090 50
-waitForService localstack 4576 50
+waitForServiceLocalStack 50 50
 
 php docker/bin/refresh-mysql-database.php || exit 1
 php docker/bin/refresh-postgres-database.php  || exit 1

docs/contribution.md

@@ -49,6 +49,6 @@ Once everything is done open a pull request on official repository.
 
 ## WTF?!
 
-* If you get `rabbitmqssl: forward host lookup failed: Unknown host, wait for service rabbitmqssl:5671` do `docker-compose down`.
+* If you get `rabbitmqssl: forward host lookup failed: Unknown host, wait for service rabbitmqssl:5671` do `docker compose down`.
 
 [back to index](index.md)

pkg/amqp-ext/Tests/Functional/AmqpCommonUseCasesTest.php

@@ -158,7 +158,7 @@ public function testConsumerReceiveMessageFromTopicDirectly()
         $this->amqpContext->declareTopic($topic);
 
         $consumer = $this->amqpContext->createConsumer($topic);
-        //guard
+        // guard
         $this->assertNull($consumer->receive(1000));
 
         $message = $this->amqpContext->createMessage(__METHOD__);
@@ -168,9 +168,9 @@ public function testConsumerReceiveMessageFromTopicDirectly()
         $actualMessage = $consumer->receive(1000);
 
         $this->assertInstanceOf(AmqpMessage::class, $actualMessage);
-        $consumer->acknowledge($message);
+        $consumer->acknowledge($actualMessage);
 
-        $this->assertEquals(__METHOD__, $message->getBody());
+        $this->assertEquals(__METHOD__, $actualMessage->getBody());
     }
 
     public function testConsumerReceiveMessageWithZeroTimeout()
@@ -181,7 +181,7 @@ public function testConsumerReceiveMessageWithZeroTimeout()
         $this->amqpContext->declareTopic($topic);
 
         $consumer = $this->amqpContext->createConsumer($topic);
-        //guard
+        // guard
         $this->assertNull($consumer->receive(1000));
 
         $message = $this->amqpContext->createMessage(__METHOD__);
@@ -192,9 +192,9 @@ public function testConsumerReceiveMessageWithZeroTimeout()
         $actualMessage = $consumer->receive(0);
 
         $this->assertInstanceOf(AmqpMessage::class, $actualMessage);
-        $consumer->acknowledge($message);
+        $consumer->acknowledge($actualMessage);
 
-        $this->assertEquals(__METHOD__, $message->getBody());
+        $this->assertEquals(__METHOD__, $actualMessage->getBody());
     }
 
     public function testPurgeMessagesFromQueue()

pkg/rdkafka/RdKafkaProducer.php

@@ -70,7 +70,7 @@ public function send(Destination $destination, Message $message): void
     /**
      * @return RdKafkaProducer
      */
-    public function setDeliveryDelay(int $deliveryDelay = null): Producer
+    public function setDeliveryDelay(?int $deliveryDelay = null): Producer
     {
         if (null === $deliveryDelay) {
             return $this;
@@ -87,7 +87,7 @@ public function getDeliveryDelay(): ?int
     /**
      * @return RdKafkaProducer
      */
-    public function setPriority(int $priority = null): Producer
+    public function setPriority(?int $priority = null): Producer
     {
         if (null === $priority) {
             return $this;
@@ -101,7 +101,7 @@ public function getPriority(): ?int
         return null;
     }
 
-    public function setTimeToLive(int $timeToLive = null): Producer
+    public function setTimeToLive(?int $timeToLive = null): Producer
     {
         if (null === $timeToLive) {
             return $this;
@@ -121,5 +121,7 @@ public function flush(int $timeout): ?int
         if (method_exists($this->producer, 'flush')) {
             return $this->producer->flush($timeout);
         }
+
+        return null;
     }
 }

pkg/snsqs/Tests/SnsQsProducerTest.php

@@ -40,7 +40,7 @@ public function testCouldBeConstructedWithRequiredArguments()
     public function testShouldThrowIfMessageIsInvalidType()
     {
         $this->expectException(InvalidMessageException::class);
-        $this->expectExceptionMessage('The message must be an instance of Enqueue\SnsQs\SnsQsMessage but it is Double\Message\P4');
+        $this->expectExceptionMessageMatches('/The message must be an instance of Enqueue\\\\SnsQs\\\\SnsQsMessage but it is Double\\\\Message\\\\P\d+/');
 
         $producer = new SnsQsProducer($this->createSnsContextMock(), $this->createSqsContextMock());