如何绑定ZeroMQ套接字与棘轮web套接字库,使实时应用程序的php应用
How to bind ZeroMQ socket with Ratchet web-socket library to make real time application for php application?
在涉及websocket, Ratchet和ZeroMQ的整个领域,我只是一个初学者。
我的基本理解是:
websocket
是帮助在服务器和客户端之间创建开放连接的东西。
Ratchet
是一个基于PHP的库,它使用PHP的核心套接字函数来创建一个PHP套接字框架,使我们能够轻松地进行PHP套接字编程。
ZeroMQ
是一个套接字库,帮助非棘轮应用程序(其他PHP脚本)通过棘轮套接字和web套接字发送数据。
我在棘轮中找到了关于"你好世界"answers"推手"的教程,但它们似乎都不完整,只教如何使用控制台。我也发现棘轮的例子在github,但它没有适当的文档。我正在寻找一个完整的例子(与专用的html页面和javascript)
下面是我正在处理的代码:这是我正在进行Ajax请求的控制器的方法之一。这个方法将创建一个新帖子(假设)。我想用ZeroMq的帮助下通过广播/推送在多个客户端浏览器中动态更新帖子列表。
控制器中的方法:
public function create_new_post(){
// ------
// code to create a new post.
// -------
// After creating a post
$response = [
'new_post_title' => $title,
'post_id' => $id
];
$context = new ZMQContext();
$socket = $context->getSocket(ZMQ::SOCKET_PUSH, 'my pusher');
$socket->connect("tcp://localhost:8000");
$socket->send(json_encode($response));
}
推杆式文件:
use Ratchet'ConnectionInterface;
use Ratchet'Wamp'WampServerInterface;
class Pusher implements WampServerInterface{
public function onPostEntry($data){
// Data that were sent by ZeroMQ through create_new_post() method
$entry_data = json_decode($data);
// AND AFTER THIS, I DONT HAVE CLUE OF WHAT TO DO NEXT !!
}
}
Shell脚本运行server:
require dirname(__DIR__) . '/vendor/autoload.php';
$loop = React'EventLoop'Factory::create();
$pusher = new MyApp'Pusher;
// Listen for the web server to make a ZeroMQ push after an ajax request
$context = new React'ZMQ'Context($loop);
$pull = $context->getSocket(ZMQ::SOCKET_PULL);
$pull->bind('tcp://127.0.0.1:8000');
$pull->on('message', array($pusher, 'onBidEntry'));
// Set up our WebSocket server for clients wanting real-time updates
$webSock = new React'Socket'Server($loop);
$webSock->listen(8080, '0.0.0.0');
$webServer = new Ratchet'Server'IoServer(
new Ratchet'Http'HttpServer(
new Ratchet'WebSocket'WsServer(
new Ratchet'Wamp'WampServer(
$pusher
)
)
),
$webSock
);
$loop->run();
Shell脚本只告诉它将在端口8080中服务,但是我怎么提到我的路由呢?让我们说,我想要打开连接,只有在页面'mysite/所有的帖子'。此外,我必须在客户端(javascript文件)中编写的脚本是什么,以及如何通过客户端更新特定的DOM对象来接收这些新数据。
我按照你所说的例子做了。在我看来,它们并不是不完整的,但我明白你的意思。Ratchet是一个服务器端脚本,只允许你编写一个实现websockets的服务,并且能够监听ZMQ消息。您将在命令行上启动Ratchet脚本,它将作为与Apache并行的服务运行。
这些都独立于websocket的客户端。按照他们的建议,我在客户端使用了Autobahn.js。这个库实现了WAMP协议。它最大限度地简化了客户端代码。
您的代码的问题是,class Pusher implements WampServerInterface
没有public function onPostEntry
。这个类必须实现WampServerInterface
,这意味着它必须至少有这些功能:
- onSubscribe(ConnectionInterface $conn, $topic)
- onUnSubscribe(ConnectionInterface $conn, $topic)
- onOpen (ConnectionInterface康涅狄格州美元)
- onClose (ConnectionInterface康涅狄格州美元)
- onPublish(ConnectionInterface $conn, $topic, $event, array $exclude, array $eligible onError(ConnectionInterface $conn, 'Exception $e)
- onZMQMessage (jsondata美元)
可以有其他更高级的功能,如call
在客户端远程过程。
在发送端(ZMQ消息),放入以下代码:
$zmq = new ZMQWrapper;
$zqm->publish('posts', $response);
class ZMQWrapper {
function __construct(){
$this->context = new ZMQContext();
$this->socket = $this->context->getSocket(ZMQ::SOCKET_PUSH);
$this->socket->setSockOpt(ZMQ::SOCKOPT_LINGER, 500);
$this->socket->connect("tcp://127.0.0.1:" . ZMQ_PORT);
}
function publish($topic, $msg){
$data = ['topic' => "mb.$topic", 'msg' => $msg];
$this->socket->send(json_encode($data), ZMQ::MODE_DONTWAIT);
}
}
在push文件中添加如下内容:
public function onSubscribe(ConnectionInterface $conn, $topic) {
$log = $this->getLogger();
$topicId = $topic->getId();
$log->info(sprintf('A client subscribed to %s', $topicId));
// you could broadcast that user x joined the discussion
}
public function onUnSubscribe(ConnectionInterface $conn, $topic) {
$log = $this->getLogger();
$topicId = $topic->getId();
$log->info(sprintf('A client unsubscribed from %s', $topicId));
// you could broadcast that user x leaved the discussion
}
public function onOpen(ConnectionInterface $conn) {
$log = $this->getLogger();
$log->info(sprintf('Client %d connected', $conn->resourceId));
$this->clients[$conn->resourceId] = array(); // this will allow you to save state information of the client, you can modify in onSubscribe and onUnsubscribe
// clients will contain the list of all clients
}
public function onClose(ConnectionInterface $conn) {
$log = $this->getLogger();
$log->info(sprintf('Client %d disconnected', $conn->resourceId));
// you could broadcast that user x leaved the discussion
}
public function onPublish(ConnectionInterface $conn, $topic, $event, array $exclude, array $eligible) {
$log = $this->getLogger();
$topicId = $topic->getId();
$log->info(sprintf('Client %d published to %s : %s', $conn->resourceId, $topicId, json_encode($event)));
foreach($topic->getIterator() as $peer){
if(!in_array($peer->WAMP->sessionId, $exclude)){
$peer->event($topicId, $event);
}
}
}
最后一块在客户机上。如果用户打开页面mysite/allposts
,在javascript中包含autobahn.js
。websocket将在变量ab
下可用。然后:
打开页面时:
var currentSession;
ab.connect(
Paths.ws,
function(session) { // onconnect
currentSession = session
onWsConnect(session)
},
function(code, reason, detail) {// onhangup
onWsDisconnect(code, reason, detail)
},{
maxRetries: 60,
retryDelay: 2000,
skipSubprotocolCheck: true
}
)
currentSession.subscribe('posts', onPostReceived)
function onPostReceived(topic, message){
//display the new post
}
关闭页面时:
currentSession.unsubscribe(topic)
你注意到我把每件事都写得很笼统。这允许我在同一个系统中处理几种类型的消息。不同的是ZMQ消息和currentSession.subscribe
的参数。
在我的实现中,我还跟踪打开连接的登录用户,但我剥离了这部分代码。
- 如何使用HTML5与UDP套接字通信
- 在node.js中写入ECONNRESET错误和套接字连接检查
- java,javascript签名的小程序(文件选择,ftp上传)安全性/套接字异常
- 套接字服务器向客户端广播,但不从两者接收
- 如何确定套接字中的变量是否匹配
- 如何通过网络套接字连接到树莓派上的Mosquitto经纪人
- 快速的套接字 io 实例/中间件
- PHP:Web套接字和html 5
- 无法从客户端发出套接字消息
- STOMP web套接字回调不'不起作用
- Will my'新对象'在套接字断开连接时被垃圾收集
- 套接字服务器在本地获得500毫秒的往返时间
- 将服务器托管的Web应用程序连接到本地Web套接字
- 通过套接字发送画布元素,JS中的最大堆栈大小
- 如何使用套接字将node.js与ada连接起来
- 套接字io客户端网络::ERR_CONNECTION_REFUSED
- 节点.js设置套接字 ID
- node.js 需要太多的 TCP 套接字
- 在不关闭套接字的情况下重启实时node.js服务器
- 如何绑定ZeroMQ套接字与棘轮web套接字库,使实时应用程序的php应用