forked from rabbitmq/rabbitmq-tutorials
-
Notifications
You must be signed in to change notification settings - Fork 0
/
receive_log_topic.php
50 lines (41 loc) · 1.32 KB
/
receive_log_topic.php
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
<?php
//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);
$channel->setPrefetchCount(1);
//Declare Exchange
$exchange_name = 'topic_logs';
$exchange = new AMQPExchange($channel);
$exchange->setType(AMQP_EX_TYPE_TOPIC);
$exchange->setName($exchange_name);
$exchange->declareExchange();
$binding_keys = array_slice($argv, 1); //accept an array of inputs delimited by space
if(empty($binding_keys)) {
file_put_contents('php://stderr', "Usage: {$argv[0]} [binding_key]...\n");
exit(1);
}
//Declare Queue
$queue = new AMQPQueue($channel);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declareQueue();
foreach($binding_keys as $binding_key) {
$queue->bind($exchange_name, $binding_key);
}
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;
//echo sprintf("Delivery Tag: %s",$message->getDeliveryTag()), PHP_EOL;
$q->nack($message->getDeliveryTag());
};
try {
$queue->consume($callback_func);
} catch(AMQPQueueException $ex) {
print_r($ex);
} catch(Exception $ex) {
print_r($ex);
}