<?php
/* Suicidal Snail
 *
 *  @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/

/*  ---------------------------------------------------------------------
 * This is our subscriber
 * It connects to the publisher and subscribes to everything. It
 * sleeps for a short time between messages to simulate doing too
 * much work. If a message is more than 1 second late, it croaks.
 */
define("MAX_ALLOWED_DELAY", 100); // msecs

function subscriber()
{
    $context = new ZMQContext();

    // Subscribe to everything
    $subscriber = new ZMQSocket($context, ZMQ::SOCKET_SUB);
    $subscriber->connect("tcp://localhost:5556");
    $subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, "");

    //  Get and process messages
    while (true) {
        $clock = $subscriber->recv();
        //  Suicide snail logic
        if (microtime(true)*100 - $clock*100 > MAX_ALLOWED_DELAY) {
            echo "E: subscriber cannot keep up, aborting", PHP_EOL;
            break;
        }

        //  Work for 1 msec plus some random additional time
        usleep(1000 + rand(0, 1000));
    }
}

/* ---------------------------------------------------------------------
 * This is our server task
 * It publishes a time-stamped message to its pub socket every 1ms.
 */
function publisher()
{
    $context = new ZMQContext();

    //  Prepare publisher
    $publisher = new ZMQSocket($context, ZMQ::SOCKET_PUB);
    $publisher->bind("tcp://*:5556");

    while (true) {
        //  Send current clock (msecs) to subscribers
        $publisher->send(microtime(true));
        usleep(1000); //  1msec wait
    }
}

/*
 * This main thread simply starts a client, and a server, and then
 * waits for the client to croak.
 */
$pid = pcntl_fork();
if ($pid == 0) {
    publisher();
    exit();
}

$pid = pcntl_fork();
if ($pid == 0) {
    subscriber();
    exit();
}
