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

void zyre_set_zcert(zyre_t *self, zcert_t *zcert){    assert (zcert);    // actor will assert check the keys    zstr_sendx (self->actor, "SET PUBLICKEY", zcert_public_txt(zcert), NULL);    zstr_sendx (self->actor, "SET SECRETKEY", zcert_secret_txt(zcert), NULL);}

voidzpipes_client_test (bool verbose){    printf (" * zpipes_client: ");    //  @selftest    zactor_t *server = zactor_new (zpipes_server, NULL);    zstr_sendx (server, "SET", "server/animate", verbose? "1": "0", NULL);    zstr_sendx (server, "BIND", "ipc://@/zpipes/local", NULL);        zpipes_client_t *reader = zpipes_client_new ("local", "test pipe");    zpipes_client_t *writer = zpipes_client_new ("local", ">test pipe");    byte buffer [100];    ssize_t bytes;    //  Expect timeout error, EAGAIN    bytes = zpipes_client_read (reader, buffer, 6, 100);    assert (bytes == -1);    assert (zpipes_client_error (reader) == EAGAIN);    bytes = zpipes_client_write (writer, "CHUNK1", 6, 100);    assert (bytes == 6);    bytes = zpipes_client_write (writer, "CHUNK2", 6, 100);    assert (bytes == 6);    bytes = zpipes_client_write (writer, "CHUNK3", 6, 100);    assert (bytes == 6);    bytes = zpipes_client_read (reader, buffer, 1, 100);    assert (bytes == 1);    bytes = zpipes_client_read (reader, buffer, 10, 100);    assert (bytes == 10);    //  Now close writer    zpipes_client_destroy (&writer);    //  Expect end of pipe (short read)    bytes = zpipes_client_read (reader, buffer, 50, 100);    assert (bytes == 7);        //  Expect end of pipe (empty chunk)    bytes = zpipes_client_read (reader, buffer, 50, 100);    assert (bytes == 0);    //  Expect illegal action (EBADF) writing on reader    bytes = zpipes_client_write (reader, "CHUNK1", 6, 100);    assert (bytes == -1);    assert (zpipes_client_error (reader) == EBADF);    zpipes_client_destroy (&reader);    zpipes_client_destroy (&writer);    zactor_destroy (&server);    //  @end    printf ("OK/n");}

static voids_bind_proxy_sockets (zactor_t *proxy, char **frontend, char **backend){    if (*frontend)        zstr_free (frontend);    if (*backend)        zstr_free (backend);    *frontend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());    *backend = zsys_sprintf (LOCALENDPOINT, s_get_available_port ());    zstr_sendx (proxy, "FRONTEND", "PULL", *frontend, NULL);    zsock_wait (proxy);    zstr_sendx (proxy, "BACKEND", "PUSH", *backend, NULL);    zsock_wait (proxy);}

voidzyre_stop (zyre_t *self){    assert (self);    zstr_sendx (self->actor, "STOP", NULL);    zsock_wait (self->actor);}

int main(int argc, char **argv) {    //    //    const char *config_file = "malamute.cfg";    if (argc >= 2)        config_file = argv[1];    //  Start Malamute server instance    zactor_t *server = zactor_new (mlm_server, "Malamute");    zstr_sendx (server, "LOAD", config_file, NULL);    //  Accept and print any message back from server    while (true) {        char *message = zstr_recv (server);        if (message) {            puts (message);            free (message);        }        else {            puts ("interrupted");            break;        }    }    //  Shutdown all services    zactor_destroy (&server);    return EXIT_SUCCESS;}

static voidserver_accept (server_t *self, const char *key, const char *value){    tuple_t *tuple = (tuple_t *) zhash_lookup (self->tuples, key);    if (tuple && streq (tuple->value, value))        return;                 //  Duplicate tuple, do nothing    //  Create new tuple    tuple = (tuple_t *) zmalloc (sizeof (tuple_t));    assert (tuple);    tuple->container = self->tuples;    tuple->key = strdup (key);    tuple->value = strdup (value);    //  Store new tuple    zhash_update (tuple->container, key, tuple);    zhash_freefn (tuple->container, key, tuple_free);    //  Deliver to calling application    zstr_sendx (self->pipe, "DELIVER", key, value, NULL);    //  Hold in server context so we can broadcast to all clients    self->cur_tuple = tuple;    engine_broadcast_event (self, NULL, forward_event);    //  Copy new tuple announcement to all remotes    zsock_t *remote = (zsock_t *) zlist_first (self->remotes);    while (remote) {        int rc = zgossip_msg_send_publish (remote, key, value, 0);        assert (rc == 0);        remote = (zsock_t *) zlist_next (self->remotes);    }}

intzyre_join (zyre_t *self, const char *group){    assert (self);    zstr_sendx (self->actor, "JOIN", group, NULL);    return 0;}

voidmdp_broker_test (bool verbose){    printf (" * mdp_broker: ");    if (verbose)        printf ("/n");        //  @selftest    zactor_t *server = zactor_new (mdp_broker, "server");    if (verbose)        zstr_send (server, "VERBOSE");    zstr_sendx (server, "BIND", "ipc://@/mdp_broker", NULL);    zsock_t *client = zsock_new (ZMQ_DEALER);    assert (client);    zsock_set_rcvtimeo (client, 2000);    zsock_connect (client, "ipc://@/mdp_broker");    //  TODO: fill this out    mdp_msg_t *request = mdp_msg_new ();    mdp_msg_destroy (&request);        zsock_destroy (&client);    zactor_destroy (&server);    //  @end    printf ("OK/n");}

zyre_t *zyre_new (const char *name){    zyre_t *self = (zyre_t *) zmalloc (sizeof (zyre_t));    assert (self);    //  Create front-to-back pipe pair for data traffic    self->inbox = zsock_new (ZMQ_PAIR);    assert (self->inbox);    char endpoint [32];    while (true) {        sprintf (endpoint, "inproc://zyre-%04x-%04x/n",                 randof (0x10000), randof (0x10000));        if (zsock_bind (self->inbox, "%s", endpoint) == 0)            break;    }    //  Create other half of traffic pipe    zsock_t *outbox = zsock_new_pair (endpoint);    assert (outbox);        //  Start node engine and wait for it to be ready    self->actor = zactor_new (zyre_node_actor, outbox);    assert (self->actor);    //  Send name, if any, to node ending    if (name)        zstr_sendx (self->actor, "SET NAME", name, NULL);        return self;}

int main (void) {  zsys_set_ipv6 (1);  zactor_t *root = zactor_new (zgossip, "root");  assert (root);  int rc = 0;  rc = zstr_sendx (root, "BIND", "tcp://*:5670", NULL);  assert (rc == 0);  zloop_t *reactor = zloop_new ();  assert (reactor);  zloop_set_verbose (reactor, true);  rc = zloop_reader (reactor, root, handle_pipe, NULL);  assert (rc == 0);  zloop_start (reactor);  zloop_destroy (&reactor);  zactor_destroy (&root);  return 0;}

intzyre_leave (zyre_t *self, const char *group){    assert (self);    zstr_sendx (self->actor, "LEAVE", group, NULL);    return 0;}

intzyre_start (zyre_t *self){    assert (self);    zstr_sendx (self->actor, "START", NULL);    return zsock_wait (self->actor) == 0? 0: -1;}

voidzauth_allow (zauth_t *self, char *address){    zstr_sendx (self->pipe, "ALLOW", address, NULL);    //  Wait for completion    free (zstr_recv (self->pipe));}

voidzauth_deny (zauth_t *self, char *address){    zstr_sendx (self->pipe, "DENY", address, NULL);    //  Wait for completion    free (zstr_recv (self->pipe));}

voidzauth_configure_gssapi (zauth_t *self, char *domain){    assert (self);    assert (domain);    zstr_sendx (self->pipe, "GSSAPI", domain, NULL);    zsocket_wait (self->pipe);}

intzap_request_reply (zap_request_t *self, char *status_code, char *status_text){    zstr_sendx (self->handler,         "1.0", self->sequence, status_code, status_text, "", "",        NULL);    return 0;}

int main(int argc, char** argv) {    if(argc != 2) {        fprintf(stderr, "Usage: %s ups_name/n", argv[0]);        exit(1);    }    char *addr = NULL;    zyre_t *n = zyre_new(argv[1]);    zyre_start(n);    zyre_join(n, "BIOS");    while(!zsys_interrupted && addr == NULL) {        zyre_event_t *e = zyre_event_new(n);        if(!e)            break;        if(zyre_event_headers(e) && zyre_event_header(e, "HAP_SERVER") != NULL) {            addr = strdup(zyre_event_header(e, "HAP_SERVER"));            printf("Address: %s/n", addr);        }        zyre_event_destroy(&e);    }    zyre_destroy(&n);    if(addr == NULL)        exit(1);    zsock_t * sc = zsock_new(ZMQ_PUB);    zsock_connect(sc, "%s", addr);    bool state = random()%2;    int timeout = 0;    while(!zsys_interrupted) {        if(timeout == 0) {            state = !state;            timeout = 5 + random()%20;        }        timeout--;        if(state) {            zstr_sendx(sc, argv[1], "ON", NULL);            zsys_debug("UPS %s ON", argv[1]);        } else {            zstr_sendx(sc, argv[1], "OFF", NULL);            zsys_debug("UPS %s OFF", argv[1]);        }        sleep(1);    }    zsock_destroy(&sc);}

voidzauth_configure_plain (zauth_t *self, const char *domain, const char *filename){    assert (self);    assert (domain);    assert (filename);    zstr_sendx (self->pipe, "PLAIN", domain, filename, NULL);    zsocket_wait (self->pipe);}

voidzauth_configure_curve (zauth_t *self, const char *domain, const char *location){    assert (self);    assert (domain);    assert (location);    zstr_sendx (self->pipe, "CURVE", domain, location, NULL);    zsocket_wait (self->pipe);}

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

voidzauth_configure_curve (zauth_t *self, char *domain, char *location){    assert (self);    assert (domain);    assert (location);    zstr_sendx (self->pipe, "CURVE", domain, location, NULL);    //  Wait for completion    free (zstr_recv (self->pipe));}

const char *zyre_name (zyre_t *self){    assert (self);    //  Hold name in zyre object so caller gets a safe reference    zstr_free (&self->name);    zstr_sendx (self->actor, "NAME", NULL);    self->name = zstr_recv (self->actor);    return self->name;}

const char *zyre_endpoint (zyre_t *self){    assert (self);    //  Hold endpoint in zyre object so caller gets a safe reference    zstr_free (&self->endpoint);    zstr_sendx (self->actor, "ENDPOINT", NULL);    self->endpoint = zstr_recv (self->actor);    return self->endpoint;}

voidzauth_configure_plain (zauth_t *self, char *domain, char *filename){    assert (self);    assert (domain);    assert (filename);    zstr_sendx (self->pipe, "PLAIN", domain, filename, NULL);    //  Wait for completion    free (zstr_recv (self->pipe));}

const char *zyre_uuid (zyre_t *self){    assert (self);    //  Hold uuid string in zyre object so caller gets a safe reference    zstr_free (&self->uuid);    zstr_sendx (self->actor, "UUID", NULL);    self->uuid = zstr_recv (self->actor);    return self->uuid;}

static int twps_create_ticket_printer(twps_server_t *self, zmsg_t *msg) {    char *type = zmsg_popstr(msg);    char *id = zmsg_popstr(msg);    zstr_free(&id);    char *path = zmsg_popstr(msg);    char *model = zmsg_popstr(msg);    wchar_t *manufacture = (wchar_t *) zmsg_popstr(msg);    wchar_t *product = (wchar_t *) zmsg_popstr(msg);    zactor_t *printer = NULL;    if (streq(type, "HID")) {        printer = zactor_new(ticket_hid_printer, path);    }#ifdef __WIN32__    else if (streq(type, "SERIAL")) {        printer = zactor_new(ticket_serial_printer, path);    }#endif    if (printer != NULL) {        if (self->verbose) {            zstr_send(printer, "VERBOSE");        }        if (self->diagnostic)            zstr_sendx(printer, "SETDIAGNOSTIC", NULL);        zstr_sendx(printer, "START", TICKET_STORE_REP_ENDPOINT, NULL);        zsock_wait(printer);        zstr_sendx(printer, "SETPRINTERSTORE", PRINTER_STORE_REP_ENDPOINT, PRINTER_STORE_PUB_ENDPOINT, NULL);        zsock_wait(printer);        zsys_info("twps server: started ticket printer %s manufacturer|product|model %ls|%ls|%s", type, manufacture,                  product, model);        zlistx_add_end(self->ticket_printers, printer);    } else {        zsys_warning("twps server: printer not added %s, %s", type, path);        zmsg_print(msg);    }    zstr_free(&type);    zstr_free(&path);    zstr_free(&model);    zstr_free((char **) &manufacture);    zstr_free((char **) &product);    return 0;}

static ints_zap_request_reply (zap_request_t *self, char *status_code, char *status_text){    if (self->verbose)        zsys_info ("zauth: - ZAP reply status_code=%s status_text=%s",                   status_code, status_text);    zstr_sendx (self->handler,                "1.0", self->sequence, status_code, status_text, "", "",                NULL);    return 0;}

voidzyre_set_announce (zyre_t *self, const char *format, ...){    assert (self);    va_list argptr;    va_start (argptr, format);    char *string = zsys_vprintf (format, argptr);    va_end (argptr);    zstr_sendx (self->actor, "SET ANNOUNCE", string, NULL);    free (string);}

voidzauth_configure_plain (zauth_t *self, char *domain, char *filename, ...){    assert (self);    assert (domain);    va_list argptr;    va_start (argptr, filename);    char *formatted = zsys_vprintf (filename, argptr);    va_end (argptr);    zstr_sendx (self->pipe, "PLAIN", domain, formatted, NULL);    free (formatted);}

voidzyre_gossip_connect (zyre_t *self, const char *format, ...){    assert (self);    va_list argptr;    va_start (argptr, format);    char *string = zsys_vprintf (format, argptr);    va_end (argptr);    zstr_sendx (self->actor, "GOSSIP CONNECT", string, NULL);    free (string);}