<?php
/**
 * 基于 ReactPHP 实现的 nio server
 */
define('DS', DIRECTORY_SEPARATOR);

require __DIR__ . DS . '..' . DS . 'vendor' . DS . 'autoload.php';

$host = '0.0.0.0';
$port = 8080;

$cartCheck = array();

$loop = React\EventLoop\Factory::create();
$socket = new React\Socket\Server("{$host}:{$port}", $loop);        // server
$connector = new React\Socket\Connector($loop);                         // client

/**
 * 阻塞方式请求其他服务
 *
 * @param $host
 * @param $port
 * @param $method
 * @param $productId
 * @return bool|mixed
 */
function checkClient($host, $port, $method, $productId)
{
    fwrite(STDOUT, 'request other client: ' . json_encode(func_get_args()) . "\n");

    $socket = @stream_socket_client("tcp://{$host}:{$port}", $errno, $errMsg);
    if ($socket === false) {
        throw new \RuntimeException("unable to create socket: " . $errMsg);
    }
    fwrite(STDOUT, "connect to server: [{$host}:{$port}]...\n");

    $message = json_encode([
        "method" => $method,
        "data" => array('productId' => $productId)
    ]);

    fwrite(STDOUT, "send to server: $message\n");
    $len = @fwrite($socket, $message);
    if ($len === 0) {
        fwrite(STDOUT, "socket closed\n");
        return false;
    }

    // 阻塞方式，等待响应，读取后返回
    $msg = @fread($socket, 4096);
    if ($msg) {
        fwrite(STDOUT, "receive server: $msg\n");
        // 获取到返回结果后，再返回
        return json_decode($msg, true);
    } elseif (feof($socket)) {
        fwrite(STDOUT, "socket closed\n");
        return false;
    }
    return true;
}

/**
 * @param $data
 * @param $method
 * @return bool|string
 */
function checkAllComplete($data, $method)
{
    global $cartCheck;
    $reProductId = $data['productId'];

    if (!isset($cartCheck[$reProductId][$method])) {
        $cartCheck[$reProductId][$method] = $data['re'];
    }

    // 齐活了
    $oneCartCheck = $cartCheck[$reProductId];
    if (isset($oneCartCheck['inventory']) && isset($oneCartCheck['product']) && isset($oneCartCheck['promo'])) {
        // 这里定义了一个cartCheck 全局变量，减产完毕，就会被设置为 true。三项为true就表示检查完毕
        $checkRe = $oneCartCheck['inventory'] && ($oneCartCheck['product']) && ($oneCartCheck['promo']);
        $reData = array('method' => 'cart', 'data' => array('product_id' => $reProductId), 're' => $checkRe, 'msg' => 'suc');
        $msg = json_encode($reData);
        return $msg;
    } else {
        return false;
    }
}

$socket->on('connection', function (React\Socket\ConnectionInterface $connection) use ($connector) {

    $connection->on('data', function ($data) use ($connection, $connector) {
        fwrite(STDOUT, "\nReactPHP server get data: {$data} , time " . date('Y-m-d H:i:s') . PHP_EOL);

        $json = json_decode($data, true);
        $method = $json["method"];
        $noBlocking = boolval($json['noBlocking']);
        $productId = $json['data']['productId'];

        if ($noBlocking) {
            // inventory
            $inventory = array('host' => '127.0.0.1', 'port' => '8081');
            $connector->connect("{$inventory['host']}:{$inventory['port']}")->then(function (React\Socket\ConnectionInterface $checkConnection) use ($productId, $connection) {
                $message = json_encode([
                    "method" => 'inventory',
                    "data" => array('productId' => $productId),
                ]);
                $checkConnection->write($message);
                fwrite(STDOUT, "request inventory server : {$message} , time " . date('Y-m-d H:i:s') . PHP_EOL);
                $checkConnection->on('data', function ($data) use ($productId, $connection, $checkConnection) {
                    fwrite(STDOUT, "received, data : {$data} , time " . date('Y-m-d H:i:s') . PHP_EOL);
                    $json = json_decode($data, true);
                    $data = $json['data'];

                    $re = checkAllComplete($data, 'inventory');
                    if ($re) {
                        $connection->write($re);
                        fwrite(STDOUT, "complete response, $re time " . date('Y-m-d H:i:s') . "\n");
//                    $connection->close();
                    }
                    $checkConnection->close();
                });
            });

            $product = array('host' => '127.0.0.1', 'port' => '8082');
            $connector->connect("{$product['host']}:{$product['port']}")->then(function (React\Socket\ConnectionInterface $checkConnection) use ($productId, $connection) {
                $message = json_encode([
                    "method" => 'product',
                    "data" => array('productId' => $productId),
                ]);
                $checkConnection->write($message);
                fwrite(STDOUT, "request inventory server : {$message} , time " . date('Y-m-d H:i:s') . PHP_EOL);
                $checkConnection->on('data', function ($data) use ($productId, $connection, $checkConnection) {
                    fwrite(STDOUT, "received, data : {$data} , time " . date('Y-m-d H:i:s') . PHP_EOL);
                    $json = json_decode($data, true);
                    $data = $json['data'];

                    $re = checkAllComplete($data, 'product');
                    if ($re) {
                        $connection->write($re);
                        fwrite(STDOUT, "complete response, $re time " . date('Y-m-d H:i:s') . "\n");
//                    $connection->close();
                    }
                    $checkConnection->end();
                });
            });

            $promo = array('host' => '127.0.0.1', 'port' => '8083');
            $connector->connect("{$promo['host']}:{$promo['port']}")->then(function (React\Socket\ConnectionInterface $checkConnection) use ($productId, $connection) {
                $message = json_encode([
                    "method" => 'promo',
                    "data" => array('productId' => $productId),
                ]);
                $checkConnection->write($message);
                fwrite(STDOUT, "request inventory server : {$message} , time " . date('Y-m-d H:i:s') . PHP_EOL);
                $checkConnection->on('data', function ($data) use ($productId, $connection, $checkConnection) {
                    fwrite(STDOUT, "received, data : {$data} , time " . date('Y-m-d H:i:s') . PHP_EOL);
                    $json = json_decode($data, true);
                    $data = $json['data'];

                    $re = checkAllComplete($data, 'promo');
                    if ($re) {
                        $connection->write($re);
                        fwrite(STDOUT, "complete response, $re time " . date('Y-m-d H:i:s') . "\n");
//                    $connection->close();
                    }
                    $checkConnection->end();
                });
            });
        } else {
            // 阻塞方式
            $checkAllSuc = false;
            // 依次发起三个检查请求
            $requestInventorySuc = checkClient('127.0.0.1', '8081', 'inventory', $productId);
            if ($requestInventorySuc) {
                if ($requestInventorySuc['data']['re']) {

                    $requestProductSuc = checkClient('127.0.0.1', '8082', 'product', $productId);
                    if ($requestProductSuc) {
                        if ($requestProductSuc['data']['re']) {

                            $requestPromoSuc = checkClient('127.0.0.1', '8083', 'promo', $productId);
                            if (isset($requestPromoSuc['data']['re'])) {
                                $checkAllSuc = true;
                            }
                        }
                    }
                }
                // 如果检查库存失败，后面流程就不用继续了
            }

            $reMsg = array('method' => 'cart', 'data' => array('product_id' => $productId), 're' => $checkAllSuc, 'msg' => 'suc');
            $data = json_encode($reMsg);
            // 这里响应，发起 加购 请求的客户端，
            $connection->write($data);
            fwrite(STDOUT, "complete response, $data time " . date('Y-m-d H:i:s') . "\n");
        }

    });
});

$loop->run();