Manpages - zmq_socket_monitor.3

zmq_socket_monitor - monitor socket events

int zmq_socket_monitor (void */*socket/, char *endpoint, int events);*

The zmq_socket_monitor() method lets an application thread track socket events (like connects) on a ZeroMQ socket. Each call to this method creates a ZMQ_PAIR socket and binds that to the specified inproc:// endpoint. To collect the socket events, you must create your own ZMQ_PAIR socket, and connect that to the endpoint.

Note that there is also a DRAFT function *zmq_socket_monitor_versioned*(3), which allows to subscribe to events that provide more information. Calling zmq_socket_monitor is equivalent to calling zmq_socket_monitor_versioned with the event_version parameter set to 1, with the exception of error cases.

The events argument is a bitmask of the socket events you wish to monitor, see Supported events below. To monitor all events, use the event value ZMQ_EVENT_ALL. NOTE: as new events are added, the catch-all value will start returning them. An application that relies on a strict and fixed sequence of events must not use ZMQ_EVENT_ALL in order to guarantee compatibility with future versions.

Each event is sent as two frames. The first frame contains an event number (16 bits), and an event value (32 bits) that provides additional data according to the event number. The second frame contains a string that specifies the affected endpoint.

    The _zmq_socket_monitor()_ method supports only connection-oriented
    transports, that is, TCP, IPC, and TIPC.

The zmq_socket_monitor() function returns a value of 0 or greater if successful. Otherwise it returns -1 and sets errno to one of the values defined below.

ETERM

The 0MQ context associated with the specified socket was terminated.

EPROTONOSUPPORT

The requested transport protocol is not supported. Monitor sockets are required to use the inproc:// transport.

EINVAL

The endpoint supplied is invalid.

Monitoring client and server sockets.

    //  Read one event off the monitor socket; return value and address
    //  by reference, if not null, and event number by value. Returns -1
    //  in case of error.

    static int
    get_monitor_event (void *monitor, int *value, char **address)
    {
        //  First frame in message contains event number and value
        zmq_msg_t msg;
        zmq_msg_init (&msg);
        if (zmq_msg_recv (&msg, monitor, 0) == -1)
            return -1;              //  Interrupted, presumably
        assert (zmq_msg_more (&msg));

        uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
        uint16_t event = *(uint16_t *) (data);
        if (value)
            *value = *(uint32_t *) (data + 2);

        //  Second frame in message contains event address
        zmq_msg_init (&msg);
        if (zmq_msg_recv (&msg, monitor, 0) == -1)
            return -1;              //  Interrupted, presumably
        assert (!zmq_msg_more (&msg));

        if (address) {
            uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
            size_t size = zmq_msg_size (&msg);
            *address = (char *) malloc (size + 1);
            memcpy (*address, data, size);
            (*address)[size] = 0;
        }
        return event;
    }

    int main (void)
    {
        void *ctx = zmq_ctx_new ();
        assert (ctx);

        //  Well monitor these two sockets
        void *client = zmq_socket (ctx, ZMQ_DEALER);
        assert (client);
        void *server = zmq_socket (ctx, ZMQ_DEALER);
        assert (server);

        //  Socket monitoring only works over inproc://
        int rc = zmq_socket_monitor (client, "tcp://127.0.0.1:9999", 0);
        assert (rc == -1);
        assert (zmq_errno () == EPROTONOSUPPORT);

        //  Monitor all events on client and server sockets
        rc = zmq_socket_monitor (client, "inproc://monitor-client", ZMQ_EVENT_ALL);
        assert (rc == 0);
        rc = zmq_socket_monitor (server, "inproc://monitor-server", ZMQ_EVENT_ALL);
        assert (rc == 0);

        //  Create two sockets for collecting monitor events
        void *client_mon = zmq_socket (ctx, ZMQ_PAIR);
        assert (client_mon);
        void *server_mon = zmq_socket (ctx, ZMQ_PAIR);
        assert (server_mon);

        //  Connect these to the inproc endpoints so theyll get events
        rc = zmq_connect (client_mon, "inproc://monitor-client");
        assert (rc == 0);
        rc = zmq_connect (server_mon, "inproc://monitor-server");
        assert (rc == 0);

        //  Now do a basic ping test
        rc = zmq_bind (server, "tcp://127.0.0.1:9998");
        assert (rc == 0);
        rc = zmq_connect (client, "tcp://127.0.0.1:9998");
        assert (rc == 0);
        bounce (client, server);

        //  Close client and server
        close_zero_linger (client);
        close_zero_linger (server);

        //  Now collect and check events from both sockets
        int event = get_monitor_event (client_mon, NULL, NULL);
        if (event == ZMQ_EVENT_CONNECT_DELAYED)
            event = get_monitor_event (client_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_CONNECTED);
        event = get_monitor_event (client_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_HANDSHAKE_SUCCEEDED);
        event = get_monitor_event (client_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_MONITOR_STOPPED);

        //  This is the flow of server events
        event = get_monitor_event (server_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_LISTENING);
        event = get_monitor_event (server_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_ACCEPTED);
        event = get_monitor_event (server_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_HANDSHAKE_SUCCEEDED);
        event = get_monitor_event (server_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_CLOSED);
        event = get_monitor_event (server_mon, NULL, NULL);
        assert (event == ZMQ_EVENT_MONITOR_STOPPED);

        //  Close down the sockets
        close_zero_linger (client_mon);
        close_zero_linger (server_mon);
        zmq_ctx_term (ctx);

        return 0 ;
    }

*zmq*(7)

This page was written by the 0MQ community. To make a change please read the 0MQ Contribution Policy at http://www.zeromq.org/docs:contributing.