Use amphp sockets && refactoring

README.md

@@ -23,10 +23,36 @@ $ composer require phpinnacle/ridge
 ```php
 <?php
 
+use Amp\Loop;
+use PHPinnacle\Ridge\Channel;
+use PHPinnacle\Ridge\Client;
+use PHPinnacle\Ridge\Config;
+use PHPinnacle\Ridge\Message;
+
 require __DIR__ . '/vendor/autoload.php';
 
-Amp\Loop::run(function () {
+Loop::run(function () {
+    $config = Config::dsn('amqp://admin:admin123@172.23.0.3');
+    $client = new Client($config);
+
+    yield $client->connect();
+
+    /** @var Channel $channel */
+    $channel = yield $client->channel();
+
+    yield $channel->queueDeclare('queue_name');
+
+    for ($i = 0; $i < 10; $i++) {
+        yield $channel->publish("test_$i", '', 'queue_name');
+    }
+
+    yield $channel->consume(function (Message $message, Channel $channel) {
+        echo $message->content() . \PHP_EOL;
+
+        yield $channel->ack($message);
+    }, 'queue_name');
 });
+
 ```
 
 More examples can be found in [`examples`](examples) directory.

benchmark/consumer.php

@@ -0,0 +1,40 @@
+<?php
+
+use Amp\Loop;
+use PHPinnacle\Ridge\Channel;
+use PHPinnacle\Ridge\Client;
+use PHPinnacle\Ridge\Config;
+use PHPinnacle\Ridge\Message;
+
+require_once __DIR__ . '/../vendor/autoload.php';
+
+Loop::run(function () {
+    $client = new Client(Config::dsn('amqp://admin:admin123@172.23.0.2'));
+
+    yield $client->connect();
+
+    /** @var Channel $channel */
+    $channel = yield $client->channel();
+
+    yield $channel->queueDeclare('bench_queue');
+    yield $channel->exchangeDeclare('bench_exchange');
+    yield $channel->queueBind('bench_queue', 'bench_exchange');
+
+    $t = null;
+    $count = 0;
+
+    yield $channel->consume(function (Message $message) use (&$t, &$count, $client) {
+        if ($t === null) {
+            $t = \microtime(true);
+        }
+
+        if ($message->content() === 'quit') {
+            \printf('Pid: %s, Count: %s, Time: %.4f' . \PHP_EOL, \getmypid(), $count, \microtime(true) - $t);
+
+            yield $client->disconnect();
+        } else {
+            ++$count;
+        }
+
+    }, 'bench_queue', '', false, true);
+});

benchmark/producer.php

@@ -0,0 +1,51 @@
+<?php
+
+use Amp\Loop;
+use PHPinnacle\Ridge\Channel;
+use PHPinnacle\Ridge\Client;
+use PHPinnacle\Ridge\Config;
+
+require_once __DIR__ . '/../vendor/autoload.php';
+
+Loop::run(function () use ($argv) {
+    $client = new Client(Config::dsn('amqp://admin:admin123@172.23.0.2'));
+
+    yield $client->connect();
+
+    /** @var Channel $channel */
+    $channel = yield $client->channel();
+
+    yield $channel->queueDeclare('bench_queue');
+    yield $channel->exchangeDeclare('bench_exchange');
+    yield $channel->queueBind('bench_queue', 'bench_exchange');
+
+    $body = <<<EOT
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz
+abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyza
+EOT;
+
+    $time = \microtime(true);
+    $max  = isset($argv[1]) ? (int) $argv[1] : 1;
+
+    $promises = [];
+
+    for ($i = 0; $i < $max; $i++) {
+        $promises[] = $channel->publish($body, 'bench_exchange');
+    }
+
+    $promises[] = $channel->publish('quit', 'bench_exchange');
+
+    yield $promises;
+
+    yield $client->disconnect();
+
+    echo \microtime(true) - $time, \PHP_EOL;
+});

composer.json

@@ -17,7 +17,8 @@
     ],
     "require": {
         "php": "^7.2",
-        "amphp/amp": "^2.0"
+        "amphp/amp": "^2.0",
+        "amphp/socket": "^0.10.11"
     },
     "require-dev": {
         "phpunit/phpunit": "^6.0"