src.dualinventive.com/dinet/common/ditest-php/rpc/PairSocket.php

69 lines
1.5 KiB
PHP

<?php
require_once(__DIR__ . '/RpcCommon.php');
if (!isset($g_zmqcontext)) {
$g_zmqcontext = new ZMQContext();
}
class PairSocket extends RpcCommon {
private $__socket;
function __construct(&$test, $uri) {
parent::__construct($test);
global $g_zmqcontext;
$this->__socket = $g_zmqcontext->getSocket(ZMQ::SOCKET_PAIR);
$this->__socket->connect($uri);
}
/**
* Expect no received messages
*
* @param for how many millis no messages must be received
*/
function recvNone($for) {
$endTime = (microtime(true) * 1000) + $for;
while (true) {
$r = $this->__socket->recv(ZMQ::MODE_NOBLOCK);
$this->_test->isFalse($r);
if ($r) {
$this->_test->failure("Got message while expecting none");
return $r;
}
if ($endTime < (microtime(true) * 1000))
break;
usleep(1000);
}
}
function recv($classMethod, $params = [], $autoPair = false) {
while (true) {
$r = $this->__socket->recv();
if (!$this->__socket->getSockOpt(ZMQ::SOCKOPT_RCVMORE))
break;
}
$msg = $this->validateRpc($r, $classMethod);
if (count($params))
$this->_test->equal($params, $msg['params']);
return $msg;
}
/**
* ..
*/
function send($data, $result = null) {
if (is_string($data)) {
$ret = $this->__socket->send($data);
} elseif (is_array($data)) {
$ret = $this->__socket->send(json_encode($data));
} else {
$this->_test->failure("RPC message is not a valid reply");
var_dump($data);
}
$this->_test->equal($this->__socket, $ret);
}
}