自学教程：C++ zsys_info函数代码示例

static voids_zdir_watch_subscribe (zdir_watch_t *watch, const char *path){    if (watch->verbose)        zsys_info ("zdir_watch: Subscribing to directory path: %s", path);    zdir_watch_sub_t *sub = (zdir_watch_sub_t *) zmalloc (sizeof (zdir_watch_sub_t));    sub->dir = zdir_new (path, NULL);    if (!sub->dir) {        if (watch->verbose)            zsys_error ("zdir_watch: Unable to create zdir for path: %s", path);        zsock_signal (watch->pipe, 1);        return;    }    int rc = zhash_insert (watch->subs, path, sub);    if (rc) {        if (watch->verbose)            zsys_error ("zdir_watch: Unable to insert path '%s' into subscription list", path);        zsock_signal (watch->pipe, 1);        return;    }    void *item = zhash_freefn (watch->subs, path, s_sub_free);    if (item != sub) {        if (watch->verbose)            zsys_error ("zdir_watch: Unable to set free fn for path %s", path);        zsock_signal (watch->pipe, 1);        return;    }    if (watch->verbose)        zsys_info ("zdir_watch: Successfully subscribed to %s", path);    zsock_signal (watch->pipe, 0);}

static voidzyre_node_dump (zyre_node_t *self){    zsys_info ("zyre_node: dump state");    zsys_info (" - name=%s uuid=%s", self->name, zuuid_str (self->uuid));    zsys_info (" - endpoint=%s", self->endpoint);    if (self->beacon_port)         zsys_info (" - discovery=beacon port=%d interval=%zu",                   self->beacon_port, self->interval);    else {        zsys_info (" - discovery=gossip");        if (self->gossip_bind)            zsys_info ("   - bind endpoint=%s", self->gossip_bind);        if (self->gossip_connect)            zsys_info ("   - connect endpoint=%s", self->gossip_connect);    }    zsys_info (" - headers=%zu:", zhash_size (self->headers));    zhash_foreach (self->headers, (zhash_foreach_fn *) zyre_node_log_pair, self);        zsys_info (" - peers=%zu:", zhash_size (self->peers));    zhash_foreach (self->peers, (zhash_foreach_fn *) zyre_node_log_item, self);    zsys_info (" - groups=%zu:", zhash_size (self->own_groups));    zhash_foreach (self->own_groups, (zhash_foreach_fn *) zyre_node_log_item, self);}

static intzyre_handler (zloop_t *loop, zsock_t *reader, void *argument){    server_t *self = (server_t *) argument;    zmsg_t *msg = zyre_recv (self->zyre);    if (!msg)        return -1;              //  Interrupted    char *command = zmsg_popstr (msg);    char *peer_id = zmsg_popstr (msg);    char *peer_name = zmsg_popstr (msg);    if (streq (command, "ENTER"))        zsys_info ("ZPIPES server appeared at %s", peer_name);    else    if (streq (command, "EXIT"))        zsys_info ("ZPIPES server vanished from %s", peer_name);    else    if (streq (command, "SHOUT")) {        char *group = zmsg_popstr (msg);        if (streq (group, "ZPIPES"))            server_process_cluster_command (self, peer_id, peer_name, msg, false);        zstr_free (&group);    }    else    if (streq (command, "WHISPER"))        server_process_cluster_command (self, peer_id, peer_name, msg, true);        zstr_free (&command);    zstr_free (&peer_id);    zstr_free (&peer_name);    zmsg_destroy (&msg);        return 0;}

intzyre_peer_send (zyre_peer_t *self, zre_msg_t **msg_p){    assert (self);    zre_msg_t *msg = *msg_p;    assert (msg);    if (self->connected) {        self->sent_sequence += 1;        zre_msg_set_sequence (msg, self->sent_sequence);        if (self->verbose)            zsys_info ("(%s) send %s to peer=%s sequence=%d",                self->origin,                zre_msg_command (msg),                self->name? self->name: "-",                zre_msg_sequence (msg));        if (zre_msg_send (msg, self->mailbox)) {            if (errno == EAGAIN) {                if (self->verbose)                    zsys_info ("(%s) disconnect from peer (EAGAIN): name=%s",                        self->origin, self->name);                zyre_peer_disconnect (self);                return -1;            }            //  Can't get any other error here            assert (false);        }    }    zre_msg_destroy (msg_p);    return 0;}

static intzyre_node_ping_peer (const char *key, void *item, void *argument){    zyre_peer_t *peer = (zyre_peer_t *) item;    zyre_node_t *self = (zyre_node_t *) argument;    if (zclock_mono () >= zyre_peer_expired_at (peer)) {        if (self->verbose)            zsys_info ("(%s) peer expired name=%s endpoint=%s",                self->name, zyre_peer_name (peer), zyre_peer_endpoint (peer));        zyre_node_remove_peer (self, peer);    }    else    if (zclock_mono () >= zyre_peer_evasive_at (peer)) {        //  If peer is being evasive, force a TCP ping.        //  TODO: do this only once for a peer in this state;        //  it would be nicer to use a proper state machine        //  for peer management.        if (self->verbose)            zsys_info ("(%s) peer seems dead/slow name=%s endpoint=%s",                self->name, zyre_peer_name (peer), zyre_peer_endpoint (peer));        zre_msg_t *msg = zre_msg_new (ZRE_MSG_PING);        zyre_peer_send (peer, &msg);        // Inform the calling application this peer is being evasive    	zstr_sendm (self->outbox, "EVASIVE");	zstr_sendm (self->outbox, zyre_peer_identity (peer));    	zstr_send (self->outbox, zyre_peer_name (peer));    }    return 0;}

int main (int argc, char *argv []){    int argn = 1;    char *filename = "mycert.txt";    if (argn < argc)        filename = argv [argn++];    zsys_info ("Creating new CURVE certificate in %s", filename);    zcert_t *cert = zcert_new ();    if (s_get_meta (cert, "Enter your full name:", "name")    ||  s_get_meta (cert, "Enter your email address:", "email")    ||  s_get_meta (cert, "Enter your organization:", "organization"))        return -1;            char *timestr = zclock_timestr ();    zcert_set_meta (cert, "created-by", "CZMQ zmakecert");    zcert_set_meta (cert, "date-created", "%s", timestr);    free (timestr);    zcert_dump (cert);    zcert_save (cert, filename);    zsys_info ("CURVE certificate created in %s and %s_secret", filename, filename);    zcert_destroy (&cert);    return 0;}

boolzyre_peer_messages_lost (zyre_peer_t *self, zre_msg_t *msg){    assert (self);    assert (msg);    //  The sequence number set by the peer, and our own calculated    //  sequence number should be the same.    if (self->verbose)        zsys_info ("(%s) recv %s from peer=%s sequence=%d",            self->origin,            zre_msg_command (msg),            self->name? self->name: "-",            zre_msg_sequence (msg));    //  HELLO always MUST have sequence = 1    if (zre_msg_id (msg) == ZRE_MSG_HELLO)        self->want_sequence = 1;    else        self->want_sequence += 1;    if (self->want_sequence != zre_msg_sequence (msg)) {        zsys_info ("(%s) seq error from peer=%s expect=%d, got=%d",            self->origin,            self->name? self->name: "-",            self->want_sequence,            zre_msg_sequence (msg));        return true;    }    return false;}

static bools_authenticate_plain (self_t *self, zap_request_t *request){    if (self->passwords) {        zhashx_refresh (self->passwords);        char *password = (char *) zhashx_lookup (self->passwords, request->username);        if (password && streq (password, request->password)) {            if (self->verbose)                zsys_info ("zauth: - allowed (PLAIN) username=%s password=%s",                           request->username, request->password);            return true;        }        else {            if (self->verbose)                zsys_info ("zauth: - denied (PLAIN) username=%s password=%s",                           request->username, request->password);            return false;        }    }    else {        if (self->verbose)            zsys_info ("zauth: - denied (PLAIN) no password file defined");        return false;    }}

static intpipe_attach_remote_writer (pipe_t *self, const char *remote, bool unicast){    assert (self);    if (self->reader == REMOTE_NODE) {        //  We're witnessing two nodes chatting, so we can drop the pipe        //  and forget all about it        pipe_destroy (&self);        return 0;    }    else    if (self->writer == NULL) {        //  This is how we indicate a remote writer        self->writer = REMOTE_NODE;        self->remote = strdup (remote);        zsys_info ("%s: attach remote writer", self->name);        if (self->reader && !unicast) {            //  Tell remote node we're acting as reader, if we got a            //  broadcast message. If we got a unicast message, the peer            //  already knows about us, so don't re-echo the message            zmsg_t *msg = zmsg_new ();            zmsg_addstr (msg, "HAVE READER");            zmsg_addstr (msg, self->name);            zyre_whisper (self->server->zyre, self->remote, &msg);            zsys_info ("%s: tell peer we are now reader", self->name);        }        return 0;    }    zsys_info ("%s: pipe already has writer: ignored", self->name);    return -1;}

static voidclient_expired (client_t *self){    if (!self->address)        zsys_info ("client %u (incomplete connection) - expired",self->unique_id);    else if (*self->address)        zsys_info ("client %u address='%s' - expired", self->unique_id, self->address);}

int main (int argc, char *argv []){    //  Get number of nodes N to simulate    //  We need 3 x N x N + 3N file handles    int max_nodes = 10;    int nbr_nodes = 0;    if (argc > 1)        max_nodes = atoi (argv [1]);    assert (max_nodes);    int max_iterations = -1;    int nbr_iterations = 0;    if (argc > 2)        max_iterations = atoi (argv [2]);    //  Our gossip network will use one fixed hub (not a Zyre node),    //  to which all nodes will connect    zactor_t *hub = zactor_new (zgossip, "hub");    zstr_sendx (hub, "BIND", "inproc://zyre-hub", NULL);            //  We address nodes as an array of actors    zactor_t **actors = (zactor_t **) zmalloc (sizeof (zactor_t *) * max_nodes);    //  We will randomly start and stop node threads    uint index;    while (!zsys_interrupted) {        index = randof (max_nodes);        //  Toggle node thread        if (actors [index]) {            zactor_destroy (&actors [index]);            actors [index] = NULL;            zsys_info ("stopped node (%d running)", --nbr_nodes);        }        else {            char node_name [10];            sprintf (node_name, "node-%d", index);            actors [index] = zactor_new (node_actor, strdup (node_name));            zsys_info ("started node (%d running)", ++nbr_nodes);        }        nbr_iterations++;        if (max_iterations > 0 && nbr_iterations >= max_iterations)            break;        //  Sleep ~300 msecs randomly so we smooth out activity        zclock_sleep (randof (100) + 100);    }    zsys_info ("stopped tester (%d iterations)", nbr_iterations);    //  Stop all remaining actors    for (index = 0; index < max_nodes; index++) {        if (actors [index])            zactor_destroy (&actors [index]);    }    free (actors);        zactor_destroy (&hub);    return 0;}

static ints_on_read_timer (zloop_t *loop, int timer_id, void *arg){    zdir_watch_t *watch = (zdir_watch_t *) arg;    void *data;    for (data = zhash_first (watch->subs); data != NULL; data = zhash_next (watch->subs))    {        zdir_watch_sub_t *sub = (zdir_watch_sub_t *) data;        zdir_t *new_dir = zdir_new (zdir_path (sub->dir), NULL);        if (!new_dir) {            if (watch->verbose)                zsys_error ("zdir_watch: Unable to create new zdir for path %s", zdir_path (sub->dir));            continue;        }        // Determine if anything has changed.        zlist_t *diff = zdir_diff (sub->dir, new_dir, "");        // Do memory management before error handling...        zdir_destroy (&sub->dir);        sub->dir = new_dir;        if (!diff) {            if (watch->verbose)                zsys_error ("zdir_watch: Unable to create diff for path %s", zdir_path (sub->dir));            continue;        }        if (zlist_size (diff) > 0) {            if (watch->verbose) {                zdir_patch_t *patch = (zdir_patch_t *) zlist_first (diff);                zsys_info ("zdir_watch: Found %d changes in %s:", zlist_size (diff), zdir_path (sub->dir));                while (patch)                {                    zsys_info ("zdir_watch:   %s %s", zfile_filename (zdir_patch_file (patch), NULL), zdir_patch_op (patch) == ZDIR_PATCH_CREATE? "created": "deleted");                    patch = (zdir_patch_t *) zlist_next (diff);                }            }            if (zsock_send (watch->pipe, "sp", zdir_path (sub->dir), diff) != 0) {                if (watch->verbose)                    zsys_error ("zdir_watch: Unable to send patch list for path %s", zdir_path (sub->dir));                zlist_destroy (&diff);            }            // Successfully sent `diff` list - now owned by receiver        }        else {            zlist_destroy (&diff);        }    }    return 0;}

static voidserver_process_cluster_command (    server_t *self,    const char *peer_id,    const char *peer_name,    zmsg_t *msg,    bool unicast){    char *request = zmsg_popstr (msg);    char *pipename = zmsg_popstr (msg);    zsys_info ("peer=%s command=%s pipe=%s unicast=%d",                peer_name, request, pipename? pipename: "-", unicast);    //  Lookup or create pipe    //  TODO: remote pipes need cleaning up with some timeout    pipe_t *pipe = NULL;    if (pipename) {        pipe = (pipe_t *) zhash_lookup (self->pipes, pipename);        if (!pipe)            pipe = pipe_new (self, pipename);    }    if (pipe && streq (request, "HAVE WRITER"))        pipe_attach_remote_writer (pipe, peer_id, unicast);    else    if (pipe && streq (request, "HAVE READER"))        pipe_attach_remote_reader (pipe, peer_id, unicast);    else    if (pipe && streq (request, "DATA")) {        //  TODO encode these commands as proper protocol        zframe_t *frame = zmsg_pop (msg);        zchunk_t *chunk = zchunk_new (zframe_data (frame), zframe_size (frame));        if (pipe->writer == REMOTE_NODE && pipe->reader) {            zsys_info ("send %d bytes to pipe", (int) zchunk_size (chunk));            pipe_send_data (pipe, &chunk);         }        else            zsys_info ("discard %d bytes, unroutable", (int) zchunk_size (chunk));                    zframe_destroy (&frame);        zchunk_destroy (&chunk);    }    else    if (pipe && streq (request, "DROP READER"))        pipe_drop_remote_reader (&pipe, peer_id);    else    if (pipe && streq (request, "DROP WRITER"))        pipe_drop_remote_writer (&pipe, peer_id);    else    if (streq (request, "DUMP"))        zyre_dump (self->zyre);    else        zsys_warning ("bad request %s from %s", request, peer_name);    zstr_free (&pipename);    zstr_free (&request);}

static voids_zdir_watch_unsubscribe (zdir_watch_t *watch, const char *path){    if (watch->verbose)        zsys_info ("zdir_watch: Unsubscribing from directory path: %s", path);    zhash_delete (watch->subs, path);    if (watch->verbose)        zsys_info ("zdir_watch: Successfully unsubscribed from %s", path);    zsock_signal (watch->pipe, 0);}

voidziflist_print (ziflist_t *self){    interface_t *iface;    for (iface = (interface_t *) zlistx_first ((zlistx_t *) self);         iface != NULL;         iface = (interface_t *) zlistx_next ((zlistx_t *) self)) {        zsys_info (" - interface name : %s", iface->name);        zsys_info (" - interface address : %s", iface->address);        zsys_info (" - interface netmask : %s", iface->netmask);        zsys_info (" - interface broadcast : %s", iface->broadcast);    }}

voidzcertstore_print (zcertstore_t *self){    if (self->location)        zsys_info ("zcertstore: certificates at location=%s:", self->location);    else        zsys_info ("zcertstore: certificates in memory");    zcert_t *cert = (zcert_t *) zhashx_first (self->certs);    while (cert) {        zcert_print (cert);        cert = (zcert_t *) zhashx_next (self->certs);    }}

static voidderegister_the_client (client_t *self){    if (*self->address)        zsys_info ("client address='%s' - de-registering", self->address);    //  Cancel all stream subscriptions    stream_t *stream = (stream_t *) zlistx_detach (self->readers, NULL);    while (stream) {        zsock_send (stream->actor, "sp", "CANCEL", self);        stream = (stream_t *) zlistx_detach (self->readers, NULL);    }    //  Cancel all service offerings    service_t *service = (service_t *) zhashx_first (self->server->services);    while (service) {        offer_t *offer = (offer_t *) zlistx_first (service->offers);        while (offer) {            if (offer->client == self)                zlistx_delete (service->offers, zlistx_cursor (service->offers));            offer = (offer_t *) zlistx_next (service->offers);        }        service = (service_t *) zhashx_next (self->server->services);    }    if (*self->address)        zhashx_delete (self->server->clients, self->address);    mlm_proto_set_status_code (self->message, MLM_PROTO_SUCCESS);}

JNIEXPORT void JNICALLJava_org_zeromq_czmq_Zsys__1_1info (JNIEnv *env, jclass c, jstring format){    char *format_ = (char *) (*env)->GetStringUTFChars (env, format, NULL);    zsys_info (format_);    (*env)->ReleaseStringUTFChars (env, format, format_);}

static voidderegister_the_client (client_t *self){	// If the client never sent CONNECTION_OPEN then self->address was	// never set, so avoid trying to dereference it.  Nothing needs to    // be cleaned up.    if (self->address) {        if (*self->address)            zsys_info ("client %u address='%s' - de-registering", self->unique_id, self->address);        //  Cancel all stream subscriptions        stream_t *stream = (stream_t *) zlistx_detach (self->readers, NULL);        while (stream) {            zsock_send (stream->actor, "sp", "CANCEL", self);            stream = (stream_t *) zlistx_detach (self->readers, NULL);        }        //  Cancel all service offerings        service_t *service = (service_t *) zhashx_first (self->server->services);        while (service) {            offer_t *offer = (offer_t *) zlistx_first (service->offers);            while (offer) {                if (offer->client == self)                    zlistx_delete (service->offers, zlistx_cursor (service->offers));                offer = (offer_t *) zlistx_next (service->offers);            }            service = (service_t *) zhashx_next (self->server->services);        }        if (*self->address)            zhashx_delete (self->server->clients, self->address);    }    mlm_proto_set_status_code (self->message, MLM_PROTO_SUCCESS);}

static ints_zap_request_reply (zap_request_t *self, char *status_code, char *status_text, unsigned char *metadata, size_t metasize){    if (self->verbose)        zsys_info ("zauth: - ZAP reply status_code=%s status_text=%s",                   status_code, status_text);    zmsg_t *msg = zmsg_new ();    int rc = zmsg_addstr(msg, "1.0");    assert (rc == 0);    rc = zmsg_addstr(msg, self->sequence);    assert (rc == 0);    rc = zmsg_addstr(msg, status_code);    assert (rc == 0);    rc = zmsg_addstr(msg, status_text);    assert (rc == 0);    rc = zmsg_addstr(msg, "");    assert (rc == 0);    rc = zmsg_addmem(msg, metadata, metasize);    assert (rc == 0);    rc = zmsg_send(&msg, self->handler);    assert (rc == 0);    return 0;}

voidzcert_print (zcert_t *self){    assert (self);    zsys_info ("zcert: metadata");    char *value = (char *) zhash_first (self->metadata);    while (value) {        zsys_info ("zcert:     %s = /"%s/"",                   zhash_cursor (self->metadata), value);        value = (char *) zhash_next (self->metadata);    }    zsys_info ("zcert: curve");    zsys_info ("zcert:     public-key = /"%s/"", self->public_txt);    zsys_info ("zcert:     secret-key = /"%s/"", self->secret_txt);}

voidzdir_watch (zsock_t *pipe, void *unused){    zdir_watch_t *watch = s_zdir_watch_new (pipe);    assert (watch);    watch->loop = zloop_new ();    assert (watch->loop);    watch->subs = zhash_new ();    assert (watch->subs);    zloop_reader (watch->loop, pipe, s_on_command, watch);    zloop_reader_set_tolerant (watch->loop, pipe); // command pipe needs to be tolerant, otherwise we'd have a hard time shutting down    s_zdir_watch_timeout (watch, 250); // default poll time of 250ms    //  Signal initialization    zsock_signal (pipe, 0);    // Dispatch above handlers    zloop_start (watch->loop);    if (watch->verbose)        zsys_info ("zdir_watch: Complete");    // signal destruction    zsock_signal (watch->pipe, 0);    // Done - cleanup and exit    s_zdir_watch_destroy (&watch);}

static voidpipe_drop_local_writer (pipe_t **self_p){    assert (self_p);    if (*self_p) {        pipe_t *self = *self_p;        //  TODO: what if self->writer is REMOTE_NODE?        self->writer = NULL;        if (self->reader) {            if (self->reader == REMOTE_NODE) {                //  Tell remote node we're dropping off                zmsg_t *msg = zmsg_new ();                zmsg_addstr (msg, "DROP WRITER");                zmsg_addstr (msg, self->name);                zyre_whisper (self->server->zyre, self->remote, &msg);                zsys_info ("%s: tell peer we stopped being writer", self->name);            }            else {                engine_send_event (self->reader, writer_dropped_event);                //  Don't destroy pipe yet - reader is still using it                *self_p = NULL;            }        }        pipe_destroy (self_p);    }}

int main (int argc, char *argv []){    //  Get number of nodes N to simulate    //  We need 3 x N x N + 3N file handles    int max_nodes = 10;    int nbr_nodes = 0;    if (argc > 1)        max_nodes = atoi (argv [1]);    int max_iterations = -1;    int nbr_iterations = 0;    if (argc > 2)        max_iterations = atoi (argv [2]);    //  We address nodes as an array of actors    zactor_t **actors = (zactor_t **) zmalloc (sizeof (zactor_t *) * max_nodes);    //  We will randomly start and stop node threads    uint index;    while (!zsys_interrupted) {        index = randof (max_nodes);        //  Toggle node thread        if (actors [index]) {            zactor_destroy (&actors [index]);            actors [index] = NULL;            zsys_info ("stopped node (%d running)", --nbr_nodes);        }        else {            actors [index] = zactor_new (node_actor, NULL);            zsys_info ("started node (%d running)", ++nbr_nodes);        }        nbr_iterations++;        if (max_iterations > 0 && nbr_iterations >= max_iterations)            break;        //  Sleep ~750 msecs randomly so we smooth out activity        zclock_sleep (randof (500) + 500);    }    zsys_info ("stopped tester (%d iterations)", nbr_iterations);    //  Stop all remaining actors    for (index = 0; index < max_nodes; index++) {        if (actors [index])            zactor_destroy (&actors [index]);    }    free (actors);    return 0;}

static voids_socket_event (agent_t *self){    //  First frame is event number and value    zframe_t *frame = zframe_recv (self->socket);    int event = *(uint16_t *) (zframe_data (frame));    int value = *(uint32_t *) (zframe_data (frame) + 2);    zframe_destroy (&frame);        //  Second frame is address    char *address = zstr_recv (self->socket);    char *description = "Unknown";    switch (event) {        case ZMQ_EVENT_ACCEPTED:            description = "Accepted";            break;        case ZMQ_EVENT_ACCEPT_FAILED:            description = "Accept failed";            break;        case ZMQ_EVENT_BIND_FAILED:            description = "Bind failed";            break;        case ZMQ_EVENT_CLOSED:            description = "Closed";            break;        case ZMQ_EVENT_CLOSE_FAILED:            description = "Close failed";            break;        case ZMQ_EVENT_DISCONNECTED:            description = "Disconnected";            break;        case ZMQ_EVENT_CONNECTED:            description = "Connected";            break;        case ZMQ_EVENT_CONNECT_DELAYED:            description = "Connect delayed";            break;        case ZMQ_EVENT_CONNECT_RETRIED:            description = "Connect retried";            break;        case ZMQ_EVENT_LISTENING:            description = "Listening";            break;        case ZMQ_EVENT_MONITOR_STOPPED:            description = "Monitor stopped";            break;        default:            zsys_error ("illegal socket monitor event: %d", event);            break;    }    if (self->verbose)        zsys_info ("zmonitor: %s - %s/n", description, address);    zstr_sendfm (self->pipe, "%d", event);    zstr_sendfm (self->pipe, "%d", value);    zstr_sendm  (self->pipe, address);    zstr_send   (self->pipe, description);    free (address);}

static bools_authenticate_gssapi (self_t *self, zap_request_t *request){    if (self->verbose)        zsys_info ("zauth: - allowed (GSSAPI) principal=%s identity=%s",                   request->principal, request->identity);    return true;}

static intclient_initialize (client_t *self){    //  Construct properties here    const char *password = zconfig_get (self->server->config, "/zmailer_server/password", "");    zsys_info ("at initialize: %s", password);    return 0;}

static voidpass_data_to_reader (client_t *self){    assert (self->pipe);    zchunk_t *chunk = zpipes_msg_get_chunk (self->request);    zsys_info ("write %d bytes", (int) zchunk_size (chunk));    pipe_send_data (self->pipe, &chunk);}

static ints_zdir_watch_timeout (zdir_watch_t *watch, int timeout){    if (watch->verbose)        zsys_info ("zdir_watch: Setting directory poll timeout to %d", timeout);    if (watch->read_timer_id != -1) {        zloop_timer_end (watch->loop, watch->read_timer_id);        watch->read_timer_id = -1;    }    watch->read_timer_id = zloop_timer (watch->loop, timeout, 0, s_on_read_timer, watch);    if (watch->verbose)        zsys_info ("zdir_watch: Successfully set directory poll timeout to %d", timeout);    return 0;}

static ints_self_handle_pipe (self_t *self){    //  Get the whole message off the pipe in one go    zmsg_t *request = zmsg_recv (self->pipe);    if (!request)        return -1;                  //  Interrupted    char *command = zmsg_popstr (request);    if (!command) {        s_self_destroy (&self);        return -1;    }    if (self->verbose)        zsys_info ("zmonitor: API command=%s", command);    if (streq (command, "LISTEN")) {        char *event = zmsg_popstr (request);        while (event) {            if (self->verbose)                zsys_info ("zmonitor: - listening to event=%s", event);            s_self_listen (self, event);            zstr_free (&event);            event = zmsg_popstr (request);        }    }    else    if (streq (command, "START")) {        s_self_start (self);        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "VERBOSE"))        self->verbose = true;    else    if (streq (command, "$TERM"))        self->terminated = true;    else {        zsys_error ("zmonitor: - invalid command: %s", command);        assert (false);    }    zstr_free (&command);    zmsg_destroy (&request);    return 0;}