1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
|
'\" t
.\" Title: zmq_ctx_socket_monitor
.\" Author: [see the "AUTHORS" section]
.\" Generator: DocBook XSL Stylesheets v1.76.1 <http://docbook.sf.net/>
.\" Date: 10/14/2014
.\" Manual: 0MQ Manual
.\" Source: 0MQ 4.0.5
.\" Language: English
.\"
.TH "ZMQ_CTX_SOCKET_MONIT" "3" "10/14/2014" "0MQ 4\&.0\&.5" "0MQ Manual"
.\" -----------------------------------------------------------------
.\" * Define some portability stuff
.\" -----------------------------------------------------------------
.\" ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.\" http://bugs.debian.org/507673
.\" http://lists.gnu.org/archive/html/groff/2009-02/msg00013.html
.\" ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.ie \n(.g .ds Aq \(aq
.el .ds Aq '
.\" -----------------------------------------------------------------
.\" * set default formatting
.\" -----------------------------------------------------------------
.\" disable hyphenation
.nh
.\" disable justification (adjust text to left margin only)
.ad l
.\" -----------------------------------------------------------------
.\" * MAIN CONTENT STARTS HERE *
.\" -----------------------------------------------------------------
.SH "NAME"
zmq_socket_monitor \- register a monitoring callback
.SH "SYNOPSIS"
.sp
\fBint zmq_socket_monitor (void \fR\fB\fI*socket\fR\fR\fB, char * \fR\fB\fI*addr\fR\fR\fB, int \fR\fB\fIevents\fR\fR\fB);\fR
.SH "DESCRIPTION"
.sp
The \fIzmq_socket_monitor()\fR function shall spawn a \fIPAIR\fR socket that publishes socket state changes (events) over the inproc:// transport to a given endpoint\&.
.sp
Messages consist of 2 Frames, the first containing the event\-id and the associated value\&. The second frame holds the affected endpoint as string\&.
.sp
The layout of the first Frame is: 16 bit event id 32 bit event value
.sp
event id and value are in the native byte order (for the machine the application is running on)\&. There is no padding between the fields\&.
.sp
The event value has to be interpreted in the context of the event id\&. See \fISupported events\fR below for details\&.
.sp
.if n \{\
.RS 4
.\}
.nf
Only connection oriented (tcp and ipc) transports are supported in this initial
implementation\&.
.fi
.if n \{\
.RE
.\}
.SH "SUPPORTED EVENTS"
.SS "ZMQ_EVENT_CONNECTED: connection established"
.sp
The \fIZMQ_EVENT_CONNECTED\fR event triggers when a connection has been established to a remote peer\&. This can happen either synchronous or asynchronous\&. Value is the FD of the newly connected socket\&.
.SS "ZMQ_EVENT_CONNECT_DELAYED: synchronous connect failed, it\(cqs being polled"
.sp
The \fIZMQ_EVENT_CONNECT_DELAYED\fR event triggers when an immediate connection attempt is delayed and its completion is being polled for\&. Value has no meaning\&.
.SS "ZMQ_EVENT_CONNECT_RETRIED: asynchronous connect / reconnection attempt"
.sp
The \fIZMQ_EVENT_CONNECT_RETRIED\fR event triggers when a connection attempt is being handled by reconnect timer\&. The reconnect interval\(cqs recomputed for each attempt\&. Value is the reconnect interval\&.
.SS "ZMQ_EVENT_LISTENING: socket bound to an address, ready to accept connections"
.sp
The \fIZMQ_EVENT_LISTENING\fR event triggers when a socket\(cqs successfully bound to a an interface\&. Value is the FD of the newly bound socket\&.
.SS "ZMQ_EVENT_BIND_FAILED: socket could not bind to an address"
.sp
The \fIZMQ_EVENT_BIND_FAILED\fR event triggers when a socket could not bind to a given interface\&. Value is the errno generated by the bind call\&.
.SS "ZMQ_EVENT_ACCEPTED: connection accepted to bound interface"
.sp
The \fIZMQ_EVENT_ACCEPTED\fR event triggers when a connection from a remote peer has been established with a socket\(cqs listen address\&. Value is the FD of the accepted socket\&.
.SS "ZMQ_EVENT_ACCEPT_FAILED: could not accept client connection"
.sp
The \fIZMQ_EVENT_ACCEPT_FAILED\fR event triggers when a connection attempt to a socket\(cqs bound address fails\&. Value is the errno generated by accept\&.
.SS "ZMQ_EVENT_CLOSED: connection closed"
.sp
The \fIZMQ_EVENT_CLOSED\fR event triggers when a connection\(cqs underlying descriptor has been closed\&. Value is the former FD of the for the closed socket\&. FD has been closed already!
.SS "ZMQ_EVENT_CLOSE_FAILED: connection couldn\(cqt be closed"
.sp
The \fIZMQ_EVENT_CLOSE_FAILED\fR event triggers when a descriptor could not be released back to the OS\&. Implementation note: ONLY FOR IPC SOCKETS\&. Value is the errno generated by unlink\&.
.SS "ZMQ_EVENT_DISCONNECTED: broken session"
.sp
The \fIZMQ_EVENT_DISCONNECTED\fR event triggers when the stream engine (tcp and ipc specific) detects a corrupted / broken session\&. Value is the FD of the socket\&.
.SH "RETURN VALUE"
.sp
The \fIzmq_socket_monitor()\fR function returns a value of 0 or greater if successful\&. Otherwise it returns \-1 and sets \fIerrno\fR to one of the values defined below\&.
.SH "ERRORS"
.PP
\fBETERM\fR
.RS 4
The 0MQ
\fIcontext\fR
associated with the specified
\fIsocket\fR
was terminated\&.
.RE
.PP
\fBEPROTONOSUPPORT\fR
.RS 4
The requested
\fItransport\fR
protocol is not supported\&. Monitor sockets are required to use the inproc:// transport\&.
.RE
.PP
\fBEINVAL\fR
.RS 4
The endpoint supplied is invalid\&.
.RE
.SH "EXAMPLE"
.PP
\fBObserving a REP socket\(cqs connection state\fR.
.sp
.if n \{\
.RS 4
.\}
.nf
#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, s, 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\&.\&.\&.\en");
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\en", event\&.value);
printf ("listening socket address %s\en", addr);
break;
case ZMQ_EVENT_ACCEPTED:
printf ("accepted socket descriptor %d\en", event\&.value);
printf ("accepted socket address %s\en", addr);
break;
case ZMQ_EVENT_CLOSE_FAILED:
printf ("socket close failure error code %d\en", event\&.value);
printf ("socket address %s\en", addr);
break;
case ZMQ_EVENT_CLOSED:
printf ("closed socket descriptor %d\en", event\&.value);
printf ("closed socket address %s\en", addr);
break;
case ZMQ_EVENT_DISCONNECTED:
printf ("disconnected socket descriptor %d\en", event\&.value);
printf ("disconnected socket address %s\en", 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 ;
}
.fi
.if n \{\
.RE
.\}
.sp
.SH "SEE ALSO"
.sp
\fBzmq\fR(7)
.SH "AUTHORS"
.sp
This page was written by the 0MQ community\&. To make a change please read the 0MQ Contribution Policy at \m[blue]\fBhttp://www\&.zeromq\&.org/docs:contributing\fR\m[]\&.
|
