echo/rabbitmq-tutorials
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added example 4 emit/receive direct

Browse Source
This commit is contained in:
Chimdi Azubuike 2015-06-20 22:43:34 -07:00
parent 3b8dd44363
commit fe6a9317db
2 changed files with 102 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
php-amqp/emit_log_direct.php Normal file
View File

@ -0,0 +1,46 @@
#!/usr/bin/env php
<?php
/**
emit_log_direct.php
@author: Chimdi Azubuike <me@chimdi.com>
*/
//Establish connection to AMQP
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('guest');
$connection->setPassword('guest');
$connection->connect();
//Declare Channel
$channel = new AMQPChannel($connection);
$routing_key = $severity = $argv[1];
if(empty($severity)) $severity = 'info';
$message = implode(' ',array_slice($argv, 2));
if(empty($message)) $message = 'Hello World!';
try {
//Declare Exchange
$exchange_name = 'direct_logs';
$exchange = new AMQPExchange($channel);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setName($exchange_name);
$exchange->declareExchange();
$queue = new AMQPQueue($channel);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->setName('monitor.1');
$queue->declareQueue();
$queue->bind($exchange_name, $routing_key);
$exchange->publish($message,$routing_key);
echo " [x] Sent {$severity}:{$message}", PHP_EOL;
} catch(Exception $ex) {
print_r($ex);
}
$connection->disconnect();

56
php-amqp/receive_logs_direct.php Normal file
View File

@ -0,0 +1,56 @@
#!/usr/bin/env php
<?php
/**
receive_logs_direct.php
@author: Chimdi Azubuike <me@chimdi.com>
*/
//Establish Connection
$connection = new AMQPConnection();
$connection->setHost('127.0.0.1');
$connection->setLogin('guest');
$connection->setPassword('guest');
$connection->connect();
//Listen on Channel
$channel = new AMQPChannel($connection);
echo " [*] Waiting for logs. To exit press CTRL+C", PHP_EOL;
$callback_func = function(AMQPEnvelope $message, AMQPQueue $q) {
echo sprintf(" [X] [%s] %s",$message->getRoutingKey(),$message->getBody()), PHP_EOL;
$q->nack($message->getDeliveryTag());
return true;
};
$severities = array_slice($argv,1);
if(empty($severities)) {
file_put_contents('php://stderr', "Usage: {$argv[0]} [info] [warning] [error]\n");
exit(1);
}
try {
//Declare Exchange
$exchange_name = 'direct_logs';
$exchange = new AMQPExchange($channel);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setName($exchange_name);
$exchange->declareExchange();
//Declare Queue
$queue = new AMQPQueue($channel);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declareQueue();
foreach($severities as $routing_key) {
$queue->bind($exchange_name, $routing_key);
}
$queue->consume($callback_func);
} catch(AMQPQueueException $ex) {
print_r($ex);
} catch(Exception $ex) {
print_r($ex);
}