1.x dev (#5)

* added rabbitmq amqp support

Author: hrodic

.integration-testing.json

@@ -25,16 +25,16 @@
  "port": 5672,
  "user": "test",
  "password": "test",
- "vhost": "test",
+ "vhost": "/",
  "fixtures": {
  "beforeFirstTest": {
  "purgeQueues": [
- "test-queue"
+ "before-first-test-queue"
  ],
  "publishMessages": [
  {
  "exchange": "test-exchange",
- "queue": "test-queue",
+ "queue": "before-first-test-queue",
  "routing_key": "before-first-test",
  "path": "tests/fixtures/before-first-test",
  "extension": "json"
@@ -43,26 +43,24 @@
  },
  "beforeTest": {
  "purgeQueues": [
- "test-queue"
+ "before-test-queue"
  ],
  "publishMessages": [
  {
  "exchange": "test-exchange",
- "queue": "test-queue",
+ "queue": "before-test-queue",
  "routing_key": "before-test",
  "path": "tests/fixtures/before-test"
  }
  ]
  },
  "afterTest": {
  "purgeQueues": [
- "test-queue"
+ "before-test-queue"
  ]
  },
  "afterLastTest": {
- "purgeQueues": [
- "test-queue"
- ]
+ "purgeQueues": []
  }
  }
  }

Makefile

@@ -18,4 +18,4 @@ test-integration: up
 merge-coverage:
 vendor/bin/phpcov merge --clover build/coverage/merged.xml build/coverage
 debug-test-integration:
-php -dxdebug.remote_mode=jit vendor/bin/phpunit --no-coverage --color --testdox --group debug -c phpunit-integration.xml.dist
+php vendor/bin/phpunit --no-coverage --color --testdox --group debug -c phpunit-integration.xml.dist

composer.json

@@ -21,21 +21,23 @@
  "prefer-stable": true,
  "require": {
  "php": "^7.3",
- "ext-pdo": "*",
- "ext-json": "*"
+ "ext-json": "*",
+ "ext-pdo": "*"
  },
  "require-dev": {
  "ext-json": "*",
  "ext-mbstring": "*",
  "ext-pdo_mysql": "*",
  "ext-xml": "*",
  "friendsofphp/php-cs-fixer": "^2.16",
+ "php-amqplib/php-amqplib": "^2.11",
  "phpunit/phpcov": "^6.0",
  "phpunit/phpunit": "^8.5",
  "squizlabs/php_codesniffer": "^3.5"
  },
  "suggest": {
- "ext-pdo_mysql": "For PDO related integration tests"
+ "ext-pdo_mysql": "For PDO related integration tests",
+ "php-amqplib/php-amqplib": "For AMQP related integration tests"
  },
  "autoload": {
  "psr-4": {

docker-compose.yml

@@ -32,11 +32,13 @@ services:
  environment:
  - RABBITMQ_VM_MEMORY_HIGH_WATERMARK=0.5
  - RABBITMQ_ERLANG_COOKIE=secret
- - RABBITMQ_DEFAULT_USER=user
- - RABBITMQ_DEFAULT_PASS=password
- - RABBITMQ_DEFAULT_VHOST=test
+ - RABBITMQ_DEFAULT_USER=test
+ - RABBITMQ_DEFAULT_PASS=test
+ - RABBITMQ_DEFAULT_VHOST=/
  tmpfs:
  - /var/lib/rabbitmq
  ports:
  - 5672:5672
  - 15672:15672
+ volumes:
+ - $PWD/docker/rabbitmq/enabled_plugins:/etc/rabbitmq/enabled_plugins

docker/rabbitmq/enabled_plugins

@@ -0,0 +1 @@
+[rabbitmq_management].

phpunit-integration.xml.dist

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/9.0/phpunit.xsd"
- bootstrap="vendor/autoload.php"
+ bootstrap="tests/integration/Bootstrap.php"
  executionOrder="depends,defects"
  forceCoversAnnotation="true"
  beStrictAboutCoversAnnotation="true"
@@ -13,6 +13,10 @@
  <const name="DB_DSN" value="mysql:host=localhost:3306;dbname=test;charset=utf8"/>
  <const name="DB_USERNAME" value="test"/>
  <const name="DB_PASSWORD" value="test"/>
+ <const name="RABBITMQ_HOST" value="localhost"/>
+ <const name="RABBITMQ_PORT" value="5672"/>
+ <const name="RABBITMQ_USER" value="test"/>
+ <const name="RABBITMQ_PASSWORD" value="test"/>
  </php>
  <testsuites>
  <testsuite name="integration">

src/Driver/AMQPConnection.php

@@ -0,0 +1,83 @@
+<?php declare(strict_types=1);
+
+namespace IntegrationTesting\Driver;
+
+use PhpAmqpLib\Channel\AMQPChannel;
+use PhpAmqpLib\Connection\AMQPStreamConnection;
+
+/**
+ * Class AMQPConnection
+ * @package IntegrationTesting\Driver
+ * @internal NOT FOR PUBLIC USE
+ */
+class AMQPConnection
+{
+ /**
+ * @var AMQPStreamConnection
+ */
+ private $connection;
+
+ private function __construct(
+ AMQPStreamConnection $connection
+ ) {
+ $this->connection = $connection;
+ $this->connection->set_close_on_destruct(true);
+ }
+
+ public static function create(
+ string $host,
+ int $port,
+ string $user,
+ string $password,
+ string $vHost = '/',
+ bool $insist = false,
+ string $loginMethod = 'AMQPLAIN',
+ string $locale = 'en_US',
+ float $connectionTimeout = 3.0,
+ float $readWriteTimeout = 3.0,
+ bool $keepAlive = false,
+ int $heartbeat = 0
+ ): self {
+ return new self(
+ new AMQPStreamConnection(
+ $host,
+ $port,
+ $user,
+ $password,
+ $vHost,
+ $insist,
+ $loginMethod,
+ null,
+ $locale,
+ $connectionTimeout,
+ $readWriteTimeout,
+ null,
+ $keepAlive,
+ $heartbeat
+ )
+ );
+ }
+
+ public function connect(): void
+ {
+ if (!$this->connection->isConnected()) {
+ $this->connection->reconnect();
+ }
+ }
+
+ public function disconnect(): void
+ {
+ if ($this->connection->isConnected()) {
+ $this->connection->close();
+ }
+ }
+
+ /**
+ * @param int $channelId
+ * @return AMQPChannel
+ */
+ public function getChannel(int $channelId = null): AMQPChannel
+ {
+ return $this->connection->channel($channelId);
+ }
+}

src/Driver/AMQPService.php

@@ -0,0 +1,37 @@
+<?php declare(strict_types=1);
+
+namespace IntegrationTesting\Driver;
+
+use PhpAmqpLib\Channel\AMQPChannel;
+
+/**
+ * Interface AMQPService
+ * @package IntegrationTesting\Driver
+ * @internal NOT FOR PUBLIC USE
+ */
+interface AMQPService
+{
+ public function createChannel(): AMQPChannel;
+
+ public function createDurableTopicExchanges(array $exchangeNames): void;
+
+ public function createDurableTopicExchange(string $name, array $arguments = []): AMQPChannel;
+
+ public function forceDeleteExchange(string $name): AMQPChannel;
+
+ public function forceDeleteQueue(string $name): AMQPChannel;
+
+ public function purgeQueue(string $name): AMQPChannel;
+
+ public function disconnect(): void;
+
+ public function createDurableQueue(string $name, array $arguments = []): AMQPChannel;
+
+ public function createDurableQueues(array $queueNames): void;
+
+ public function bindExchangeToQueueByRoutingKeys(string $exchangeName, string $queueName, array $routingKeys): void;
+
+ public function publishMessage(AMQPChannel $channel, string $body, array $properties, string $exchange, string $routingKey): void;
+
+ public function consumeMessage(AMQPChannel $channel, string $consumerTag, string $queue, callable $callback): void;
+}

src/Driver/FileSystem.php

@@ -7,6 +7,11 @@
 use IntegrationTesting\Exception\TestingException;
 use Iterator;
 
+/**
+ * Class FileSystem
+ * @package IntegrationTesting\Driver
+ * @internal NOT FOR PUBLIC USE
+ */
 class FileSystem
 {
  /**
@@ -53,11 +58,12 @@ public function getFileListIteratorFromPathByExtension(string $path, string $ext
  /**
  * @param Iterator $iterator
  * @param callable $callback
+ * @throws TestingException
  */
  public function runCallbackOnEachFileIteratorContents(Iterator $iterator, callable $callback): void
  {
  foreach ($iterator as $filePath) {
- $callback(file_get_contents($filePath));
+ $callback($this->getFileContents($filePath));
  }
  }
 }

src/Driver/PDOConnection.php

@@ -4,6 +4,11 @@
 
 use PDO;
 
+/**
+ * Class PDOConnection
+ * @package IntegrationTesting\Driver
+ * @internal NOT FOR PUBLIC USE
+ */
 class PDOConnection
 {
  private $PDO;