worker

$context = new ZMQContext();
// Socket for work input
$work = new ZMQSocket($context, ZMQ::SOCKET_PULL);
$work->setSockOpt(ZMQ::SOCKOPT_HWM, 1);
$work->connect('tcp://localhost:5566');

// Socket for results output
$sink = new ZMQSocket($context, ZMQ::SOCKET_PUSH);
$sink->connect('tcp://localhost:5567');

// Socket for control input
$controller = new ZMQSocket($context, ZMQ::SOCKET_SUB);
$controller->connect('tcp://localhost:5568');
$controller->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, "");

$poll = new ZMQPoll();
$poll->add($work, ZMQ::POLL_IN);
$poll->add($controller, ZMQ::POLL_IN);
$read = $write = array();

$writer = new Zend_Log_Writer_Stream('/home/DMitchell/tmp/worker.log');
$logger = new Zend_Log($writer);
$message = 'A worker starting to log.';
$logger->info($message);

while(true) {
    $events = $poll->poll($read, $write, 5000);
    if($events) {
        $message = 'A worker received an event!';
        $logger->info($message);
        foreach ($read as $socket) {
            if($socket === $work) {
                $json = $work->recv();
                $finalArray = json_decode($json);
                foreach ($finalArray as $keyString) {
                    // DO THE WORK
                }
                $results = json_encode($timeArray);
                $sink->send($results);
            } elseif ($socket === $controller) {
                exit();
            }
        }
    }
}