For this, the checkout service will offer a simple REP ZeroMQ socket (or a ROUTER socket, in a concurrent setup). After receiving a checkout order, the checkout service will then communicate with the inventory service to check if the required items are available and to reduce the stock amount by the item amounts in the cart. If that was successful, it will publish the checkout order on a PUB socket that other services can listen on.

If a cart consists of multiple items, the checkout service will need to make multiple calls to the inventory service. In this example, you will learn how to make multiple requests in parallel in order to speed up execution. We will also use the react/zmq library, which offers an asynchronous interface for the ZeroMQ library and the react/promise library that will help you to better handle an asynchronous application.

Start by creating a new composer.json file in a new checkout/ directory and initialize the project with composer install:

{ 
  "name": "packt-php7/chp7-checkout", 
  "type": "project", 
  "authors": [{ 
    "name": "Martin Helmich", 
    "email": "[email protected]" 
  }], 
  "require": { 
    "php": ">= 7.0", 
    "react/zmq": "^0.3.0",

    "react/promise": "^2.2", 
    "ext-zmq": "*", 
    "ext-ev": "*" 
  }, 
  "autoload": { 
    "psr-4": { 
      "Packt\Chp7\Checkout": "src/" 
    } 
  } 

This file is similar to the inventory service's composer.json; the only difference is the PSR-4 namespace and the additional requirements react/zmq, react/promise, and ext-ev. If you are using Docker for your development setup, you can simply copy your existing Dockerfile from the inventory service.

Continue by creating a server.json file in your checkout/ directory. As with any React application (remember the Ratchet application from Chapter 6, Building a Chat Application), the first thing you need to do is to create an event loop that you can then run:

<?php 
use ReactMQFactory; 
use ReactMQContext; 
 
require 'vendor/autoload.php'; 
 
$loop = Factory::create(); 
$ctx  = new Context($loop); 
 
$loop->run(); 

Note that we're using the ReactMQContext class instead of the ZMQContext class now. The React context class offers the same interface, but extends its base class by some functionalities to better support asynchronous programming.

You can already start this program and it will run infinitely, but it will not actually do anything just yet. As the checkout service should offer a REP socket to which clients should send requests, you should continue by creating and binding a new REP socket before running the event loop:

// ... 
$ctx = new Context($loop); 
 
$socket = $ctx->getSocket(ZMQ::SOCKET_REP);

$socket->bind('tcp://0.0.0.0:5557'); 
 
$loop->run(); 

ReactPHP applications are asynchronous; instead of just calling recv() on the socket to wait for the next incoming message, you can now register an event handler on the socket that will be called by ReactPHP's event loop as soon as a message is received:

// ... 
 
$socket = $ctx->getSocket(ZMQ::SOCKET_REP); 
$socket->bind('tcp://0.0.0.0:5557'); 
$socket->on('message', function(string $msg) use ($socket) {

    echo "received message $msg.
";

    $socket->send('Response text');

}); 
 
$loop->run(); 

This callback solution works similar to other asynchronous libraries that you will most commonly encounter when developing client-site JavaScript code. The basic principle is the same: the $socket->on(...) method simply registers an event listener that can be called at any later point in time whenever a new message is received. The execution of the code will continue immediately (in contrast to this, compare the regular $socket->recv() function that blocks until a new message is received) and the $loop->run() method is called. This call starts the actual event loop that is responsible for calling the registered event listener when new messages are received. The event loop will block until it is interrupted (for example, by a SIGINT signal that you can trigger with Ctrl + C on the command line).

When working with asynchronous code, it is often just a matter of time until you find yourself in "callback hell". Imagine you want to send two consecutive ZeroMQ requests (for example, first asking the inventory service if a given article is available and then actually instructing the inventory service to reduce the stock by the required amount). You can implement this using multiple sockets and the 'message' event that you have seen previously. However, this will quickly become an unmaintainable mess:

$socket->on('message', function(string $msg) use ($socket, $ctx) { 
    $check = $ctx->getSocket(ZMQ::SOCKET_REQ); 
    $check->connect('tcp://identity:5557'); 
    $check->send(/* checkArticle JSON-RPC here */); 
    $check->on('message', function(string $msg) use ($socket, $ctx) { 
        $take = $ctx->getSocket(ZMQ::SOCKET_REQ); 
        $take->connect('tcp://identity:5557'); 
        $take->send(/* takeArticle JSON-RPC here */); 
        $take->on('message', function(string $msg) use ($socket) { 
            $socket->send('success'); 
        }); 
    }); 
}); 

The preceding code snippet is just an example of how complicated this might get; in our case, you would even need to consider that each checkout order can contain any number of articles, each of them requiring two new requests to the identity service.

To make life better, you can implement this functionality using promises (see the following box for a detailed explanation of the concept). A good implementation of promises is provided by the react/promise library that should already be declared in your composer.json file.

We can now put this principle to use by writing an asynchronous client class for communicating with our inventory service. As that service communicates using JSON-RPC, we will now implement the PacktChp7CheckoutJsonRpcClient class. This class is initialized with a ZeroMQ context, and for convenience, also the remote service's URL:

namespace PacktChp7Checkout; 
 
use ReactPromisePromiseInterface; 
use ReactMQContext; 
 
class JsonRpcClient 
{ 
    private $context; 
    private $url; 
 
    public function __construct(Context $context, string $url) 
    { 
        $this->context = $context; 
        $this->url     = $url; 
    } 
 
    public function request(string $method, array $params = []): PromiseInterface 
    { 
    } 
} 

In this example, the class already contains a request method that accepts a method name and a set of parameters, and should return an implementation of ReactPromisePromiseInterface.

In the request() method, you can now open a new REQ socket and send a JSON-RPC request to it:

public function request(string $method, array $params = []): PromiseInterface 
{ 
    $body = json_encode([

        'jsonrpc' => '2.0',

        'method'  => $method,

        'params'  => $params,

    ]);

    $sock = $this->context->getSocket(MQ::SOCKET_REQ);

    $sock->connect($this->url);

    $sock->send($body); 
} 

Since the request() method is supposed to work asynchronously, you cannot simply call the recv() method and block until a result is received. Instead, we will need to return a promise for the response value that can be resolved later, whenever a response message is received on the REQ socket. For this, you can use the ReactPromiseDeferred class:

$body = json_encode([ 
    'jsonrpc' => '2.0', 
    'method'  => $method, 
    'params'  => $params, 
]); 
$deferred = new Deferred(); 
 
$sock = $this->context->getSocket(MQ::SOCKET_REQ); 
$sock->connect($this->url); 
$sock->on('message', function(string $response) use ($deferred) {

    $deferred->resolve($response);

}); 
$sock->send($body); 
 
return $deferred->promise();

This is a prime example of how promises work: you can use the Deferred class to create and return a promise for a value that is not yet available. Remember: the function passed into the $sock->on(...) method will not be called immediately, but at any later point in time when a response was actually received. As soon as this event occurs, the promise that was returned by the request function is resolved with the actual response value.

As the response message contains a JSON-RPC response, you need to evaluate this response before fulfilling the promise that you made to the caller of the request function. As a JSON-RPC response can also contain an error, it is worth noting that you can also reject a promise (for example, when an error occurred while waiting for the response):

$sock->on('message', function(string $response) use ($deferred) { 
    $response = json_decode($response);

    if (isset($response->result)) {

        $deferred->resolve($response->result);

    } elseif (isset($response->error)) {

        $deferred->reject(new Exception(

            $response->error->message,

            $response->error->code

        );

    } else {

        $deferred->reject(new Exception('invalid response'));

    } 
}); 

You can now use this JSON-RPC client class in your server.php to actually communicate with the inventory service on each incoming checkout request. Let's start with a simple example on how you can use the new class to chain two consecutive JSON-RPC calls together:

$client = new JsonRpcClient($ctx, 'tcp://inventory:5557'); 
$client->request('checkArticle', [1000]) 
    ->then(function(bool $ok) use ($client) { 
        if ($ok) { 
            return $client->request('takeArticle', [1000]); 
        } else { 
            throw new Exception("Article is not available"); 
        } 
    }) 
    ->then(function(bool $ok) { 
        if ($ok) { 
            echo "Successfully took 1 item of article 1000"; 
        } 
    }, function(Exception $error) { 
        echo "An error occurred: ${error->getMessage()}
"; 
    }); 

As you can see, the PromiseInterface's then function accepts two parameters (each both a new function): the first function will be called as soon as the promise was resolved with an actual value; the second function will be called in case the promise was rejected.

If a function passed to then(...) returns a new value, the then function will return a new promise for this value. An exception to this rule is when the callback function returns a new promise itself (in our case, in which $client->request is called again within the then() callback). In this case, the returned promise replaces the original promise. This means that chained calls to the then() function actually listen on the second promise.

Let's put this to use in the server.php file. In contrast to the preceding example, you need to consider that each checkout order may contain multiple articles. This means that you will need to execute multiple checkArticle requests to the inventory service:

$client = new JsonRpcClient($ctx, 'tcp://inventory:5557'); 
$socket->on('message', function(string $msg) use ($socket, $client) { 
    $request = json_decode($msg);

    $promises = [];

    foreach ($request->cart as $article) {

        $promises[] = $client->request('checkArticle', [$article->articlenumber, $article->amount]); 
    } 
}); 

In this example, we assume that incoming checkout orders are JSON encoded messages that look like the following example:

{ 
  "cart": [ 
    "articlenumber": 1000, 
    "amount": 2 
  ] 
} 

In the current version of our server.php, we call the JSON-RPC client multiple times and collect the returned promises in an array. However, we do not actually do anything with them yet. You could now call the then() function on each of these promises with a callback that will be called for each article with a boolean parameter indicating whether this one article is available. However, for processing the order correctly, we need to know if all articles from the checkout order are available. So what you need to do is not to wait on each promise separately, but to wait until all of them are completed. This is what the ReactPromiseall function is for: this function takes a list of promises as parameters and returns a new promise that is fulfilled as soon as all supplied promises are fulfilled:

$request = json_decode($msg); 
$promises = []; 
 
foreach ($request->cart as $article) { 
    $promises[] = $client->request('checkArticle', [$article->articlenumber, $article->amount]); 
} 
 
ReactPromiseall($promises)->then(function(array $values) use ($socket) {

    if (array_sum($values) == count($values)) {

        echo "all required articles are available";

    } else {

        $socket->send(json_encode([

            'error' => 'not all required articles are available'

        ]);

    }
});

If not all required articles are available in the inventory service, you can answer the request early with an error message, as there is no need to continue any further. If all articles are available, you'll need a set of subsequent requests to actually reduce the inventory by the specified amounts.

In the next step, you can now extend your server to run the second set of requests to the inventory service after all of the checkArticle method calls have successfully returned. This can be done by following the same way as before using the ReactPromiseall method:

ReactPromiseall($promises)->then(function(array $values) use ($socket, $request) { 
    $promises = [];

    if (array_sum($values) == count($values)) {

        foreach ($request->cart as $article) {

            $promises[] = $client->request('takeArticle', [$article->articlenumber, $article->amount]);

        }

        ReactPromiseall($promises)->then(function() use ($socket) {

            $socket->send(json_encode([

                'result' => true

            ]);

        } 
    } else { 
        $socket->send(json_encode([ 
            'error' => 'not all required articles are available' 
        ]); 
    } 
}); 

In order to actually test this new server, let's write a short test script that tries to execute an example checkout order. For this, create a new client.php file in your checkout/ directory:

$ctx  = new ZMQContext(); 
$sock = $ctx->getSocket(ZMQ::SOCKET_REQ); 
$sock->connect('tcp://checkout:5557'); 
$sock->send(json_encode([ 
    'cart' => [ 
        ['articlenumber' => 1000, 'amount' => 3], 
        ['articlenumber' => 1001, 'amount' => 2] 
    ] 
])); 
 
$result = $sock->recv(); 
var_dump($result); 

To run both the checkout service and the test script, you can extend your docker-compose.yml file in your project's root directory with the new checkout service:

checkout:

  build: checkout

  volumes:

    - checkout:/usr/src/app

  links:

    - inventory:inventory 
inventory: 
  build: inventory 
  ports: 
    - 5557 
  volumes: 
    - inventory:/usr/src/app 

For the test script, add a second Compose configuration file, docker-compose.testing.yml:

test: 
  build: checkout 
  command: php client.php 
  volumes: 
    - checkout:/usr/src/app 
  links: 
    - checkout:checkout 

Afterwards, you can test your checkout service using the following command line commands:

$ docker-compose up -d 
$ docker-compose -f docker-compose.testing.yml run --rm test

The following screenshot shows an example output of both the test script and both server scripts (in this example, some additional echo statements have been added to make the server more verbose):

An example output of a checkout order being processed by the checkout and inventory services