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

static voids_self_switch (self_t *self, zsock_t *input, zsock_t *output){    //  We use the low-level libzmq API for best performance    void *zmq_input = zsock_resolve (input);    void *zmq_output = zsock_resolve (output);    void *zmq_capture = self->capture ? zsock_resolve (self->capture) : NULL;    zmq_msg_t msg;    zmq_msg_init (&msg);    while (true) {        if (zmq_recvmsg (zmq_input, &msg, ZMQ_DONTWAIT) == -1)            break;      //  Presumably EAGAIN        int send_flags = zsock_rcvmore (zmq_input) ? ZMQ_SNDMORE : 0;        if (zmq_capture) {            zmq_msg_t dup;            zmq_msg_init (&dup);            zmq_msg_copy (&dup, &msg);            if (zmq_sendmsg (zmq_capture, &dup, send_flags) == -1)                zmq_msg_close (&dup);        }        if (zmq_sendmsg (zmq_output, &msg, send_flags) == -1) {            zmq_msg_close (&msg);            break;        }    }}

intzmailer_msg_send (zmailer_msg_t *self, zsock_t *output){    assert (self);    assert (output);    if (zsock_type (output) == ZMQ_ROUTER)        zframe_send (&self->routing_id, output, ZFRAME_MORE + ZFRAME_REUSE);    size_t frame_size = 2 + 1;          //  Signature and message ID    switch (self->id) {        case ZMAILER_MSG_MAIL:            frame_size += 2;            //  version            frame_size += 1 + strlen (self->from);            frame_size += 4;            if (self->to)                frame_size += strlen (self->to);            frame_size += 4;            if (self->subject)                frame_size += strlen (self->subject);            frame_size += 4;            if (self->request)                frame_size += strlen (self->request);            break;    }    //  Now serialize message into the frame    zmq_msg_t frame;    zmq_msg_init_size (&frame, frame_size);    self->needle = (byte *) zmq_msg_data (&frame);    PUT_NUMBER2 (0xAAA0 | 0);    PUT_NUMBER1 (self->id);    size_t nbr_frames = 1;              //  Total number of frames to send    switch (self->id) {        case ZMAILER_MSG_MAIL:            PUT_NUMBER2 (1);            PUT_STRING (self->from);            if (self->to) {                PUT_LONGSTR (self->to);            }            else                PUT_NUMBER4 (0);    //  Empty string            if (self->subject) {                PUT_LONGSTR (self->subject);            }            else                PUT_NUMBER4 (0);    //  Empty string            if (self->request) {                PUT_LONGSTR (self->request);            }            else                PUT_NUMBER4 (0);    //  Empty string            break;    }    //  Now send the data frame    zmq_msg_send (&frame, zsock_resolve (output), --nbr_frames? ZMQ_SNDMORE: 0);    return 0;}

intzmsg_send (zmsg_t **self_p, void *dest){    assert (self_p);    assert (dest);    zmsg_t *self = *self_p;    int rc = 0;    void *handle = zsock_resolve (dest);    if (self) {        assert (zmsg_is (self));        if (zlist_size (self->frames) == 0)            return -1;          //  Sending an empty message is an error                zframe_t *frame = (zframe_t *) zlist_pop (self->frames);        while (frame) {            rc = zframe_send (&frame, handle,                zlist_size (self->frames)? ZFRAME_MORE: 0);            if (rc != 0)                break;            frame = (zframe_t *) zlist_pop (self->frames);        }        zmsg_destroy (self_p);    }    return rc;}

intzpoller_remove (zpoller_t *self, void *reader){    assert (self);    assert (reader);    int rc = 0;#ifdef ZMQ_HAVE_POLLER    void *socket = zsock_resolve (reader);    if (socket)        rc = zmq_poller_remove (self->zmq_poller, socket);    else        rc = zmq_poller_remove_fd (self->zmq_poller, *(SOCKET *) reader);#else    size_t num_readers_before = zlist_size (self->reader_list);    zlist_remove (self->reader_list, reader); // won't fail with non-existent reader    size_t num_readers_after = zlist_size (self->reader_list);    if (num_readers_before != num_readers_after)        self->need_rebuild = true;    else {        errno = EINVAL;        rc    = -1;    }#endif    return rc;}

intzstr_recvx (void *source, char **string_p, ...){    assert (source);    void *handle = zsock_resolve (source);    zmsg_t *msg = zmsg_recv (handle);    if (!msg)        return -1;    //  Filter a signal that may come from a dying actor    if (zmsg_signal (msg) >= 0) {        zmsg_destroy (&msg);        return -1;    }    int count = 0;    va_list args;    va_start (args, string_p);    while (string_p) {        *string_p = zmsg_popstr (msg);        string_p = va_arg (args, char **);        count++;    }    va_end (args);    zmsg_destroy (&msg);    return count;}

intmdp_client_getsockopt (mdp_client_t *self, 	int option, void *optval, size_t *optvallen){    assert (self);    assert (self->client);    return zmq_getsockopt (zsock_resolve(self->client), option, optval, optvallen);}

static ints_rebuild_poll_set (zpoller_t *self){    freen (self->poll_set);    self->poll_set = NULL;    freen (self->poll_readers);    self->poll_readers = NULL;    self->poll_size = zlist_size (self->reader_list);    self->poll_set = (zmq_pollitem_t *)                      zmalloc (self->poll_size * sizeof (zmq_pollitem_t));    self->poll_readers = (void **) zmalloc (self->poll_size * sizeof (void *));    if (!self->poll_set || !self->poll_readers)        return -1;    uint reader_nbr = 0;    void *reader = zlist_first (self->reader_list);    while (reader) {        self->poll_readers [reader_nbr] = reader;        void *socket = zsock_resolve (reader);        if (socket == NULL) {            self->poll_set [reader_nbr].socket = NULL;            self->poll_set [reader_nbr].fd = *(SOCKET *) reader;        }        else            self->poll_set [reader_nbr].socket = socket;        self->poll_set [reader_nbr].events = ZMQ_POLLIN;        reader_nbr++;        reader = zlist_next (self->reader_list);    }    self->need_rebuild = false;    return 0;}

intzgossip_msg_send (zgossip_msg_t **self_p, void *output){    assert (self_p);    assert (*self_p);    assert (output);    //  Save routing_id if any, as encode will destroy it    zgossip_msg_t *self = *self_p;    zframe_t *routing_id = self->routing_id;    self->routing_id = NULL;    //  Encode zgossip_msg message to a single zmsg    zmsg_t *msg = zgossip_msg_encode (&self);        //  If we're sending to a ROUTER, send the routing_id first    if (zsocket_type (zsock_resolve (output)) == ZMQ_ROUTER) {        assert (routing_id);        zmsg_prepend (msg, &routing_id);    }    else        zframe_destroy (&routing_id);            if (msg && zmsg_send (&msg, output) == 0)        return 0;    else        return -1;              //  Failed to encode, or send}

void *zactor_resolve (void *self){    assert (self);    if (zactor_is (self))        return zsock_resolve (((zactor_t *) self)->pipe);    else        return self;}

static self_t *s_self_new (zsock_t *pipe, zsock_t *sock){    self_t *self = (self_t *) zmalloc (sizeof (self_t));    self->pipe = pipe;    self->monitored = zsock_resolve (sock);    self->poller = zpoller_new (self->pipe, NULL);    return self;}

zgossip_msg_t *zgossip_msg_recv_nowait (void *input){    assert (input);    zmsg_t *msg = zmsg_recv_nowait (input);    //  If message came from a router socket, first frame is routing_id    zframe_t *routing_id = NULL;    if (zsocket_type (zsock_resolve (input)) == ZMQ_ROUTER) {        routing_id = zmsg_pop (msg);        //  If message was not valid, forget about it        if (!routing_id || !zmsg_next (msg))            return NULL;        //  Malformed or empty    }    zgossip_msg_t * zgossip_msg = zgossip_msg_decode (&msg);    if (zsocket_type (zsock_resolve (input)) == ZMQ_ROUTER)        zgossip_msg->routing_id = routing_id;    return zgossip_msg;}

voidzcert_apply (zcert_t *self, void *zocket){    assert (self);#if (ZMQ_VERSION_MAJOR == 4)    void *handle = zsock_resolve (zocket);    if (zsys_has_curve ()) {        zsocket_set_curve_secretkey_bin (handle, self->secret_key);        zsocket_set_curve_publickey_bin (handle, self->public_key);    }#endif}

static self_t *s_self_new (zsock_t *pipe, void *sock){    self_t *self = (self_t *) zmalloc (sizeof (self_t));    if (!self)        return NULL;    self->pipe = pipe;    self->monitored = zsock_resolve (sock);    self->poller = zpoller_new (self->pipe, NULL);    if (!self->poller)        s_self_destroy (&self);    return self;}

voidzyre_peer_connect (zyre_peer_t *self, zuuid_t *from, const char *endpoint){    assert (self);    assert (!self->connected);    //  Create new outgoing socket (drop any messages in transit)    self->mailbox = zsock_new (ZMQ_DEALER);    if (!self->mailbox)        return;             //  Null when we're shutting down        //  Set our own identity on the socket so that receiving node    //  knows who each message came from. Note that we cannot use    //  the UUID directly as the identity since it may contain a    //  zero byte at the start, which libzmq does not like for    //  historical and arguably bogus reasons that it nonetheless    //  enforces.    byte routing_id [ZUUID_LEN + 1] = { 1 };    memcpy (routing_id + 1, zuuid_data (from), ZUUID_LEN);    int rc = zmq_setsockopt (zsock_resolve (self->mailbox),                             ZMQ_IDENTITY, routing_id, ZUUID_LEN + 1);    assert (rc == 0);    //  Set a high-water mark that allows for reasonable activity    zsock_set_sndhwm (self->mailbox, PEER_EXPIRED * 100);    //  Send messages immediately or return EAGAIN    zsock_set_sndtimeo (self->mailbox, 0);    //  Connect through to peer node    rc = zsock_connect (self->mailbox, "%s", endpoint);    if (rc != 0) {        zsys_error ("(%s) cannot connect to endpoint=%s",                    self->origin, endpoint);        //  Don't really have any error handling yet; if connect        //  fails, there's something wrong with connect endpoint?        assert (false);    }    assert (rc == 0);    if (self->verbose)        zsys_info ("(%s) connect to peer: endpoint=%s",                   self->origin, endpoint);    self->endpoint = strdup (endpoint);    self->connected = true;    self->ready = false;}

static ints_send_string (void *dest, bool more, char *string){    assert (dest);    void *handle = zsock_resolve (dest);    size_t len = strlen (string);    zmq_msg_t message;    zmq_msg_init_size (&message, len);    memcpy (zmq_msg_data (&message), string, len);    if (zmq_sendmsg (handle, &message, more ? ZMQ_SNDMORE : 0) == -1) {        zmq_msg_close (&message);        return -1;    }    else        return 0;}

char *zstr_recv_nowait (void *dest){    assert (dest);    void *handle = zsock_resolve (dest);    zmq_msg_t message;    zmq_msg_init (&message);    if (zmq_recvmsg (handle, &message, ZMQ_DONTWAIT) < 0)        return NULL;    size_t size = zmq_msg_size (&message);    char *string = (char *) malloc (size + 1);    memcpy (string, zmq_msg_data (&message), size);    zmq_msg_close (&message);    string [size] = 0;    return string;}

char *zstr_recv (void *source){    assert (source);    void *handle = zsock_resolve (source);    zmq_msg_t message;    zmq_msg_init (&message);    if (zmq_recvmsg (handle, &message, 0) < 0)        return NULL;    size_t size = zmq_msg_size (&message);    char *string = (char *) malloc (size + 1);    memcpy (string, zmq_msg_data (&message), size);    zmq_msg_close (&message);    string [size] = 0;    return string;}

intzpoller_add (zpoller_t *self, void *reader){    assert (self);    assert (reader);    int rc = 0;#ifdef ZMQ_HAVE_POLLER    void *socket = zsock_resolve (reader);    if (socket)        rc = zmq_poller_add (self->zmq_poller, socket, reader, ZMQ_POLLIN);    else        rc = zmq_poller_add_fd (self->zmq_poller, *(SOCKET *) reader, reader, ZMQ_POLLIN);#else    zlist_append (self->reader_list, reader);    self->need_rebuild = true;#endif    return rc;}

static ints_rebuild_pollset (zloop_t *self){    free (self->pollset);    free (self->readact);    free (self->pollact);    self->pollset = NULL;    self->readact = NULL;    self->pollact = NULL;    self->poll_size = zlistx_size (self->readers) + zlistx_size (self->pollers);    self->pollset = (zmq_pollitem_t *) zmalloc (self->poll_size * sizeof (zmq_pollitem_t));    if (!self->pollset)        return -1;    self->readact = (s_reader_t *) zmalloc (self->poll_size * sizeof (s_reader_t));    if (!self->readact)        return -1;    self->pollact = (s_poller_t *) zmalloc (self->poll_size * sizeof (s_poller_t));    if (!self->pollact)        return -1;    s_reader_t *reader = (s_reader_t *) zlistx_first (self->readers);    uint item_nbr = 0;    while (reader) {        zmq_pollitem_t poll_item = { zsock_resolve (reader->sock), 0, ZMQ_POLLIN };        self->pollset [item_nbr] = poll_item;        self->readact [item_nbr] = *reader;        item_nbr++;        reader = (s_reader_t *) zlistx_next (self->readers);    }    s_poller_t *poller = (s_poller_t *) zlistx_first (self->pollers);    while (poller) {        self->pollset [item_nbr] = poller->item;        self->pollact [item_nbr] = *poller;        item_nbr++;        poller = (s_poller_t *) zlistx_next (self->pollers);    }    self->need_rebuild = false;    return 0;}

intzstr_recvx (void *source, char **string_p, ...){    assert (source);    void *handle = zsock_resolve (source);        zmsg_t *msg = zmsg_recv (handle);    if (!msg)        return -1;            va_list args;    va_start (args, string_p);    while (string_p) {        *string_p = zmsg_popstr (msg);        string_p = va_arg (args, char **);    }    va_end (args);    zmsg_destroy (&msg);    return 0;}

voidzbeacon (zsock_t *pipe, void *args){    self_t *self = s_self_new (pipe);    assert (self);    //  Signal successful initialization    zsock_signal (pipe, 0);    while (!self->terminated) {        //  Poll on API pipe and on UDP socket        zmq_pollitem_t pollitems [] = {            { zsock_resolve (self->pipe), 0, ZMQ_POLLIN, 0 },            { NULL, self->udpsock, ZMQ_POLLIN, 0 }        };        long timeout = -1;        if (self->transmit) {            timeout = (long) (self->ping_at - zclock_mono ());            if (timeout < 0)                timeout = 0;        }        int pollset_size = self->udpsock? 2: 1;        if (zmq_poll (pollitems, pollset_size, timeout * ZMQ_POLL_MSEC) == -1)            break;              //  Interrupted        if (pollitems [0].revents & ZMQ_POLLIN)            s_self_handle_pipe (self);        if (pollitems [1].revents & ZMQ_POLLIN)            s_self_handle_udp (self);        if (self->transmit        &&  zclock_mono () >= self->ping_at) {            //  Send beacon to any listening peers            if (zsys_udp_send (self->udpsock, self->transmit, &self->broadcast, sizeof (inaddr_t)))                //  Try to recreate UDP socket on interface                s_self_prepare_udp (self);            self->ping_at = zclock_mono () + self->interval;        }    }    s_self_destroy (&self);}

intzpubsub_filter_send (zpubsub_filter_t *self, zsock_t *output){    assert (self);    assert (output);    if (zsock_type (output) == ZMQ_ROUTER)        zframe_send (&self->routing_id, output, ZFRAME_MORE + ZFRAME_REUSE);    size_t frame_size = 2 + 1;          //  Signature and message ID    switch (self->id) {        case ZPUBSUB_FILTER_FILTER:            frame_size += 2;            //  magic            frame_size += 2;            //  version            frame_size += 1 + strlen (self->partition);            frame_size += 1 + strlen (self->topic);            break;    }    //  Now serialize message into the frame    zmq_msg_t frame;    zmq_msg_init_size (&frame, frame_size);    self->needle = (byte *) zmq_msg_data (&frame);    PUT_NUMBER2 (0xAAA0 | 7);    PUT_NUMBER1 (self->id);    size_t nbr_frames = 1;              //  Total number of frames to send        switch (self->id) {        case ZPUBSUB_FILTER_FILTER:            PUT_NUMBER2 (ZPUBSUB_FILTER_MAGIC_NUMBER);            PUT_NUMBER2 (ZPUBSUB_FILTER_VERSION);            PUT_STRING (self->partition);            PUT_STRING (self->topic);            break;    }    //  Now send the data frame    zmq_msg_send (&frame, zsock_resolve (output), --nbr_frames? ZMQ_SNDMORE: 0);        return 0;}

//  --------------------------------------------------------------------------//  Send (More) message to destination socket, and destroy the message after sending//  it successfully. If the message has no frames, sends nothing but destroys//  the message anyhow. Nullifies the caller's reference to the message (as//  it is a destructor).CZMQ_EXPORT int    zmsg_sendm (zmsg_t **self_p, void *dest){    assert (self_p);    assert (dest);    zmsg_t *self = *self_p;    int rc = 0;    void *handle = zsock_resolve (dest);    if (self) {        assert (zmsg_is (self));        zframe_t *frame = (zframe_t *) zlist_pop (self->frames);        while (frame) {            rc = zframe_send (&frame, handle,ZFRAME_MORE);            if (rc != 0)                break;            frame = (zframe_t *) zlist_pop (self->frames);        }        if (rc == 0)            zmsg_destroy (self_p);    }    return rc;}

zmsg_t *zmsg_recv (void *source){    assert (source);    zmsg_t *self = zmsg_new ();    if (!self)        return NULL;    void *handle = zsock_resolve (source);    while (true) {        zframe_t *frame = zframe_recv (handle);        if (!frame) {            zmsg_destroy (&self);            break;              //  Interrupted or terminated        }        if (zmsg_append (self, &frame)) {            zmsg_destroy (&self);            break;        }        if (!zsock_rcvmore (handle))            break;              //  Last message frame    }    return self;}

intzmailer_msg_recv (zmailer_msg_t *self, zsock_t *input){    assert (input);    if (zsock_type (input) == ZMQ_ROUTER) {        zframe_destroy (&self->routing_id);        self->routing_id = zframe_recv (input);        if (!self->routing_id || !zsock_rcvmore (input)) {            zsys_warning ("zmailer_msg: no routing ID");            return -1;          //  Interrupted or malformed        }    }    zmq_msg_t frame;    zmq_msg_init (&frame);    int size = zmq_msg_recv (&frame, zsock_resolve (input), 0);    if (size == -1) {        zsys_warning ("zmailer_msg: interrupted");        goto malformed;         //  Interrupted    }    //  Get and check protocol signature    self->needle = (byte *) zmq_msg_data (&frame);    self->ceiling = self->needle + zmq_msg_size (&frame);    uint16_t signature;    GET_NUMBER2 (signature);    if (signature != (0xAAA0 | 0)) {        zsys_warning ("zmailer_msg: invalid signature");        //  TODO: discard invalid messages and loop, and return        //  -1 only on interrupt        goto malformed;         //  Interrupted    }    //  Get message id and parse per message type    GET_NUMBER1 (self->id);    switch (self->id) {        case ZMAILER_MSG_MAIL:            {                uint16_t version;                GET_NUMBER2 (version);                if (version != 1) {                    zsys_warning ("zmailer_msg: version is invalid");                    goto malformed;                }            }            GET_STRING (self->from);            GET_LONGSTR (self->to);            GET_LONGSTR (self->subject);            GET_LONGSTR (self->request);            break;        default:            zsys_warning ("zmailer_msg: bad message ID");            goto malformed;    }    //  Successful return    zmq_msg_close (&frame);    return 0;    //  Error returns    malformed:        zsys_warning ("zmailer_msg: zmailer_msg malformed message, fail");        zmq_msg_close (&frame);        return -1;              //  Invalid message}

intzpubsub_filter_recv (zpubsub_filter_t *self, zsock_t *input){    assert (input);        if (zsock_type (input) == ZMQ_ROUTER) {        zframe_destroy (&self->routing_id);        self->routing_id = zframe_recv (input);        if (!self->routing_id || !zsock_rcvmore (input)) {            zsys_warning ("zpubsub_filter: no routing ID");            return -1;          //  Interrupted or malformed        }    }    zmq_msg_t frame;    zmq_msg_init (&frame);    int size = zmq_msg_recv (&frame, zsock_resolve (input), 0);    if (size == -1) {        zsys_warning ("zpubsub_filter: interrupted");        goto malformed;         //  Interrupted    }    //  Get and check protocol signature    self->needle = (byte *) zmq_msg_data (&frame);    self->ceiling = self->needle + zmq_msg_size (&frame);        uint16_t signature;    GET_NUMBER2 (signature);    if (signature != (0xAAA0 | 7)) {        zsys_warning ("zpubsub_filter: invalid signature");        //  TODO: discard invalid messages and loop, and return        //  -1 only on interrupt        goto malformed;         //  Interrupted    }    //  Get message id and parse per message type    GET_NUMBER1 (self->id);    switch (self->id) {        case ZPUBSUB_FILTER_FILTER:            {                uint16_t magic;                GET_NUMBER2 (magic);                if (magic != ZPUBSUB_FILTER_MAGIC_NUMBER) {                    zsys_warning ("zpubsub_filter: magic is invalid");                    goto malformed;                }            }            {                uint16_t version;                GET_NUMBER2 (version);                if (version != ZPUBSUB_FILTER_VERSION) {                    zsys_warning ("zpubsub_filter: version is invalid");                    goto malformed;                }            }            GET_STRING (self->partition);            GET_STRING (self->topic);            break;        default:            zsys_warning ("zpubsub_filter: bad message ID");            goto malformed;    }    //  Successful return    zmq_msg_close (&frame);    return 0;    //  Error returns    malformed:        zsys_warning ("zpubsub_filter: zpubsub_filter malformed message, fail");        zmq_msg_close (&frame);        return -1;              //  Invalid message}

intzgossip_msg_send (zgossip_msg_t *self, zsock_t *output){    assert (self);    assert (output);    if (zsock_type (output) == ZMQ_ROUTER)        zframe_send (&self->routing_id, output, ZFRAME_MORE + ZFRAME_REUSE);    size_t frame_size = 2 + 1;          //  Signature and message ID    switch (self->id) {        case ZGOSSIP_MSG_HELLO:            frame_size += 1;            //  version            break;        case ZGOSSIP_MSG_PUBLISH:            frame_size += 1;            //  version            frame_size += 1 + strlen (self->key);            frame_size += 4;            if (self->value)                frame_size += strlen (self->value);            frame_size += 4;            //  ttl            break;        case ZGOSSIP_MSG_PING:            frame_size += 1;            //  version            break;        case ZGOSSIP_MSG_PONG:            frame_size += 1;            //  version            break;        case ZGOSSIP_MSG_INVALID:            frame_size += 1;            //  version            break;    }    //  Now serialize message into the frame    zmq_msg_t frame;    zmq_msg_init_size (&frame, frame_size);    self->needle = (byte *) zmq_msg_data (&frame);    PUT_NUMBER2 (0xAAA0 | 0);    PUT_NUMBER1 (self->id);    size_t nbr_frames = 1;              //  Total number of frames to send    switch (self->id) {        case ZGOSSIP_MSG_HELLO:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_PUBLISH:            PUT_NUMBER1 (1);            PUT_STRING (self->key);            if (self->value) {                PUT_LONGSTR (self->value);            }            else                PUT_NUMBER4 (0);    //  Empty string            PUT_NUMBER4 (self->ttl);            break;        case ZGOSSIP_MSG_PING:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_PONG:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_INVALID:            PUT_NUMBER1 (1);            break;    }    //  Now send the data frame    zmq_msg_send (&frame, zsock_resolve (output), --nbr_frames? ZMQ_SNDMORE: 0);    return 0;}

/////  Probe the supplied reference. If it looks like a zsock_t instance, return//  the underlying libzmq socket handle; else if it looks like a file        //  descriptor, return NULL; else if it looks like a libzmq socket handle,   //  return the supplied value. Takes a polymorphic socket reference.         void * QZsock::resolve (void *self){    void * rv = zsock_resolve (self);    return rv;}

intzyre_peer_connect (zyre_peer_t *self, zuuid_t *from, const char *endpoint, uint64_t expired_timeout){    assert (self);    assert (!self->connected);    //  Create new outgoing socket (drop any messages in transit)    self->mailbox = zsock_new (ZMQ_DEALER);    if (!self->mailbox)        return -1;             //  Null when we're shutting down    //  Set our own identity on the socket so that receiving node    //  knows who each message came from. Note that we cannot use    //  the UUID directly as the identity since it may contain a    //  zero byte at the start, which libzmq does not like for    //  historical and arguably bogus reasons that it nonetheless    //  enforces.    byte routing_id [ZUUID_LEN + 1] = { 1 };    memcpy (routing_id + 1, zuuid_data (from), ZUUID_LEN);    int rc = zmq_setsockopt (zsock_resolve (self->mailbox),                             ZMQ_IDENTITY, routing_id, ZUUID_LEN + 1);    assert (rc == 0);    //  Set a high-water mark that allows for reasonable activity    zsock_set_sndhwm (self->mailbox, expired_timeout * 100);    //  Send messages immediately or return EAGAIN    zsock_set_sndtimeo (self->mailbox, 0);    //  If the peer is a link-local IPv6 address but the interface is not set,    //  use ZSYS_INTERFACE_ADDRESS if provided    zrex_t *rex = zrex_new (NULL);    char endpoint_iface [NI_MAXHOST] = {0};    if (zsys_ipv6 () && zsys_interface () && strlen(zsys_interface ()) &&            !streq (zsys_interface (), "*") &&            zrex_eq (rex, endpoint, "^tcp://(fe80[^%]+)(://d+)$")) {        const char *hostname, *port;        zrex_fetch (rex, &hostname, &port, NULL);        strcat (endpoint_iface, "tcp://");        strcat (endpoint_iface, hostname);        strcat (endpoint_iface, "%");        strcat (endpoint_iface, zsys_interface ());        strcat (endpoint_iface, port);    } else        strcat (endpoint_iface, endpoint);    zrex_destroy (&rex);    //  Connect through to peer node    rc = zsock_connect (self->mailbox, "%s", endpoint_iface);    if (rc != 0) {        zsys_debug ("(%s) cannot connect to endpoint=%s",                    self->origin, endpoint_iface);        zsock_destroy (&self->mailbox);        return -1;    }    if (self->verbose)        zsys_info ("(%s) connect to peer: endpoint=%s",                   self->origin, endpoint_iface);    self->endpoint = strdup (endpoint_iface);    self->connected = true;    self->ready = false;    return 0;}