ZeroMQ 官方地址 :http://api.zeromq.org/4-2:zmq-socket-monitor
zmq_socket_monitor(3) ØMQ Manual - ØMQ/4.1.0
Name
zmq_socket_monitor - 注册一个监控回调函数
Synopsis
int zmq_socket_monitor (void *socket,char * *addr,int events);
Description
zmq_socket_monitor() 函数会产生一个PAIR类型的socket,用来把socket状态改变(事件)通过inproc://传输方式广播到制定的终结点(endpoint)上。
消息包括两个帧,第一部分包含着事件ID和与其相关联的值。第二帧以字符串方式保存收到影响的终结点。
第一帧的组织方式是:16 bit的事件ID和32 bit的事件值。
事件ID和19:23值是按照本地的字节序排序的(即这个运行着的进程所在的机器)。在这两部分之间没有其它的数据。
事件值必须根据根据当前的context和事件ID进行解析。更多细节请查看下面可支持的事件。
只有方向确定的传输方式(tcp、ipc)才能支持此初始行为。
支持的事件
ZMQ_EVENT_CONNECTED:链接已建立
当和远程的另一端的连接建立好的时候,ZMQ_EVENT_CONNECTED事件会被触发。同步和异步事件都会发生触发此事件。事件值是新连接的socket的FD。
ZMQ_EVENT_CONNECT_DELAYED:同步连接失败,仍在进行重试
当一个请求立即连接的尝试被延迟并且仍然在尝试的时候,此事件被触发。事件值没有意义。
ZMQ_EVENT_CONNECT_RETRIED:尝试异步连接/重连
当一个连接尝试被重连计时器捕获后此事件被触发。重连间隔根据所有的尝试情况进行计算。事件值是重连间隔。
ZMQ_EVENT_LISTENING:socket已经绑定了某个地址,准备好接受连接请求
当一个socket成功的绑定在一个端口上的时候此事件被触发。事件值是新绑定的socket的FD。
ZMQ_EVENT_BIND_Failed:socket无法绑定在这个地址上
当一个socket无法绑定在给定的端口上时此事件被触发。事件值是绑定函数修改后的errno值。
ZMQ_EVENT_ACCEPTED:连接请求被接受
一个从远端到来的连接被一个绑定了地址的socket接受并建立了连接是会触发此事件。事件值是被接受socket的FD。
ZMQ_EVENT_ACCEPT_Failed:无法接受客户端的连接请求
当一个连接请求试图连接另一个socket失败的时候会触发此事件。事件值是accept设置的errno值。
ZMQ_EVENT_CLOSED:连接关闭
当一个连接的底层描述符被关闭是会触发此事件。事件值是被关闭的socket的FD。此时这个FD已经被关闭了。
ZMQ_EVENT_CLOSE_Failed:连接无法被关闭
当一个描述符无法被释放回OS的时候会触发此事件。注意:只对IPC socket有效。事件值是释放失败时设置的errno值。
ZMQ_EVENT_disCONNECTED:会话被破坏
当流引擎(尤其是TCP、IPC)出现了崩溃的/被破坏的会话时,此事件被触发。事件值是socket的FD。
Return value
当函数zmq_socket_monitor() 执行成功时,返回0或者更大值。否则返回 -1,并且设置errno为下列指定值。
Errors
ETERM
与被指定的socket关联的ZMQ context 被终结了。
EPROTONOSUPPORT
无法支持被请求的传输协议。监视socket必须要使用inproc://方式进行传输。
EINVAL
提供的地址节点不能用。
Example
监视一个REP socket的连接状态
#include <stdio.h>
#include <zmq.h>
#include <pthread.h>
#include <string.h>
#include <assert.h>
static int read_msg(void* s,zmq_event_t* event,char* ep)
{
int rc ;
zmq_msg_t msg1; // binary part
zmq_msg_init (&msg1);
zmq_msg_t msg2; // address part
zmq_msg_init (&msg2);
rc = zmq_msg_recv (&msg1,s,0);
if (rc == -1 && zmq_errno() == ETERM)
return 1 ;
assert (rc != -1);
assert (zmq_msg_more(&msg1) != 0);
rc = zmq_msg_recv (&msg2,0);
if (rc == -1 && zmq_errno() == ETERM)
return 1;
assert (rc != -1);
assert (zmq_msg_more(&msg2) == 0);
// copy binary data to event struct
const char* data = (char*)zmq_msg_data(&msg1);
memcpy(&(event->event),data,sizeof(event->event));
memcpy(&(event->value),data+sizeof(event->event),sizeof(event->value));
// copy address part
const size_t len = zmq_msg_size(&msg2) ;
ep = memcpy(ep,zmq_msg_data(&msg2),len);
*(ep + len) = 0 ;
return 0 ;
}
// REP socket monitor thread
static void *rep_socket_monitor (void *ctx)
{
zmq_event_t event;
static char addr[1025] ;
int rc;
printf("starting monitor...\n");
void *s = zmq_socket (ctx,ZMQ_PAIR);
assert (s);
rc = zmq_connect (s,"inproc://monitor.rep");
assert (rc == 0);
while (!read_msg(s,&event,addr)) {
switch (event.event) {
case ZMQ_EVENT_LISTENING:
printf ("listening socket descriptor %d\n",event.value);
printf ("listening socket address %s\n",addr);
break;
case ZMQ_EVENT_ACCEPTED:
printf ("accepted socket descriptor %d\n",event.value);
printf ("accepted socket address %s\n",addr);
break;
case ZMQ_EVENT_CLOSE_Failed:
printf ("socket close failure error code %d\n",event.value);
printf ("socket address %s\n",addr);
break;
case ZMQ_EVENT_CLOSED:
printf ("closed socket descriptor %d\n",event.value);
printf ("closed socket address %s\n",addr);
break;
case ZMQ_EVENT_disCONNECTED:
printf ("disconnected socket descriptor %d\n",event.value);
printf ("disconnected socket address %s\n",addr);
break;
}
}
zmq_close (s);
return NULL;
}
int main()
{
const char* addr = "tcp://127.0.0.1:6666" ;
pthread_t thread ;
// Create the infrastructure
void *ctx = zmq_init (1);
assert (ctx);
// REP socket
void* rep = zmq_socket (ctx,ZMQ_REP);
assert (rep);
// REP socket monitor,all events
int rc = zmq_socket_monitor (rep,"inproc://monitor.rep",ZMQ_EVENT_ALL);
assert (rc == 0);
rc = pthread_create (&thread,NULL,rep_socket_monitor,ctx);
assert (rc == 0);
rc = zmq_bind (rep,addr);
assert (rc == 0);
// Allow some time for event detection
zmq_sleep (1);
// Close the REP socket
rc = zmq_close (rep);
assert (rc == 0);
zmq_term (ctx);
return 0 ;
}