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

voidzproxy_test (bool verbose){    printf (" * zproxy: ");    //  @selftest    zctx_t *ctx = zctx_new ();    void *frontend = zsocket_new (ctx, ZMQ_PULL);    int rc = zsocket_bind (frontend, "inproc://frontend");    assert (rc == 0);    void *backend = zsocket_new (ctx, ZMQ_PUSH);    rc = zsocket_bind (backend, "inproc://backend");    assert (rc == 0);    zproxy_t *proxy = zproxy_new (ctx, frontend, backend);    //  Connect application sockets to proxy    void *faucet = zsocket_new (ctx, ZMQ_PUSH);    rc = zsocket_connect (faucet, "inproc://frontend");    assert (rc == 0);    void *sink = zsocket_new (ctx, ZMQ_PULL);    rc = zsocket_connect (sink, "inproc://backend");    assert (rc == 0);    //  Send some messages and check they arrived    zstr_send (faucet, "Hello");    char *string = zstr_recv (sink);    assert (streq (string, "Hello"));    zstr_free (&string);        //  Check pause/resume functionality    zproxy_pause (proxy);    zstr_send (faucet, "World");    zproxy_resume (proxy);    string = zstr_recv (sink);    assert (streq (string, "World"));    zstr_free (&string);        //  Create capture socket, must be a PULL socket    void *capture = zsocket_new (ctx, ZMQ_PULL);    rc = zsocket_bind (capture, "inproc://capture");    assert (rc == 0);    //  Switch on capturing, check that it works    zproxy_capture (proxy, "inproc://capture");    zstr_send (faucet, "Hello");        string = zstr_recv (sink);    assert (streq (string, "Hello"));    zstr_free (&string);        string = zstr_recv (capture);    assert (streq (string, "Hello"));    zstr_free (&string);    zproxy_destroy (&proxy);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");}

static void test_zmq (flux_reactor_t *reactor){    zctx_t *zctx;    void *zs[2];    flux_watcher_t *r, *w;    ok ((zctx = zctx_new ()) != NULL,        "zmq: created zmq context");    zs[0] = zsocket_new (zctx, ZMQ_PAIR);    zs[1] = zsocket_new (zctx, ZMQ_PAIR);    ok (zs[0] && zs[1]        && zsocket_bind (zs[0], "inproc://test_zmq") == 0        && zsocket_connect (zs[1], "inproc://test_zmq") == 0,        "zmq: connected ZMQ_PAIR sockets over inproc");    r = flux_zmq_watcher_create (reactor, zs[0], FLUX_POLLIN, zmqreader, NULL);    w = flux_zmq_watcher_create (reactor, zs[1], FLUX_POLLOUT, zmqwriter, NULL);    ok (r != NULL && w != NULL,        "zmq: nonblocking reader and writer created");    flux_watcher_start (r);    flux_watcher_start (w);    ok (flux_reactor_run  (reactor, 0) == 0,        "zmq: reactor ran to completion after %d messages", zmqwriter_msgcount);    flux_watcher_stop (r);    flux_watcher_stop (w);    flux_watcher_destroy (r);    flux_watcher_destroy (w);    zsocket_destroy (zctx, zs[0]);    zsocket_destroy (zctx, zs[1]);    zctx_destroy (&zctx);}

//  Checks whether client can connect to serverstatic bools_can_connect (zctx_t *ctx, void **server, void **client){    int port_nbr = zsocket_bind (*server, "tcp://127.0.0.1:*");    assert (port_nbr > 0);    int rc = zsocket_connect (*client, "tcp://127.0.0.1:%d", port_nbr);    assert (rc == 0);    //  Give the connection time to fail if that's the plan    zclock_sleep (200);    //  By default PUSH sockets block if there's no peer    zsock_set_sndtimeo (*server, 200);    zstr_send (*server, "Hello, World");    zpoller_t *poller = zpoller_new (*client, NULL);    bool success = (zpoller_wait (poller, 400) == *client);    zpoller_destroy (&poller);    zsocket_destroy (ctx, *client);    zsocket_destroy (ctx, *server);    *server = zsocket_new (ctx, ZMQ_PUSH);    assert (*server);    *client = zsocket_new (ctx, ZMQ_PULL);    assert (*client);    return success;}

void *zthread_fork (zctx_t *ctx, zthread_attached_fn *thread_fn, void *args){    //  Create our end of the pipe    //  Pipe has HWM of 1 at both sides to block runaway writers    void *pipe = zsocket_new (ctx, ZMQ_PAIR);    assert (pipe);    zsockopt_set_hwm (pipe, 1);    zsocket_bind (pipe, "inproc://zctx-pipe-%p", pipe);    //  Prepare argument shim for child thread    shim_t *shim = (shim_t *) zmalloc (sizeof (shim_t));    shim->attached = thread_fn;    shim->args = args;    shim->ctx = zctx_shadow (ctx);    //  Connect child pipe to our pipe    shim->pipe = zsocket_new (shim->ctx, ZMQ_PAIR);    assert (shim->pipe);    zsockopt_set_hwm (shim->pipe, 1);    zsocket_connect (shim->pipe, "inproc://zctx-pipe-%p", pipe);    s_thread_start (shim);    return pipe;}

static agent_t *s_agent_new (zctx_t *ctx, void *control){    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));    self->ctx = ctx;    self->control = control;    self->state = waiting;    self->dealer = zsocket_new (ctx, ZMQ_DEALER);    //  Connect our data socket to caller's endpoint    self->data = zsocket_new (ctx, ZMQ_PAIR);    char *endpoint = zstr_recv (self->control);    int rc = zsocket_connect (self->data, "%s", endpoint);    assert (rc != -1);    free (endpoint);    //  Create new client codec using cert from API    byte public_key [32];    byte secret_key [32];    rc = zmq_recv (self->control, public_key, 32, 0);    assert (rc == 32);    rc = zmq_recv (self->control, secret_key, 32, 0);    assert (rc == 32);        zcert_t *cert = zcert_new_from (public_key, secret_key);    self->codec = curve_codec_new_client (cert);    zcert_destroy (&cert);    return self;}

intmain (int argc, char *argv[]){    if (argc != 3) {        exit (-1);    }    int numb_msgs = atoi (argv[2]);    zctx_t *ctx = zctx_new ();    void *dealer = zsocket_new (ctx, ZMQ_DEALER);    zsocket_set_linger (dealer, -1);    zsocket_connect (dealer, "%s:9000", argv[1]);    void *sub = zsocket_new (ctx, ZMQ_SUB);    zsocket_connect (sub, "%s:9002", argv[1]);    zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "all", 4);    int64_t time[2];    zmq_pollitem_t pollitem[1] = { {sub, 0, ZMQ_POLLIN}    };    zmq_poll (pollitem, 1, -1);    zmsg_t *signal = zmsg_recv (sub);    zmsg_destroy (&signal);    char blob[SIZE] = { 0 };    zmsg_t *msg = zmsg_new ();    zframe_t *frame = zframe_new (blob, SIZE);    zmsg_add (msg, frame);    time[0] = zclock_time ();    int i;    for (i = 0; i < numb_msgs; i++) {        zmsg_t *nmsg = zmsg_dup (msg);        zmsg_send (&nmsg, dealer);    }    time[1] = zclock_time ();    zmsg_destroy (&msg);    zmq_poll (pollitem, 1, -1);    msg = zmsg_recv (sub);    zmsg_destroy (&msg);    msg = zmsg_new ();    frame = zframe_new (time, sizeof (int64_t) * 2);    zmsg_add (msg, frame);    zmsg_send (&msg, dealer);    zctx_destroy (&ctx);}

static agent_t *s_agent_new (zctx_t *ctx, void *control){    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));    self->ctx = ctx;    self->control = control;    self->router = zsocket_new (ctx, ZMQ_ROUTER);    //  Connect our data socket to caller's endpoint    self->data = zsocket_new (ctx, ZMQ_PAIR);    char *endpoint = zstr_recv (self->control);    int rc = zsocket_connect (self->data, "%s", endpoint);    assert (rc != -1);    free (endpoint);    //  Create new client codec using cert from API    byte public_key [32];    byte secret_key [32];    rc = zmq_recv (self->control, public_key, 32, 0);    assert (rc == 32);    rc = zmq_recv (self->control, secret_key, 32, 0);    assert (rc == 32);    self->cert = zcert_new_from (public_key, secret_key);            self->metadata = zhash_new ();    zhash_autofree (self->metadata);    self->clients = zhash_new ();    self->max_clients = 100;    self->max_pending = 10;    self->client_ttl = 3600;    //  60 minutes    self->pending_ttl = 60;     //  60 seconds    return self;}

int main (void){    clonesrv_t *self = (clonesrv_t *) zmalloc (sizeof (clonesrv_t));    self->port = 5556;    self->ctx = zctx_new ();    self->kvmap = zhash_new ();    self->loop = zloop_new ();    zloop_set_verbose (self->loop, FALSE);    //  Set up our clone server sockets    self->snapshot  = zsocket_new (self->ctx, ZMQ_ROUTER);    self->publisher = zsocket_new (self->ctx, ZMQ_PUB);    self->collector = zsocket_new (self->ctx, ZMQ_PULL);    zsocket_bind (self->snapshot,  "tcp://*:%d", self->port);    zsocket_bind (self->publisher, "tcp://*:%d", self->port + 1);    zsocket_bind (self->collector, "tcp://*:%d", self->port + 2);    //  Register our handlers with reactor    zmq_pollitem_t poller = { self->snapshot, 0, ZMQ_POLLIN };    zloop_poller (self->loop, &poller, s_snapshots, self);    poller.socket = self->collector;    zloop_poller (self->loop, &poller, s_collector, self);    zloop_timer  (self->loop, 1000, 0, s_flush_ttl, self);    //  Run reactor until process interrupted    zloop_start (self->loop);    zloop_destroy (&self->loop);    zhash_destroy (&self->kvmap);    zctx_destroy (&self->ctx);    free (self);    return 0;}

void *zthread_fork (zctx_t *ctx, zthread_attached_fn *thread_fn, void *args){    shim_t *shim = NULL;    //  Create our end of the pipe    void *pipe = zsocket_new (ctx, ZMQ_PAIR);    if (pipe) {        zsocket_set_hwm (pipe, zctx_hwm (ctx));        zsocket_bind (pipe, "inproc://zctx-pipe-%p", pipe);    }    else        return NULL;        //  Prepare argument shim for child thread    shim = (shim_t *) zmalloc (sizeof (shim_t));    if (shim) {        shim->attached = thread_fn;        shim->args = args;        shim->ctx = zctx_shadow (ctx);        if (!shim->ctx)            return NULL;    }    else        return NULL;        //  Connect child pipe to our pipe    shim->pipe = zsocket_new (shim->ctx, ZMQ_PAIR);    if (!shim->pipe)        return NULL;    zsocket_set_hwm (shim->pipe, 1);    zsocket_connect (shim->pipe, "inproc://zctx-pipe-%p", pipe);        s_thread_start (shim);    return pipe;}

bstar_t *bstar_new (int primary, char *local, char *remote){    bstar_t        *self;    self = (bstar_t *) zmalloc (sizeof (bstar_t));    //  Initialize the Binary Star    self->ctx = zctx_new ();    self->loop = zloop_new ();    self->state = primary? STATE_PRIMARY: STATE_BACKUP;    //  Create publisher for state going to peer    self->statepub = zsocket_new (self->ctx, ZMQ_PUB);    zsocket_bind (self->statepub, local);    //  Create subscriber for state coming from peer    self->statesub = zsocket_new (self->ctx, ZMQ_SUB);    zsocket_set_subscribe (self->statesub, "");    zsocket_connect (self->statesub, remote);    //  Set-up basic reactor events    zloop_timer (self->loop, BSTAR_HEARTBEAT, 0, s_send_state, self);    zmq_pollitem_t poller = { self->statesub, 0, ZMQ_POLLIN };    zloop_poller (self->loop, &poller, s_recv_state, self);    return self;}

void *server_task (void *server_args) {    //  Frontend socket talks to clients over TCP    zctx_t *ctx = zctx_new ();    void *frontend = zsocket_new(ctx, ZMQ_ROUTER);    char str[20];    strcpy(str, "tcp://*:");    strcat(str, PORT);    zsocket_bind(frontend, str);    //  Backend socket talks to workers over inproc    void *backend = zsocket_new (ctx, ZMQ_DEALER);    zsocket_bind (backend, "inproc://backend");    //  Launch pool of worker threads, precise number is not critical    //   for (thread_nbr = 0; thread_nbr < 5; thread_nbr++)    zthread_fork (ctx, server_worker, server_args);    //  Connect backend to frontend via a proxy    zmq_proxy (frontend, backend, NULL);    printf("back/n");    zctx_destroy (&ctx);    return NULL;}

static voidcollabclient_remakeSockets( cloneclient_t *cc ){    cc->snapshot = zsocket_new (cc->ctx, ZMQ_DEALER);    zsocket_connect (cc->snapshot,	collabclient_makeAddressString( cc->address,	cc->port + socket_offset_snapshot));        cc->subscriber = zsocket_new (cc->ctx, ZMQ_SUB);    zsocket_set_subscribe (cc->subscriber, "");    zsocket_connect (cc->subscriber,	collabclient_makeAddressString( cc->address,	cc->port + socket_offset_subscriber));    zsocket_set_subscribe (cc->subscriber, SUBTREE);    cc->publisher = zsocket_new (cc->ctx, ZMQ_PUSH);    zsocket_connect (cc->publisher,	collabclient_makeAddressString( cc->address,	cc->port + socket_offset_publisher));    int fd = 0;    size_t fdsz = sizeof(fd);    int rc = zmq_getsockopt( cc->subscriber, ZMQ_FD, &fd, &fdsz );    printf("subscriber rc:%d fd:%d/n", rc, fd );    GDrawAddReadFD( 0, fd, cc, zeromq_subscriber_fd_callback );  }

int main (int argc, char *argv []){    //  Socket to talk to server	printf ("Collecting updates from weather server.../n");        zctx_t *context = zctx_new ();	void *subscriber = zsocket_new (context, ZMQ_PULL);	zsocket_set_hwm(subscriber, 100000); 		int rc = zsocket_connect (subscriber, RECEIVE_SOCKET);	assert (rc == 0);	void *worker = zsocket_new(context, ZMQ_PUSH);	rc = zsocket_bind(worker, WORKER_SOCKET);		assert(rc == 0);	int nthreads = 0;	for (nthreads=0; nthreads < 50; nthreads++)	{		zthread_fork(context, parser_thread, NULL);	}	//  Subscribe to zipcode, default is NYC, 10001	while(1)	{		int size;		char *string = safe_recv_from_server (subscriber, &size);		parse_notifications(string, size,  worker);		free (string);	}printf("Ending /n");	zctx_destroy (&context);	return 0;}

int main (void){    zctx_t *ctx = zctx_new ();    void *frontend = zsocket_new (ctx, ZMQ_ROUTER);    void *backend = zsocket_new (ctx, ZMQ_ROUTER);    zsocket_bind (frontend, "tcp://*:5555");    //  For clients    zsocket_bind (backend,  "tcp://*:5556");    //  For workers    //  Queue of available workers    zlist_t *workers = zlist_new ();        //  The body of this example is exactly the same as lruqueue2.    //  .skip    while (1) {        zmq_pollitem_t items [] = {            { backend,  0, ZMQ_POLLIN, 0 },            { frontend, 0, ZMQ_POLLIN, 0 }        };        //  Poll frontend only if we have available workers        int rc = zmq_poll (items, zlist_size (workers)? 2: 1, -1);        if (rc == -1)            break;              //  Interrupted        //  Handle worker activity on backend        if (items [0].revents & ZMQ_POLLIN) {            //  Use worker address for LRU routing            zmsg_t *msg = zmsg_recv (backend);            if (!msg)                break;          //  Interrupted            zframe_t *address = zmsg_unwrap (msg);            zlist_append (workers, address);            //  Forward message to client if it's not a READY            zframe_t *frame = zmsg_first (msg);            if (memcmp (zframe_data (frame), LRU_READY, 1) == 0)                zmsg_destroy (&msg);            else                zmsg_send (&msg, frontend);        }        if (items [1].revents & ZMQ_POLLIN) {            //  Get client request, route to first available worker            zmsg_t *msg = zmsg_recv (frontend);            if (msg) {                zmsg_wrap (msg, (zframe_t *) zlist_pop (workers));                zmsg_send (&msg, backend);            }        }    }    //  When we're done, clean up properly    while (zlist_size (workers)) {        zframe_t *frame = (zframe_t *) zlist_pop (workers);        zframe_destroy (&frame);    }    zlist_destroy (&workers);    zctx_destroy (&ctx);    return 0;    //  .until}

int main (int argc, char *argv []){    //  First argument is this broker's name    //  Other arguments are our peers' names    //    if (argc < 2) {        printf ("syntax: peering1 me {you}.../n");        exit (EXIT_FAILURE);    }    char *self = argv [1];    printf ("I: preparing broker at %s.../n", self);    srandom ((unsigned) time (NULL));    zctx_t *ctx = zctx_new ();        //  Bind state backend to endpoint    void *statebe = zsocket_new (ctx, ZMQ_PUB);    zsocket_bind (statebe, "ipc://%s-state.ipc", self);        //  Connect statefe to all peers    void *statefe = zsocket_new (ctx, ZMQ_SUB);    zsockopt_set_subscribe (statefe, "");    int argn;    for (argn = 2; argn < argc; argn++) {        char *peer = argv [argn];        printf ("I: connecting to state backend at '%s'/n", peer);        zsocket_connect (statefe, "ipc://%s-state.ipc", peer);    }    //  .split main loop    //  The main loop sends out status messages to peers, and collects    //  status messages back from peers. The zmq_poll timeout defines    //  our own heartbeat:    while (1) {        //  Poll for activity, or 1 second timeout        zmq_pollitem_t items [] = { { statefe, 0, ZMQ_POLLIN, 0 } };        int rc = zmq_poll (items, 1, 1000 * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Interrupted        //  Handle incoming status messages        if (items [0].revents & ZMQ_POLLIN) {            char *peer_name = zstr_recv (statefe);            char *available = zstr_recv (statefe);            printf ("%s - %s workers free/n", peer_name, available);            free (peer_name);            free (available);        }        else {            //  Send random values for worker availability            zstr_sendm (statebe, self);            zstr_sendf (statebe, "%d", randof (10));        }    }    zctx_destroy (&ctx);    return EXIT_SUCCESS;}

int game_thread( void * _parms ) {  GameThreadParms * parms = (GameThreadParms*)_parms;  GameState gs;  SharedRenderState rs;  game_init( gs, rs );  gs.zmq_control_socket = zsocket_new( parms->zmq_context, ZMQ_PAIR );  {    int ret = zsocket_connect( gs.zmq_control_socket, "inproc://control_game" );    assert( ret == 0 );  }    gs.zmq_render_socket = zsocket_new( parms->zmq_context, ZMQ_PAIR );  zsocket_bind( gs.zmq_render_socket, "inproc://game_render" );  gs.zmq_input_req = zsocket_new( parms->zmq_context, ZMQ_REQ );  {    int ret = zsocket_connect( gs.zmq_input_req, "inproc://input" );    assert( ret == 0 );  }  unsigned int baseline = SDL_GetTicks();  int framenum = 0;  while ( true ) {    unsigned int now = SDL_GetTicks();    unsigned int target_frame = ( now - baseline ) / GAME_DELAY;    if ( framenum <= target_frame ) {      framenum++;      // NOTE: build the state of the world at t = framenum * GAME_DELAY,      // under normal conditions that's a time in the future      // (the exception to that is if we are catching up on ticking game frames)      game_tick( now, gs, rs );      // notify the render thread that a new game state is ready.      // on the next render frame, it will start interpolating between the previous state and this new one      zstr_sendf( gs.zmq_render_socket, "%d %f %f %f %f %f %f %f %f %f", baseline + framenum * GAME_DELAY, rs.position.x, rs.position.y, rs.orientation.w, rs.orientation.x, rs.orientation.y, rs.orientation.z, rs.smoothed_angular.x, rs.smoothed_angular.y, rs.smoothed_angular.z );    } else {      int ahead = framenum * GAME_DELAY - ( now - baseline );      assert( ahead > 0 );      printf( "game sleep %d ms/n", ahead );      SDL_Delay( ahead );    }    char * cmd = zstr_recv_nowait( gs.zmq_control_socket );    if ( cmd != NULL ) {      assert( strcmp( cmd, "stop" ) == 0 );      free( cmd );      break;    }  }  return 0;}

static void *broker_task (void *args){    //  Prepare our context and sockets    zctx_t *ctx = zctx_new ();    void *frontend = zsocket_new (ctx, ZMQ_DEALER);    zsocket_bind (frontend, "tcp://*:5555");    void *backend = zsocket_new (ctx, ZMQ_DEALER);    zsocket_bind (backend, "tcp://*:5556");    zmq_proxy (frontend, backend, NULL);    zctx_destroy (&ctx);    return NULL;}

int main (int argc, char** argv){    int port = 5556;#ifdef FF_USE_LIBGC    GC_INIT();    set_program_name (argv[0]);#endif    if( argc >= 2 )	port = atoi( argv[1] );	    clonesrv_t *self = (clonesrv_t *) zmalloc (sizeof (clonesrv_t));    self->port = port;    self->ctx = zctx_new ();    self->kvmap = zhash_new ();    self->loop = zloop_new ();    zloop_set_verbose (self->loop, false);    //  Set up our clone server sockets    self->snapshot  = zsocket_new (self->ctx, ZMQ_ROUTER);    zsocket_bind (self->snapshot,  "tcp://*:%d", self->port + socket_srv_offset_snapshot );    self->publisher = zsocket_new (self->ctx, ZMQ_PUB);    zsocket_bind (self->publisher, "tcp://*:%d", self->port + socket_srv_offset_publisher );    self->collector = zsocket_new (self->ctx, ZMQ_PULL);    zsocket_bind (self->collector, "tcp://*:%d", self->port + socket_srv_offset_collector );    self->ping = zsocket_new (self->ctx, ZMQ_REP);    zsocket_bind (self->ping,      "tcp://*:%d", self->port + socket_srv_offset_ping );    //  Register our handlers with reactor    zmq_pollitem_t poller = { 0, 0, ZMQ_POLLIN };    poller.socket = self->snapshot;    zloop_poller (self->loop, &poller, s_snapshots, self);    poller.socket = self->collector;    zloop_poller (self->loop, &poller, s_collector, self);    zloop_timer (self->loop, 1000, 0, s_flush_ttl, self);    poller.socket = self->ping;    zloop_poller (self->loop, &poller, s_ping, self);            DEBUG ("I: server up and running on port:%d...", port );    //  Run reactor until process interrupted    zloop_start (self->loop);    zloop_destroy (&self->loop);    zhash_destroy (&self->kvmap);    zctx_destroy (&self->ctx);    free (self);    return 0;}

voidplatanos_init (platanos_t ** platanos,               compute_t * compute, char *id, zctx_t * ctx){    *platanos = malloc (sizeof (platanos_t));    (*platanos)->router = zsocket_new (ctx, ZMQ_ROUTER);    (*platanos)->dealer = zsocket_new (ctx, ZMQ_DEALER);    zmq_setsockopt ((*platanos)->dealer, ZMQ_IDENTITY, id, strlen (id));    platanos_poll_init (&((*platanos)->poll), *platanos);    (*platanos)->compute = compute;}

voidzmonitor_test (bool verbose){    printf (" * zmonitor: ");    if (verbose)        printf ("/n");    //  @selftest    zctx_t *ctx = zctx_new ();    bool result;    void *sink = zsocket_new (ctx, ZMQ_PULL);    zmonitor_t *sinkmon = zmonitor_new (ctx,        sink, ZMQ_EVENT_LISTENING | ZMQ_EVENT_ACCEPTED);    zmonitor_set_verbose (sinkmon, verbose);    //  Check sink is now listening    int port_nbr = zsocket_bind (sink, "tcp://127.0.0.1:*");    assert (port_nbr != -1);    result = s_check_event (sinkmon, ZMQ_EVENT_LISTENING);    assert (result);    void *source = zsocket_new (ctx, ZMQ_PUSH);    zmonitor_t *sourcemon = zmonitor_new (ctx,        source, ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED);    zmonitor_set_verbose (sourcemon, verbose);    zsocket_connect (source, "tcp://127.0.0.1:%d", port_nbr);    //  Check source connected to sink    result = s_check_event (sourcemon, ZMQ_EVENT_CONNECTED);    assert (result);    //  Check sink accepted connection    result = s_check_event (sinkmon, ZMQ_EVENT_ACCEPTED);    assert (result);    //  Destroy sink to trigger a disconnect event on the source    //  PH: disabled since this causes an access violation in    //  zmonitor_destroy as the socket is no longer valid.//     zsocket_destroy (ctx, sink);//     result = s_check_event (sourcemon, ZMQ_EVENT_DISCONNECTED);//     assert (result);    zmonitor_destroy (&sinkmon);    zmonitor_destroy (&sourcemon);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");}

intzre_log_msg_test (bool verbose){    printf (" * zre_log_msg: ");    //  @selftest    //  Simple create/destroy test    zre_log_msg_t *self = zre_log_msg_new (0);    assert (self);    zre_log_msg_destroy (&self);    //  Create pair of sockets we can send through    zctx_t *ctx = zctx_new ();    assert (ctx);    void *output = zsocket_new (ctx, ZMQ_DEALER);    assert (output);    zsocket_bind (output, "inproc://selftest");    void *input = zsocket_new (ctx, ZMQ_ROUTER);    assert (input);    zsocket_connect (input, "inproc://selftest");        //  Encode/send/decode and verify each message type    self = zre_log_msg_new (ZRE_LOG_MSG_LOG);    zre_log_msg_set_level (self, 123);    zre_log_msg_set_event (self, 123);    zre_log_msg_set_node (self, 123);    zre_log_msg_set_peer (self, 123);    zre_log_msg_set_time (self, 123);    zre_log_msg_set_data (self, "Life is short but Now lasts for ever");    zre_log_msg_send (&self, output);        self = zre_log_msg_recv (input);    assert (self);    assert (zre_log_msg_level (self) == 123);    assert (zre_log_msg_event (self) == 123);    assert (zre_log_msg_node (self) == 123);    assert (zre_log_msg_peer (self) == 123);    assert (zre_log_msg_time (self) == 123);    assert (streq (zre_log_msg_data (self), "Life is short but Now lasts for ever"));    zre_log_msg_destroy (&self);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

voidzre_peer_connect (zre_peer_t *self, char *reply_to, char *endpoint){    assert (self);    assert (!self->connected);    //  Create new outgoing socket (drop any messages in transit)    self->mailbox = zsocket_new (self->ctx, ZMQ_DEALER);        //  Set our caller 'From' identity so that receiving node knows    //  who each message came from.    zsocket_set_identity (self->mailbox, reply_to);    //  Set a high-water mark that allows for reasonable activity    zsocket_set_sndhwm (self->mailbox, PEER_EXPIRED * 100);        //  Send messages immediately or return EAGAIN    zsocket_set_sndtimeo (self->mailbox, 0);        //  Connect through to peer node    zsocket_connect (self->mailbox, "tcp://%s", endpoint);    self->endpoint = strdup (endpoint);    self->connected = true;    self->ready = false;}

static voidserver_worker (void *args, zctx_t *ctx, void *pipe){    void *worker = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (worker, "inproc://backend");    while (true) {        //  The DEALER socket gives us the reply envelope and message        zmsg_t *msg = zmsg_recv (worker);        zframe_t *identity = zmsg_pop (msg);        zframe_t *content = zmsg_pop (msg);        assert (content);        zmsg_destroy (&msg);        //  Send 0..4 replies back        int reply, replies = randof (5);        for (reply = 0; reply < replies; reply++) {            //  Sleep for some fraction of a second            zclock_sleep (randof (1000) + 1);            zframe_send (&identity, worker, ZFRAME_REUSE + ZFRAME_MORE);            zframe_send (&content, worker, ZFRAME_REUSE);        }        zframe_destroy (&identity);        zframe_destroy (&content);    }}

static void *client_task (void *args){    zctx_t *ctx = zctx_new ();    void *client = zsocket_new (ctx, ZMQ_DEALER);    //  Set random identity to make tracing easier    char identity [10];    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));    zsocket_set_identity (client, identity);    zsocket_connect (client, "tcp://localhost:5570");    zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };    int request_nbr = 0;    while (true) {        //  Tick once per second, pulling in arriving messages        int centitick;        for (centitick = 0; centitick < 100; centitick++) {            zmq_poll (items, 1, 10 * ZMQ_POLL_MSEC);            if (items [0].revents & ZMQ_POLLIN) {                zmsg_t *msg = zmsg_recv (client);                zframe_print (zmsg_last (msg), identity);                zmsg_destroy (&msg);            }        }        zstr_send (client, "request #%d");    }    zctx_destroy (&ctx);    return NULL;}

static zyre_node_t *zyre_node_new (zctx_t *ctx, void *pipe){    zyre_node_t *self = (zyre_node_t *) zmalloc (sizeof (zyre_node_t));    self->ctx = ctx;    self->pipe = pipe;    self->inbox = zsocket_new (ctx, ZMQ_ROUTER);    if (self->inbox == NULL) {        free (self);        return NULL;            //  Interrupted 0MQ call    }    self->port = zsocket_bind (self->inbox, "tcp://*:*");    if (self->port < 0) {        free (self);        return NULL;            //  Interrupted 0MQ call    }    self->beacon = zbeacon_new (self->ctx, ZRE_DISCOVERY_PORT);    if (!self->beacon) {        free (self);        return NULL;            //  Exhausted process sockets    }    self->uuid = zuuid_new ();    self->peers = zhash_new ();    self->peer_groups = zhash_new ();    self->own_groups = zhash_new ();    self->headers = zhash_new ();    zhash_autofree (self->headers);    return self;}

curve_client_t *curve_client_new (zcert_t **cert_p){    curve_client_t *self = (curve_client_t *) zmalloc (sizeof (curve_client_t));    assert (self);    self->ctx = zctx_new ();    self->control = zthread_fork (self->ctx, s_agent_task, NULL);    //  Create separate data socket, send address on control socket    self->data = zsocket_new (self->ctx, ZMQ_PAIR);    assert (self->data);    int rc = zsocket_bind (self->data, "inproc://data-%p", self->data);    assert (rc != -1);    zstr_sendfm (self->control, "inproc://data-%p", self->data);       //  Now send cert on control socket as well    rc = zmq_send (self->control, zcert_public_key (*cert_p), 32, ZMQ_SNDMORE);    assert (rc == 32);    rc = zmq_send (self->control, zcert_secret_key (*cert_p), 32, 0);    assert (rc == 32);        zcert_destroy (cert_p);    return self;}

static void* client_routine(void* arg){  zctx_t* ctx = zctx_new();  void* client = zsocket_new(ctx, ZMQ_REQ);  zsocket_connect(client, "ipc://frontend.ipc");  while (1) {    char* reply;    zstr_send(client, "Hello");    reply = zstr_recv(client);    if (NULL == reply)      break;    fprintf(stdout, "client: %s/n", reply);    free(reply);    sleep(1);  }  zctx_destroy(&ctx);  return NULL;}

// Basic request-reply client using REQ socket//static void *client_task(void *args){	zctx_t *ctx = zctx_new();	void *client = zsocket_new(ctx, ZMQ_REQ);#if (defined (WIN32))	zsocket_connect(client, "tcp://localhost:5672"); // frontend#else	zsocket_connect(client, "ipc://frontend.ipc");#endif	// Send request, get reply	while (1) {		zstr_send(client, "HELLO");		char *reply = zstr_recv(client);		if (!reply) {			break;		}		printf("Client: %s/n", reply);		free(reply);		zclock_sleep(1);	}	zctx_destroy(&ctx);	return NULL;}

//  Worker using REQ socket to do load-balancing//static void *worker_task(void *args){	zctx_t *ctx = zctx_new();	void *worker = zsocket_new(ctx, ZMQ_REQ);#if (defined (WIN32))	zsocket_connect(worker, "tcp://localhost:5673"); // backend#else	zsocket_connect(worker, "ipc://backend.ipc");#endif	//  Tell broker we're ready for work	zframe_t *frame = zframe_new(WORKER_READY, strlen(WORKER_READY));	zframe_send(&frame, worker, 0);	//  Process messages as they arrive	while (1) {		zmsg_t *msg = zmsg_recv(worker);		if (!msg)			break;              //  Interrupted		zframe_print(zmsg_last(msg), "Worker: ");		zframe_reset(zmsg_last(msg), "OK", 2);		zmsg_send(&msg, worker);	}	zctx_destroy(&ctx);	return NULL;}

static agent_t *s_agent_new (zctx_t *ctx, void *pipe){    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));    if (!self)        return NULL;    self->ctx = ctx;    self->pipe = pipe;    self->whitelist = zhash_new ();    if (self->whitelist)        self->blacklist = zhash_new ();    //  Create ZAP handler and get ready for requests    if (self->blacklist)        self->handler = zsocket_new (self->ctx, ZMQ_REP);    if (self->handler) {        if (zsocket_bind (self->handler, ZAP_ENDPOINT) == 0)            zstr_send (self->pipe, "OK");        else            zstr_send (self->pipe, "ERROR");    }    else        s_agent_destroy (&self);    return self;}