php-enqueue · makasim · Aug 3, 2017 · Aug 3, 2017 · Aug 3, 2017
diff --git a/pkg/amqp-lib/tutorial/emit_log.php b/pkg/amqp-lib/tutorial/emit_log.php
@@ -0,0 +1,33 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpTopic;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$topic = $context->createTopic('logs');
+$topic->setType(AmqpTopic::TYPE_FANOUT);
+
+$context->declareTopic($topic);
+
+$data = implode(' ', array_slice($argv, 1));
+if (empty($data)) {
+    $data = 'info: Hello World!';
+}
+$message = $context->createMessage($data);
+
+$context->createProducer()->send($topic, $message);
+
+echo ' [x] Sent ', $data, "\n";
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/emit_log_direct.php b/pkg/amqp-lib/tutorial/emit_log_direct.php
@@ -0,0 +1,37 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpTopic;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$topic = $context->createTopic('direct_logs');
+$topic->setType(AmqpTopic::TYPE_DIRECT);
+
+$context->declareTopic($topic);
+
+$severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';
+
+$data = implode(' ', array_slice($argv, 2));
+if (empty($data)) {
+    $data = 'Hello World!';
+}
+
+$message = $context->createMessage($data);
+$message->setRoutingKey($severity);
+
+$context->createProducer()->send($topic, $message);
+
+echo ' [x] Sent ',$severity,':',$data," \n";
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/emit_log_topic.php b/pkg/amqp-lib/tutorial/emit_log_topic.php
@@ -0,0 +1,37 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpTopic;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$topic = $context->createTopic('topic_logs');
+$topic->setType(AmqpTopic::TYPE_TOPIC);
+
+$context->declareTopic($topic);
+
+$routing_key = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'anonymous.info';
+
+$data = implode(' ', array_slice($argv, 2));
+if (empty($data)) {
+    $data = 'Hello World!';
+}
+
+$message = $context->createMessage($data);
+$message->setRoutingKey($routing_key);
+
+$context->createProducer()->send($topic, $message);
+
+echo ' [x] Sent ',$routing_key,':',$data," \n";
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/new_task.php b/pkg/amqp-lib/tutorial/new_task.php
@@ -0,0 +1,35 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpMessage;
+use Interop\Amqp\AmqpQueue;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$queue = $context->createQueue('task_queue');
+$queue->addFlag(AmqpQueue::FLAG_DURABLE);
+
+$context->declareQueue($queue);
+
+$data = implode(' ', array_slice($argv, 1));
+if (empty($data)) {
+    $data = 'Hello World!';
+}
+$message = $context->createMessage($data);
+$message->setDeliveryMode(AmqpMessage::DELIVERY_MODE_PERSISTENT);
+
+$context->createProducer()->send($queue, $message);
+
+echo ' [x] Sent ', $data, "\n";
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/receive.php b/pkg/amqp-lib/tutorial/receive.php
@@ -0,0 +1,33 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpConsumer;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+    'receive_method' => 'basic_consume',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$queue = $context->createQueue('hello');
+$context->declareQueue($queue);
+
+$consumer = $context->createConsumer($queue);
+$consumer->addFlag(AmqpConsumer::FLAG_NOACK);
+
+echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";
+
+while (true) {
+    if ($message = $consumer->receive()) {
+        echo ' [x] Received ', $message->getBody(), "\n";
+    }
+}
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/receive_logs.php b/pkg/amqp-lib/tutorial/receive_logs.php
@@ -0,0 +1,41 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpTopic;
+use Interop\Amqp\Impl\AmqpBind;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+    'receive_method' => 'basic_consume',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$topic = $context->createTopic('logs');
+$topic->setType(AmqpTopic::TYPE_FANOUT);
+
+$context->declareTopic($topic);
+
+$queue = $context->createTemporaryQueue();
+
+$context->bind(new AmqpBind($topic, $queue));
+
+$consumer = $context->createConsumer($queue);
+$consumer->addFlag(AmqpConsumer::FLAG_NOACK);
+
+echo ' [*] Waiting for logs. To exit press CTRL+C', "\n";
+
+while (true) {
+    if ($message = $consumer->receive()) {
+        echo ' [x] ', $message->getBody(), "\n";
+    }
+}
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/receive_logs_direct.php b/pkg/amqp-lib/tutorial/receive_logs_direct.php
@@ -0,0 +1,49 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpTopic;
+use Interop\Amqp\Impl\AmqpBind;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+    'receive_method' => 'basic_consume',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$topic = $context->createTopic('direct_logs');
+$topic->setType(AmqpTopic::TYPE_DIRECT);
+
+$context->declareTopic($topic);
+
+$queue = $context->createTemporaryQueue();
+
+$severities = array_slice($argv, 1);
+if (empty($severities)) {
+    file_put_contents('php://stderr', "Usage: $argv[0] [info] [warning] [error]\n");
+    exit(1);
+}
+
+foreach ($severities as $severity) {
+    $context->bind(new AmqpBind($topic, $queue, $severity));
+}
+
+$consumer = $context->createConsumer($queue);
+$consumer->addFlag(AmqpConsumer::FLAG_NOACK);
+
+echo ' [*] Waiting for logs. To exit press CTRL+C', "\n";
+
+while (true) {
+    if ($message = $consumer->receive()) {
+        echo ' [x] '.$message->getRoutingKey().':'.$message->getBody()."\n";
+    }
+}
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/receive_logs_topic.php b/pkg/amqp-lib/tutorial/receive_logs_topic.php
@@ -0,0 +1,49 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+use Interop\Amqp\AmqpConsumer;
+use Interop\Amqp\AmqpTopic;
+use Interop\Amqp\Impl\AmqpBind;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+    'receive_method' => 'basic_consume',
+];
+
+$connection = new AmqpConnectionFactory($config);
+$context = $connection->createContext();
+
+$topic = $context->createTopic('topic_logs');
+$topic->setType(AmqpTopic::TYPE_TOPIC);
+
+$context->declareTopic($topic);
+
+$queue = $context->createTemporaryQueue();
+
+$binding_keys = array_slice($argv, 1);
+if (empty($binding_keys)) {
+    file_put_contents('php://stderr', "Usage: $argv[0] [binding_key]\n");
+    exit(1);
+}
+
+foreach ($binding_keys as $binding_key) {
+    $context->bind(new AmqpBind($topic, $queue, $binding_key));
+}
+
+$consumer = $context->createConsumer($queue);
+$consumer->addFlag(AmqpConsumer::FLAG_NOACK);
+
+echo ' [*] Waiting for logs. To exit press CTRL+C', "\n";
+
+while (true) {
+    if ($message = $consumer->receive()) {
+        echo ' [x] '.$message->getRoutingKey().':'.$message->getBody()."\n";
+    }
+}
+
+$context->close();
diff --git a/pkg/amqp-lib/tutorial/rpc_client.php b/pkg/amqp-lib/tutorial/rpc_client.php
@@ -0,0 +1,56 @@
+<?php
+
+require_once __DIR__.'/vendor/autoload.php';
+
+use Enqueue\AmqpLib\AmqpConnectionFactory;
+
+$config = [
+    'host' => 'localhost',
+    'port' => 5672,
+    'user' => 'guest',
+    'pass' => 'guest',
+    'receive_method' => 'basic_consume',
+];
+
+class FibonacciRpcClient
+{
+    /** @var \Interop\Amqp\AmqpContext */
+    private $context;
+
+    /** @var \Interop\Amqp\AmqpQueue */
+    private $callback_queue;
+
+    public function __construct(array $config)
+    {
+        $this->context = (new AmqpConnectionFactory($config))->createContext();
+        $this->callback_queue = $this->context->createTemporaryQueue();
+    }
+
+    public function call($n)
+    {
+        $corr_id = uniqid();
+
+        $message = $this->context->createMessage((string) $n);
+        $message->setCorrelationId($corr_id);
+        $message->setReplyTo($this->callback_queue->getQueueName());
+
+        $this->context->createProducer()->send(
+            $this->context->createQueue('rpc_queue'),
+            $message
+        );
+
+        $consumer = $this->context->createConsumer($this->callback_queue);
+
+        while (true) {
+            if ($message = $consumer->receive()) {
+                if ($message->getCorrelationId() == $corr_id) {
+                    return (int) ($message->getBody());
+                }
+            }
+        }
+    }
+}
+
+$fibonacci_rpc = new FibonacciRpcClient($config);
+$response = $fibonacci_rpc->call(30);
+echo ' [.] Got ', $response, "\n";