Consumer and producer files for blog post : http://bit.ly/14SaWpp

rpc-consumer.php

<?php
require_once __DIR__ . '/../vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPConnection;
use PhpAmqpLib\Message\AMQPMessage;


define('AMQP_DEBUG_FM', true);
$host = '127.0.0.1';
//$host = '192.168.2.3';

$connection = new AMQPConnection($host , 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('rpc_queue', false, false, false, false);

function fib($n) {
//    if ($n == 0)
//        return 0;
//    if ($n == 1)
//        return 1;
//    return fib($n-1) + fib($n-2);
return $n;
}


echo " [x] Awaiting RPC requests\n";
$callback = function($req) {
    $n = intval($req->body);
    echo " [.] fib(", $n, ")\n";

    $msg = new AMQPMessage(
        (string) fib($n),
        array('correlation_id' => $req->get('correlation_id'))
    );

    $req->delivery_info['channel']->basic_publish(
        $msg, '', $req->get('reply_to'));
    $req->delivery_info['channel']->basic_ack(
        $req->delivery_info['delivery_tag']);
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume('rpc_queue', '', false, false, false, false, $callback);

while(count($channel->callbacks)) {
    $channel->wait();
}

$channel->close();
$connection->close();

?>

rpc-producer.php

<?php

require_once __DIR__ . '/../vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPConnection;
use PhpAmqpLib\Message\AMQPMessage;

class FibonacciRpcClient
{
    private $connection;
    private $channel;
    private $callback_queue;
    private $response;
    private $corr_id;

    public function __construct()
    {
        define('AMQP_DEBUG_FM', true);
        $host = '127.0.0.1';
//        $host = '192.168.2.3';

        $this->connection = new AMQPConnection(
            $host , 5672, 'guest', 'guest');
        $this->channel    = $this->connection->channel();
        list($this->callback_queue, ,) = $this->channel->queue_declare(
            "",
            false,
            false,
            true,
            false
        );
        $this->channel->basic_consume(
            $this->callback_queue,
            '',
            false,
            false,
            false,
            false,
            array($this, 'on_response')
        );
    }

    public function on_response($rep)
    {
        if ($rep->get('correlation_id') == $this->corr_id) {
            $this->response = $rep->body;
        }
    }

    public function call($n)
    {
        $this->response = null;
        $this->corr_id  = uniqid();

        $msg = new AMQPMessage(
            (string) $n,
            array(
                 'correlation_id' => $this->corr_id,
                 'reply_to'       => $this->callback_queue
            )
        );
        $this->channel->basic_publish($msg, '', 'rpc_queue');
        while (!$this->response) {
            $this->channel->wait();
        }

        return intval($this->response);
    }
}



$fibonacci_rpc = new FibonacciRpcClient();

for ($i = 0; $i <= 20000; $i++) {

    $response = $fibonacci_rpc->call(30);
    echo " [.] Got ", $response, "\n";
}


?>