zmq socket example

gistfile1.php

<?php

// Wrapper class for ZMQ PECL extension

class ZMQ_Socket {

    private $_context;
    private $_socket;
    private $_signer;
    private $_dsn;

    private $_connected = false;
    
    function __construct($opts = null) {

    }

    function connect($type=null) {
        if ($this->_connected) {
            // Already connected
            return;
        }

        $this->_dsn = "tcp://127.0.0.1:5555";

        if (!$this->_context) {
            $this->_context = new ZMQContext(1, false);
        }

        if (!$this->_signer){
            $this->_signer = new URLSafeTimedSerializer(SIG_SECRET, 'ffxiah');
        }

        $this->_socket = new ZMQSocket($this->_context, $type);
        $this->_socket->connect($this->_dsn);

        $this->_connected = true;

    }

    // Lazy connect prevents errors when forking
    function send($data) {

        $this->connect(ZMQ::SOCKET_PUSH);

        if (is_array($data)) {
            $data = json_encode($data);
        }
        
        if ($this->_signer){
            $data = $this->_signer->dumps($data);
        }

        $this->_socket->send($data, ZMQ::MODE_NOBLOCK);
    }

    function recv(){
        $this->connect(ZMQ::SOCKET_PULL);

        $data = $this->_socket->recv();

        if ($this->_signer){
            $data = $this->_signer->loads($data);
        }

        return (Core::isJSON($data)) ? json_decode($data, true) : $data;
    }

    function disconnect(){
        $this->_socket->disconnect($this->_dsn);
    }
}

$pull = new ZMQ_Socket();
$pull->connect(ZMQ::SOCKET_PULL);

$push = new ZMQ_Socket();
$push->send('blah');
$push->disconnect();


echo $pull->recv();
$pull->disconnect();