zbus-php-client

function biz($broker){ 
    $producer = new Producer($broker);
    $msg = new Message();
    $msg->url = '/abc';
    $msg->topic = 'MyTopic';
    $msg->tag = 'Stock.A.中文';
    $msg->body = 'From PHP sync 中文';
    
    $res = $producer->publish($msg);
    echo $res . PHP_EOL;
}


$loop = new EventLoop(); 
$broker = new Broker($loop, "localhost:15555;localhost:15556", true); // включить синхронный режим

$broker->on('ready', function() use($loop, $broker){  
    //запустить после готовности
    try {  biz($broker); } catch (Exception $e){ echo $e->getMessage() . PHP_EOL; }
    
    $broker->close();  
    $loop->stop(); //остановить в любом случае
}); 

$loop->runOnce();

$messageHandler = function($msg, $client){ //на что следует обратить внимание
    echo $msg->tag . PHP_EOL;
    echo $msg . PHP_EOL;
};


Logger::$Level = Logger::DEBUG; //изменить на info, чтобы отключить подробные сообщения
$loop = new EventLoop(); 
$broker = new Broker($loop, "localhost:15555;localhost:15556"); //HA, проверить?!
 

$ctrl = new ControlHeaders();
$ctrl->topic = "MyTopic"; 
$ctrl->topic_mask = Protocol::MASK_DISK;
//$ctrl->group_name_auto = true;
//$ctrl->consume_group = "MyTopic_Group12";
//$ctrl->group_mask = Protocol::MASK_DISK;
//$ctrl->group_filter = "abc.*";  


$c = new Consumer($broker, $ctrl);  
$c->connectionCount = 1;
$c->messageHandler = $messageHandler; 

$c->start();  
$loop->run();

require_once '../../zbus.php';

$b = new ClientBootstrap();

$b->ha(false)    //режим прямого подключения
->async(false)
->serviceName("MyRpc")
->serviceAddress("localhost:15555");//когда HA включена, адрес интерпретируется как Tracker/NameServer

$b->run(function ($rpc, $boostrap) {
    
    $rpc->module = "InterfaceExample";
    
    // 1)слабо типизированный вызов
    $req = new Request("plus", array( 1, 2 ));  //создать запрос, метод+параметры+[модуль]
    $res = $rpc->invoke($req);
    echo $res . PHP_EOL;
    
    // 2)сильно типизированный вызов
    $res = $rpc->plus(1, 2);
    echo $res . PHP_EOL;
    
    $res = $rpc->testEncoding();
    echo $res . PHP_EOL;
    
    $res = $rpc->noReturn();
    echo $res. ### RPC сервис


Требуется включить файл '../../zbus.php';

// Функции возвращают тип данных, кроме Message, в формате JSON
class MyService {   
    public function echo($msg){
        return $msg . ", From PHP";
    }
    
    public function plus($a, $b) {
        return $a + $b;
    }  
    
    public function testEncoding() {
        return "中文";
    }
    
    public function noReturn() {
        
    }
    
    public function getUser(){
        return array("name"=>"Hong", "age"=>"18");
    }
    
    public function getBin(){ 
        $bytes = array();
        for($i = 0; $i < 10; $i++){
            array_push($bytes, 0);
        } 
        
        $string = implode(array_map("chr", $bytes));
        return base64_encode($string);
    }
    
    public function throwException(){
        throw new Exception("exception throw!");
    }  
    
    // Перенаправление на страницу
    public function redirect(){
        $msg = new Message();
        $msg->status = 302;
        $msg->headers['location'] = "/"; 
        return $msg;
    }
}  


Logger::$Level = Logger::DEBUG;
  
$b = new ServiceBootstrap(); 
$b->addModule("InterfaceExample", new MyService()); // Идентификатор модуля, отображается в URL

$b->serviceName('MyRpc')
  ->serviceAddress('localhost:15555;localhost:15556')
  ->connectionCount(1)
  ->enableDoc(true)
  ->start();