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

static zsock_t *s_create_socket (char *type_name, char *endpoints){    //  This array matches ZMQ_XXX type definitions    assert (ZMQ_PAIR == 0);    char *type_names [] = {        "PAIR", "PUB", "SUB", "REQ", "REP",        "DEALER", "ROUTER", "PULL", "PUSH",        "XPUB", "XSUB", type_name    };    //  We always match type at least at end of table    int index;    for (index = 0; strneq (type_name, type_names [index]); index++) ;    if (index > ZMQ_XSUB) {        zsys_error ("zproxy: invalid socket type '%s'", type_name);        return NULL;    }    zsock_t *sock = zsock_new (index);    if (sock) {        if (zsock_attach (sock, endpoints, true)) {            zsys_error ("zproxy: invalid endpoints '%s'", endpoints);            zsock_destroy (&sock);        }    }    return sock;}

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);}

//  atexit or manual termination for the processvoidzsys_shutdown (void){    if (!s_initialized) {        return;    }    s_initialized = false;    //  The atexit handler is called when the main function exits;    //  however we may have zactor threads shutting down and still    //  trying to close their sockets. So if we suspect there are    //  actors busy (s_open_sockets > 0), then we sleep for a few    //  hundred milliseconds to allow the actors, if any, to get in    //  and close their sockets.    ZMUTEX_LOCK (s_mutex);    size_t busy = s_open_sockets;    ZMUTEX_UNLOCK (s_mutex);    if (busy)        zclock_sleep (200);    //  No matter, we are now going to shut down    //  Print the source reference for any sockets the app did not    //  destroy properly.    ZMUTEX_LOCK (s_mutex);    s_sockref_t *sockref = (s_sockref_t *) zlist_pop (s_sockref_list);    while (sockref) {        assert (sockref->filename);        zsys_error ("dangling '%s' socket created at %s:%d",                    zsys_sockname (sockref->type),                    sockref->filename, (int) sockref->line_nbr);        zmq_close (sockref->handle);        free (sockref);        sockref = (s_sockref_t *) zlist_pop (s_sockref_list);    }    zlist_destroy (&s_sockref_list);    ZMUTEX_UNLOCK (s_mutex);    //  Close logsender socket if opened (don't do this in critical section)    if (s_logsender) {        zsys_close (s_logsender, NULL, 0);        s_logsender = NULL;    }    if (s_open_sockets == 0)        zmq_term (s_process_ctx);    else        zsys_error ("dangling sockets: cannot terminate ZMQ safely");    ZMUTEX_DESTROY (s_mutex);    //  Free dynamically allocated properties    free (s_interface);    free (s_logident);#if defined (__UNIX__)    closelog ();                //  Just to be pedantic#endif}

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 zsock_t *s_self_create_socket (self_t *self, char *type_name, char *endpoints, proxy_socket selected_socket){    //  This array matches ZMQ_XXX type definitions    assert (ZMQ_PAIR == 0);    char *type_names [] = {        "PAIR", "PUB", "SUB", "REQ", "REP",        "DEALER", "ROUTER", "PULL", "PUSH",        "XPUB", "XSUB", type_name    };    //  We always match type at least at end of table    int index;    for (index = 0; strneq (type_name, type_names [index]); index++) ;    if (index > ZMQ_XSUB) {        zsys_error ("zproxy: invalid socket type '%s'", type_name);        return NULL;    }    zsock_t *sock = zsock_new (index);    if (sock) {#if (ZMQ_VERSION_MAJOR == 4)        if (self->domain [selected_socket]) {            // Apply authentication domain            zsock_set_zap_domain (sock, self->domain [selected_socket]);        }        if (self->auth_type [selected_socket] == AUTH_PLAIN) {            // Enable plain authentication            zsock_set_plain_server (sock, 1);        }        else        if (self->auth_type [selected_socket] == AUTH_CURVE) {            // Apply certificate keys            char *public_key = self->public_key [selected_socket];            assert(public_key);            char *secret_key = self->secret_key [selected_socket];            assert(secret_key);            zsock_set_curve_publickey (sock, public_key);            zsock_set_curve_secretkey (sock, secret_key);            // Enable curve authentication            zsock_set_curve_server (sock, 1);        }#endif        if (zsock_attach (sock, endpoints, true)) {            zsys_error ("zproxy: invalid endpoints '%s'", endpoints);            zsock_destroy (&sock);        }    }    return sock;}

static voidsomeactor_recv_api (someactor_t *self){//  Get the whole message of the pipe in one go    zmsg_t *request = zmsg_recv (self->pipe);    if (!request)       return;        //  Interrupted    char *command = zmsg_popstr (request);    if (streq (command, "START"))        zsock_signal (self->pipe, someactor_start (self));    else    if (streq (command, "STOP"))        zsock_signal (self->pipe, someactor_stop (self));    else    if (streq (command, "VERBOSE")) {        self->verbose = true;        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "$TERM"))        //  The $TERM command is send by zactor_destroy() method        self->terminated = true;    else {        zsys_error ("invalid command '%s'", command);        assert (false);    }}

intzloop_timer (zloop_t *self, size_t delay, size_t times, zloop_timer_fn handler, void *arg){    assert (self);    //  Catch excessive use of timers    if (self->max_timers && zlistx_size (self->timers) == self->max_timers) {        zsys_error ("zloop: timer limit reached (max=%d)", self->max_timers);        return -1;    }    int timer_id = s_next_timer_id (self);    s_timer_t *timer = s_timer_new (timer_id, delay, times, handler, arg);    if (timer) {        timer->list_handle = zlistx_add_end (self->timers, timer);        if (!timer->list_handle) {            s_timer_destroy (&timer);            return -1;        }        if (self->verbose)            zsys_debug ("zloop: register timer id=%d delay=%d times=%d",                        timer_id, (int) delay, (int) times);        return timer_id;    }    else        return -1;}

static voidserver_connect (server_t *self, const char *endpoint){    zsock_t *remote = zsock_new (ZMQ_DEALER);    assert (remote);          //  No recovery if exhausted    //  Never block on sending; we use an infinite HWM and buffer as many    //  messages as needed in outgoing pipes. Note that the maximum number    //  is the overall tuple set size.    zsock_set_sndhwm (remote, 0);    if (zsock_connect (remote, "%s", endpoint)) {        zsys_error ("bad zgossip endpoint '%s'", endpoint);        zsock_destroy (&remote);        return;    }    //  Send HELLO and then PUBLISH for each tuple we have    zgossip_msg_send_hello (remote);    tuple_t *tuple = (tuple_t *) zhash_first (self->tuples);    while (tuple) {        int rc = zgossip_msg_send_publish (remote, tuple->key, tuple->value, 0);        assert (rc == 0);        tuple = (tuple_t *) zhash_next (self->tuples);    }    //  Now monitor this remote for incoming messages    engine_handle_socket (self, remote, remote_handler);    zlist_append (self->remotes, remote);}

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

static zmsg_t *server_method (server_t *self, const char *method, zmsg_t *msg){    //  Connect to a remote    zmsg_t *reply = NULL;    if (streq (method, "CONNECT")) {        char *endpoint = zmsg_popstr (msg);        assert (endpoint);        server_connect (self, endpoint);        zstr_free (&endpoint);    }    else    if (streq (method, "PUBLISH")) {        char *key = zmsg_popstr (msg);        char *value = zmsg_popstr (msg);        server_accept (self, key, value);        zstr_free (&key);        zstr_free (&value);    }    else    if (streq (method, "STATUS")) {        //  Return number of tuples we have stored        reply = zmsg_new ();        assert (reply);        zmsg_addstr (reply, "STATUS");        zmsg_addstrf (reply, "%d", (int) zhashx_size (self->tuples));    }    else        zsys_error ("unknown zgossip method '%s'", method);    return reply;}

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 voids_self_configure (self_t *self, int port_nbr){    assert (port_nbr);    self->port_nbr = port_nbr;    s_self_prepare_udp (self);    zstr_send (self->pipe, self->hostname);    if (streq (self->hostname, ""))        zsys_error ("No broadcast interface found, (ZSYS_INTERFACE=%s)", zsys_interface ());}

voidzsys_set_max_sockets (size_t max_sockets){    zsys_init ();    ZMUTEX_LOCK (s_mutex);    //  If the app is misusing this method, burn it with fire    if (s_open_sockets)        zsys_error ("zsys_max_sockets() is not valid after creating sockets");    assert (s_open_sockets == 0);    s_max_sockets = max_sockets ? max_sockets : zsys_socket_limit ();    ZMUTEX_UNLOCK (s_mutex);}

static ints_self_handle_pipe (self_t *self){    //  Get just the command off the pipe    char *command = zstr_recv (self->pipe);    if (!command)        return -1;                  //  Interrupted    if (self->verbose)        zsys_info ("zbeacon: API command=%s", command);    if (streq (command, "VERBOSE"))        self->verbose = true;    else    if (streq (command, "CONFIGURE")) {        int port;        int rc = zsock_recv (self->pipe, "i", &port);        assert (rc == 0);        s_self_configure (self, port);    }    else    if (streq (command, "PUBLISH")) {        zframe_destroy (&self->transmit);        zsock_recv (self->pipe, "fi", &self->transmit, &self->interval);        assert (zframe_size (self->transmit) <= UDP_FRAME_MAX);        if (self->interval == 0)            self->interval = INTERVAL_DFLT;        //  Start broadcasting immediately        self->ping_at = zclock_mono ();    }    else    if (streq (command, "SILENCE"))        zframe_destroy (&self->transmit);    else    if (streq (command, "SUBSCRIBE")) {        zframe_destroy (&self->filter);        self->filter = zframe_recv (self->pipe);        assert (zframe_size (self->filter) <= UDP_FRAME_MAX);    }    else    if (streq (command, "UNSUBSCRIBE"))        zframe_destroy (&self->filter);    else    if (streq (command, "$TERM"))        self->terminated = true;    else {        zsys_error ("zbeacon: - invalid command: %s", command);        assert (false);    }    zstr_free (&command);    return 0;}

static voids_upstream_handle_pipe (upstream_t *self){    char *command = zstr_recv (self->pipe);    if (streq (command, "$TERM")) {        self->terminated = true;    }    else    if (streq (command, "CONNECT")) {        char *endpoint = zstr_recv (self->pipe);        int rc = zsock_connect (self->push, "%s", endpoint);        if (rc != -1)            self->connected = true;        zstr_free (&endpoint);        zsock_bsend (self->pipe, "2", rc);    }    else    if (streq (command, "BIND")) {        char *endpoint = zstr_recv (self->pipe);        int rc = zsock_bind (self->push, "%s", endpoint);        if (rc != -1)            self->connected = true;        zstr_free (&endpoint);        zsock_bsend (self->pipe, "2", rc);    }    else    if (streq (command, "SET_SIZE")) {        int64_t size;        zsock_brecv (self->pipe, "8", &size);        self->size = size;    }    else    if (streq (command, "GET_SIZE")) {        zsock_bsend (self->pipe, "8", self->size);    }    else    if (streq (command, "SET_VARIANCE")) {        int64_t variance;        zsock_brecv (self->pipe, "8", &variance);        self->variance = variance;    }    else    if (streq (command, "GET_VARIANCE")) {        zsock_bsend (self->pipe, "8", self->variance);    }    else {        zsys_error ("upstream: invalid command: %s", command);        assert (false);    }    zstr_free (&command);}

intmain (int argc, char *argv []){    puts ("glard v1.0.1 -- GL-AR150 demo'n");    //  Defaults    bool verbose = false;    bool console = false;    bool ipv6 = false;    char *iface = "wlan0";    int argn;    for (argn = 1; argn < argc; argn++) {        if (streq (argv [argn], "--help")        ||  streq (argv [argn], "-h")) {            puts ("glard [options] ...");            puts ("  --help / -h            this help");            puts ("  --verbose / -v         verbose test output");            puts ("  --interface / -i       use this interface");            puts ("  --console / -c         remote control console");            puts ("  --ipv6 / -6            connect over IPv6");            return 0;        }        if (streq (argv [argn], "--verbose")        ||  streq (argv [argn], "-v"))            verbose = true;        else        if (streq (argv [argn], "--console")        ||  streq (argv [argn], "-c"))            console = true;        else        if (streq (argv [argn], "--interface")        ||  streq (argv [argn], "-i"))            iface = argv [++argn];        else        if (streq (argv [argn], "--ipv6")        ||  streq (argv [argn], "-6"))            ipv6 = true;        else {            zsys_error ("unknown option: %s/n", argv [argn]);            return -1;        }    }    zsys_set_ipv6(ipv6);    glar_node_t *node = glar_node_new (iface, console);    glar_node_set_verbose (node, verbose);    glar_node_execute (node);    glar_node_destroy (&node);    return 0;}

zuuid_t *zuuid_new (void){    zuuid_t *self = (zuuid_t *) zmalloc (sizeof (zuuid_t));    assert (self);#if defined (__WINDOWS__)    //  Windows always has UUID support    UUID uuid;    assert (sizeof (uuid) == ZUUID_LEN);    UuidCreate (&uuid);    zuuid_set (self, (byte *) &uuid);#elif defined (__UTYPE_ANDROID) || !defined (HAVE_UUID)    //  No UUID system calls, so generate a random string    byte uuid [ZUUID_LEN];    int fd = open ("/dev/urandom", O_RDONLY);    if (fd != -1) {        ssize_t bytes_read = read (fd, uuid, ZUUID_LEN);        assert (bytes_read == ZUUID_LEN);        close (fd);        zuuid_set (self, uuid);    }    else {        //  We couldn't read /dev/urandom and we have no alternative        //  strategy        zsys_error (strerror (errno));        assert (false);    }#elif defined (__UTYPE_OPENBSD) || defined (__UTYPE_FREEBSD) || defined (__UTYPE_NETBSD)    uuid_t uuid;    uint32_t status = 0;    uuid_create (&uuid, &status);    if (status != uuid_s_ok) {        zuuid_destroy (&self);        return NULL;    }    byte buffer [ZUUID_LEN];    uuid_enc_be (&buffer, &uuid);    zuuid_set (self, buffer);#elif defined (__UTYPE_LINUX) || defined (__UTYPE_OSX) || defined (__UTYPE_SUNOS) || defined (__UTYPE_SUNSOLARIS) || defined (__UTYPE_GNU)    uuid_t uuid;    assert (sizeof (uuid) == ZUUID_LEN);    uuid_generate (uuid);    zuuid_set (self, (byte *) uuid);#else#   error "Unknown UNIX TYPE"#endif    return self;}

voidzsys_socket_error (const char *reason){#if defined (__WINDOWS__)    switch (WSAGetLastError ()) {        case WSAEINTR:        errno = EINTR;      break;        case WSAEBADF:        errno = EBADF;      break;        case WSAEWOULDBLOCK:  errno = EAGAIN;     break;        case WSAEINPROGRESS:  errno = EAGAIN;     break;        case WSAENETDOWN:     errno = ENETDOWN;   break;        case WSAECONNRESET:   errno = ECONNRESET; break;        case WSAECONNABORTED: errno = EPIPE;      break;        case WSAESHUTDOWN:    errno = ECONNRESET; break;        case WSAEINVAL:       errno = EPIPE;      break;        default:              errno = GetLastError ();    }#endif    if (  errno == EAGAIN       || errno == ENETDOWN       || errno == EHOSTUNREACH       || errno == ENETUNREACH       || errno == EINTR       || errno == EPIPE       || errno == ECONNRESET#if defined (ENOPROTOOPT)       || errno == ENOPROTOOPT#endif#if defined (EHOSTDOWN)       || errno == EHOSTDOWN#endif#if defined (EOPNOTSUPP)       || errno == EOPNOTSUPP#endif#if defined (EWOULDBLOCK)       || errno == EWOULDBLOCK#endif#if defined (EPROTO)       || errno == EPROTO#endif#if defined (ENONET)       || errno == ENONET#endif          )        return;             //  Ignore error and try again    else {        zsys_error ("(UDP) error '%s' on %s", strerror (errno), reason);        assert (false);    }}

intmain (int argc, char **argv){    if (argc < 2) {        zsys_warning ("Need argument");        zsys_info ("usage: %s <regex>", argv [0]);        return EXIT_FAILURE;    }    zrex_t *regex = zrex_new (argv [1]);    if (!regex) {        zsys_error ("zrex_new () failed.");        return EXIT_FAILURE;    }    if (!zrex_valid (regex)) {        zsys_error ("regex not valid: %s", zrex_strerror (regex));        return EXIT_FAILURE;    }    zsys_info ("zrex valid");    return EXIT_SUCCESS;}

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_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;}

voidzsys_set_io_threads (size_t io_threads){    zsys_init ();    ZMUTEX_LOCK (s_mutex);    if (s_open_sockets)        zsys_error ("zsys_io_threads() is not valid after creating sockets");    assert (s_open_sockets == 0);    zmq_term (s_process_ctx);    s_io_threads = io_threads;    s_process_ctx = zmq_init ((int) s_io_threads);#if (ZMQ_VERSION >= ZMQ_MAKE_VERSION (3, 2, 0))    //  TODO: this causes TravisCI to break; libzmq does not return a    //  valid socket on zmq_socket(), after this...    zmq_ctx_set (s_process_ctx, ZMQ_MAX_SOCKETS, s_max_sockets);#endif    ZMUTEX_UNLOCK (s_mutex);}

/** * constructor */boca_hid_printer_t *boca_hid_new(const char *device_path) {    boca_hid_printer_t *self = (boca_hid_printer_t *) calloc(1, sizeof(boca_hid_printer_t));    if (!self) return NULL;    self->device_path = strdup(device_path);#ifdef __WIN32__    HANDLE handle;    handle = CreateFile(device_path,                        0,                        FILE_SHARE_READ | FILE_SHARE_WRITE,                        NULL,                        OPEN_EXISTING,                        FILE_ATTRIBUTE_NORMAL,                        NULL);    if (handle != INVALID_HANDLE_VALUE) {        PHIDP_PREPARSED_DATA preparsed_data;        if (HidD_GetPreparsedData(handle, &preparsed_data)) {            HIDP_CAPS capabilities;            NTSTATUS ret = HidP_GetCaps(preparsed_data, &capabilities);            if (ret == HIDP_STATUS_SUCCESS) {                self->input_length = capabilities.InputReportByteLength;                self->output_length = capabilities.OutputReportByteLength;                self->feature_length = capabilities.FeatureReportByteLength;                zsys_info("hid printer: input length %d output length %d feature length %d", self->input_length,                          self->output_length, self->feature_length);            } else {                zsys_warning("hid printer: can not get hid printer capabilities.");            }        } else {            zsys_warning("hid printer: can not get hid prepared data");        }        CloseHandle(handle);    } else {        zsys_warning("hid printer: can not open hid device code %d", GetLastError());    }#endif    self->device = hid_open_path(self->device_path);    if (self->device == NULL) {        zsys_error("hid printer: could not create device from path %s", device_path);        boca_hid_destroy(&self);        return NULL;    }    return self;}

static voids_api_command (agent_t *self){    char *command = zstr_recv (self->pipe);    if (streq (command, "TERMINATE")) {        self->terminated = true;        zstr_send (self->pipe, "OK");    }    else    if (streq (command, "VERBOSE")) {        char *verbose = zstr_recv (self->pipe);        self->verbose = *verbose == '1';        free (verbose);    }    else        zsys_error ("zmonitor unexpected API command '%s'/n", command);    free (command);}

intzsys_daemonize (const char *workdir){#if (defined (__UNIX__))    //  Defines umask for new files this process will create    mode_t file_mask = 027;     //  Complement of 0750    //  Recreate our process as a child of init    int fork_result = fork ();    if (fork_result < 0)        //  < 0 is an error        return -1;              //  Could not fork    else    if (fork_result > 0)        //  > 0 is the parent process        exit (0);               //  End parent process    //  Move to a safe and known directory, which is supplied as an    //  argument to this function (or not, if workdir is NULL or empty).    if (workdir && chdir (workdir)) {        zsys_error ("cannot chdir to '%s'", workdir);        return -1;    }    //  Close all open file descriptors inherited from the parent    //  process, to reduce the resources we use    int file_handle = sysconf (_SC_OPEN_MAX);    while (file_handle)        close (file_handle--);  //  Ignore any errors    //  Set the umask for new files we might create    umask (file_mask);    //  Set standard input and output to the null device so that any    //  code that assumes that these files are open will work    file_handle = open ("/dev/null", O_RDWR);    int fh_stdout = dup (file_handle);    int fh_stderr = dup (file_handle);    assert (fh_stdout);    assert (fh_stderr);    //  Ignore any hangup signal from the controlling console    signal (SIGHUP, SIG_IGN);#endif    return 0;}

static proxy_sockets_self_selected_socket (zmsg_t *request){    char *socket_name = zmsg_popstr (request);    assert (socket_name);    proxy_socket socket = NONE;    if (streq (socket_name, "FRONTEND"))        socket = FRONTEND;    else    if (streq (socket_name, "BACKEND"))        socket = BACKEND;    else {        zsys_error ("zproxy: invalid proxy socket selection: %s", socket_name);        assert (false);    }    return socket;}

book_t *shelf_next_book (shelf_t *self){ int size = PQntuples(self->all_books_res); if((self->all_books_res!=NULL) && (PQresultStatus(self->all_books_res)==PGRES_TUPLES_OK) && (self->row<size)) {  zsys_info ("get %i out of %i /n", self->row+1, size);  char *id = PQgetvalue (self->all_books_res, self->row, 0);  assert (id);  char *author = PQgetvalue (self->all_books_res, self->row, 1);  assert (author);  char *title = PQgetvalue (self->all_books_res, self->row, 2);  assert (title);  book_t *book = book_new (id, author, title);  assert (book);  zsys_info ("get object ok ! /n");  self->row = self->row + 1;  return book; } zsys_error ("could not get next object !/n"); return NULL;} 

static ints_stream_engine_handle_command (stream_engine_t *self){    char *method = zstr_recv (self->cmdpipe);    if (!method)        return -1;              //  Interrupted; exit zloop    if (self->verbose)        zsys_debug ("mlm_stream_simple: API command=%s", method);    if (streq (method, "VERBOSE"))        self->verbose = true;       //  Start verbose logging    else    if (streq (method, "$TERM"))        self->terminated = true;    //  Shutdown the engine    else    if (streq (method, "COMPILE")) {        void *client;        char *pattern;        zsock_recv (self->cmdpipe, "ps", &client, &pattern);        s_stream_engine_compile (self, client, pattern);        zstr_free (&pattern);    }    else    if (streq (method, "CANCEL")) {        void *client;        zsock_recv (self->cmdpipe, "p", &client);        s_stream_engine_cancel (self, client);    }    //  Cleanup pipe if any argument frames are still waiting to be eaten    if (zsock_rcvmore (self->cmdpipe)) {        zsys_error ("mlm_stream_simple: trailing API command frames (%s)", method);        zmsg_t *more = zmsg_recv (self->cmdpipe);        zmsg_print (more);        zmsg_destroy (&more);    }    zstr_free (&method);    return self->terminated? -1: 0;}

static ints_get_available_port (){    int port_nbr = -1;    int attempts = 0;    // Choosing a random port for better results in case multiple tests are running    // in parallel on the same machine, such as during CI runs    while (port_nbr == -1 && attempts++ < 10) {        port_nbr = 49152 + randof (16383);        zsock_t *server = zsock_new (ZMQ_PUSH);        assert (server);        port_nbr = zsock_bind (server, LOCALENDPOINT, port_nbr);        zsock_destroy (&server);    }    if (port_nbr < 0) {        zsys_error ("zproxy: failed to find an available port number");        assert (false);    }    return port_nbr;}

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 (self->verbose)        zsys_info ("zauth: API command=%s", command);    if (streq (command, "ALLOW")) {        char *address = zmsg_popstr (request);        while (address) {            if (self->verbose)                zsys_info ("zauth: - whitelisting ipaddress=%s", address);            zhashx_insert (self->whitelist, address, "OK");            zstr_free (&address);            address = zmsg_popstr (request);        }        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "DENY")) {        char *address = zmsg_popstr (request);        while (address) {            if (self->verbose)                zsys_info ("zauth: - blacklisting ipaddress=%s", address);            zhashx_insert (self->blacklist, address, "OK");            zstr_free (&address);            address = zmsg_popstr (request);        }        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "PLAIN")) {        //  Get password file and load into zhash table        //  If the file doesn't exist we'll get an empty table        char *filename = zmsg_popstr (request);        zhashx_destroy (&self->passwords);        self->passwords = zhashx_new ();        if (zhashx_load (self->passwords, filename) && self->verbose)            zsys_info ("zauth: could not load file=%s", filename);        zstr_free (&filename);        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "CURVE")) {        //  If location is CURVE_ALLOW_ANY, allow all clients. Otherwise        //  treat location as a directory that holds the certificates.        char *location = zmsg_popstr (request);        if (streq (location, CURVE_ALLOW_ANY))            self->allow_any = true;        else {            zcertstore_destroy (&self->certstore);            // FIXME: what if this fails?            self->certstore = zcertstore_new (location);            self->allow_any = false;        }        zstr_free (&location);        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "GSSAPI"))        //  GSSAPI authentication is not yet implemented here        zsock_signal (self->pipe, 0);    else    if (streq (command, "VERBOSE")) {        self->verbose = true;        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "$TERM"))        self->terminated = true;    else {        zsys_error ("zauth: - invalid command: %s", command);        assert (false);    }    zstr_free (&command);    zmsg_destroy (&request);    return 0;}