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

void read_serial(void * cvoid, zctx_t * context, void * pipe) {  char * buf;  serialconfig_t * config = (serialconfig_t*)cvoid;    FILE * in = config->in; // fopen("/dev/ttyO1", "r");  size_t nbytes=2047;  //  Prepare our context and publisher  buf = (char *) malloc(nbytes+1) ;  fprintf(stderr, "bound/n");  // first line is always garbage  getline(&buf, &nbytes, in);  child_handshake(pipe);  zsocket_destroy(context, pipe);  void* socket = zsocket_new(context, ZMQ_PUB);  zsocket_bind(socket, "inproc://raw_serial");    while ( getline(&buf, &nbytes, in) != -1 ) {#ifdef DEBUG    puts("line:");    puts(buf);#endif    zmsg_t * msg = zmsg_new();    zmsg_pushstr(msg, buf); // does buf need to be copied?    zmsg_send(&msg, socket);  }  fprintf(stderr, "error reading from stdin/n");  zsocket_destroy(context, socket);}

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

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) {        zsocket_destroy(self->ctx, self->inbox);        free (self);        return NULL;            //  Interrupted 0MQ call    }    self->beacon = zbeacon_new (self->ctx, ZRE_DISCOVERY_PORT);    if (!self->beacon) {        zsocket_destroy(self->ctx, self->inbox);        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;}

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

PoolFrontend::~PoolFrontend() {		printf("PoolFrontend stopped./n");		zsocket_destroy(mCtx, mRouter);	zsocket_destroy(mCtx, mDealer);		zctx_destroy(&mCtx);	}

voidzre_peer_connect (zre_peer_t *self, char *reply_to, char *endpoint){    //  If already connected, destroy old socket and start again    if (self->connected)        zsocket_destroy (self->ctx, self->mailbox);    //  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;}

voidprob_disconnect(prob_client_t pc){    if (zsocket_disconnect(pc->zocket, "%s", pc->file) != 0) Warning(info, "Could not disconnect from zocket %s", pc->file);    zsocket_destroy(pc->ctx, pc->zocket);    zctx_destroy(&(pc->ctx));}

static gbooleando_heartbeat (GPPWorker *self){  GPPWorkerPrivate *priv = GET_PRIV (self);  if (--priv->liveness == 0) {    g_warning ("W: heartbeat failure, can't reach queue/n");    g_warning ("W: reconnecting in %zd msec.../n", priv->interval);    g_source_remove (priv->frontend_source);    priv->frontend_source = 0;    g_io_channel_unref (priv->frontend_channel);    if (priv->interval < INTERVAL_MAX)      priv->interval *= 2;    zsocket_destroy (priv->ctx, priv->frontend);    g_timeout_add (priv->interval, (GSourceFunc) do_start, self);    return FALSE;  }  zframe_t *frame = zframe_new (PPP_HEARTBEAT, 1);  zframe_send (&frame, priv->frontend, 0);  /* We need to do that for some reason ... */  check_socket_activity (priv->frontend_channel, G_IO_IN, self);  return TRUE;}

//  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);    zstr_send (*server, "Hello, World");    zpoller_t *poller = zpoller_new (*client, NULL);    bool success = (zpoller_wait (poller, 200) == *client);    zpoller_destroy (&poller);    zsocket_destroy (ctx, *client);    zsocket_destroy (ctx, *server);    *server = zsocket_new (ctx, ZMQ_PUSH);    *client = zsocket_new (ctx, ZMQ_PULL);    return success;}

/* closeZMQ will destroy the context and  * associated socket */static void closeZMQ(instanceData* pData) {    errmsg.LogError(0, NO_ERRCODE, "closeZMQ called");    if(s_context && pData->socket) {        if(pData->socket != NULL) {            zsocket_destroy(s_context, pData->socket);        }    }}

char *  test_line() {  zctx_t *context = zctx_new();  lineconfig_t config;  config.line_id = 2;  config_t base = { "uuid", "inproc:broker", "inproc:portwatcher",                         "inproc:registration", "http://something.com" };  config.base_config = &base;  void * line_in = zsocket_new(context, ZMQ_PUB);  zsocket_bind (line_in, "inproc://line");  // filter(identity, serial_reader, remote, context);    zclock_log("pre-fork");  void * pipe = zthread_fork(context, line_listener, &config);  // TODO a real test  zstr_send(pipe, "ping");  char * result = zstr_recv(pipe);  mu_assert_str("pong response", "pong", result);  free(result);  // do some channel switching  int i,j;  for(i=0;i<100;i++) {    // first channel    for(j=0;j<3; j++) {      val_msg(line_in, "foo", 12);    }    for(j=0;j<3;j++) {      val_msg(line_in, "bar", 99);    }  }  zstr_send(line_in, "DESTROY");  // temp  printf("destroy message sent/n");  result = zstr_recv(pipe);  mu_assert_str("destroy response", "ok", result);  free(result);  zsocket_destroy(context,pipe);  printf("freed pipe/n");  zsocket_destroy(context, line_in);  zctx_destroy(&context);  return NULL;}

void s_mdp_client_connect_to_broker (mdp_client_t *self){    if (self->client)        zsocket_destroy (self->ctx, self->client);    self->client = zsocket_new (self->ctx, ZMQ_DEALER);    zmq_connect (self->client, self->broker);    if (self->verbose)        zclock_log ("I: connecting to broker at %s...", self->broker);}

static void client_task (void *args){    client_state* state = (client_state*) args;    void *frontend = zsocket_new (state->context, ZMQ_DEALER);    zsocket_set_identity (frontend, state->identity);    zsocket_connect (frontend,state->server_url);        void *backend = zsocket_new (state->context, ZMQ_DEALER);    zsocket_bind (backend, "inproc://backend");    zthread_fork (state->context, worker_task, state);        client_loop(state, frontend, backend);        zsocket_destroy(state->context, &frontend);    zsocket_destroy(state->context, &backend);    zctx_destroy (&state->context);}

void *_rrwrk_create_socket(zctx_t *ctx, void *socket, int type){    if (socket) {        zsocket_destroy(ctx, socket); //** Destroy existing socket    }    socket = zsocket_new(ctx, type);    assert(socket);    return socket;}

void collabclient_sessionReconnect( void* ccvp ){#ifdef BUILD_COLLAB    cloneclient_t* cc = (cloneclient_t*)ccvp;    zsocket_destroy( cc->ctx, cc->snapshot   );    zsocket_destroy( cc->ctx, cc->subscriber );    zsocket_destroy( cc->ctx, cc->publisher  );    collabclient_remakeSockets( cc );    cc->publisher_sendseq = 1;    FontView* fv = FontViewFindUI( FontViewFind_byCollabPtr, cc );    if( fv )    {	collabclient_sessionJoin( cc, fv );    }#endif}

int remove_rule(zctx_t * context, zhash_t * hash, char * rule_id) {  void * pipe = zhash_lookup(hash, rule_id);  if(!pipe) return 0;  zstr_send(pipe, "Destroy");  recv_sync("ok", pipe);  zhash_delete(hash, rule_id);  zsocket_destroy(context, pipe);  zclock_log("rule %s destroyed", rule_id);  return 1;}

//  ---------------------------------------------------------------------//  Destroy nodevoid lsd_destroy(lsd_handle_t* self) {	assert(self);	zstr_send (self->pipe, "STOP");	zre_node_destroy (&(self->interface));	zsocket_destroy (self->ctx, self->pipe);	if(self->owns_ctx)		zctx_destroy (&self->ctx);	free(self);}

static ints_subscriber (zloop_t *loop, zmq_pollitem_t *poller, void *args){    clonesrv_t *self = (clonesrv_t *) args;    //  Get state snapshot if necessary    if (self->kvmap == NULL) {        self->kvmap = zhash_new ();        void *snapshot = zsocket_new (self->ctx, ZMQ_DEALER);        zsocket_connect (snapshot, "tcp://localhost:%d", self->peer);        zclock_log ("I: asking for snapshot from: tcp://localhost:%d",                    self->peer);        zstr_sendm (snapshot, "ICANHAZ?");        zstr_send (snapshot, ""); // blank subtree to get all        while (true) {            kvmsg_t *kvmsg = kvmsg_recv (snapshot);            if (!kvmsg)                break;          //  Interrupted            if (streq (kvmsg_key (kvmsg), "KTHXBAI")) {                self->sequence = kvmsg_sequence (kvmsg);                kvmsg_destroy (&kvmsg);                break;          //  Done            }            kvmsg_store (&kvmsg, self->kvmap);        }        zclock_log ("I: received snapshot=%d", (int) self->sequence);        zsocket_destroy (self->ctx, snapshot);    }    //  Find and remove update off pending list    kvmsg_t *kvmsg = kvmsg_recv (poller->socket);    if (!kvmsg)        return 0;    if (strneq (kvmsg_key (kvmsg), "HUGZ")) {        if (!s_was_pending (self, kvmsg)) {            //  If active update came before client update, flip it            //  around, store active update (with sequence) on pending            //  list and use to clear client update when it comes later            zlist_append (self->pending, kvmsg_dup (kvmsg));        }        //  If update is more recent than our kvmap, apply it        if (kvmsg_sequence (kvmsg) > self->sequence) {            self->sequence = kvmsg_sequence (kvmsg);            kvmsg_store (&kvmsg, self->kvmap);            zclock_log ("I: received update=%d", (int) self->sequence);        }        else            kvmsg_destroy (&kvmsg);    }    else        kvmsg_destroy (&kvmsg);    return 0;}

voidzre_peer_disconnect (zre_peer_t *self){    //  If connected, destroy socket and drop all pending messages    assert (self);    if (self->connected) {        zsocket_destroy (self->ctx, self->mailbox);        free (self->endpoint);        self->endpoint = NULL;        self->connected = false;    }}

void collabclient_free( void** ccvp ){#ifdef BUILD_COLLAB        cloneclient_t* cc = (cloneclient_t*)*ccvp;    if( !cc )	return;    if( cc->magic_number != MAGIC_VALUE )    {	fprintf(stderr,"collabclient_free() called on an invalid client object!/n");	return;    }    cc->magic_number = MAGIC_VALUE + 1;    {	int fd = 0;	size_t fdsz = sizeof(fd);	int rc = zmq_getsockopt( cc->subscriber, ZMQ_FD, &fd, &fdsz );	GDrawRemoveReadFD( 0, fd, cc );    }            zsocket_destroy( cc->ctx, cc->snapshot   );    zsocket_destroy( cc->ctx, cc->subscriber );    zsocket_destroy( cc->ctx, cc->publisher  );    BackgroundTimer_remove( cc->roundTripTimer );    FontView* fv = FontViewFind( FontViewFind_byCollabPtr, cc );    if( fv )    {	fv->b.collabClient = 0;    }    zhash_destroy (&cc->kvmap);    free(cc->address);    free(cc);    *ccvp = 0;#endif}

static void op_fini (void *impl){    ctx_t *ctx = impl;    assert (ctx->magic == MODHANDLE_MAGIC);    if (ctx->sock)        zsocket_destroy (ctx->zctx, ctx->sock);    if (ctx->uuid)        free (ctx->uuid);    if (ctx->uri)        free (ctx->uri);    ctx->magic = ~MODHANDLE_MAGIC;    free (ctx);}

static voidserver_destroy (server_t **self_p){    assert (self_p);    if (*self_p) {        server_t *self = *self_p;        zsocket_destroy (self->ctx, self->dealer);        fmq_msg_destroy (&self->request);        fmq_msg_destroy (&self->reply);        free (self->endpoint);                free (self);        *self_p = NULL;    }}

static bool ConnectBitcoin() {		const proto::ServerInfo& sinfo = gServerInfo;	printf("Connecting to bitcoin: %s:%d .../n", sinfo.host().c_str(), sinfo.router());		zsocket_destroy(gCtx, gServer);	gServer = zsocket_new(gCtx, ZMQ_DEALER);	int err = zsocket_connect(gServer, "tcp://%s:%d", sinfo.host().c_str(), sinfo.router());	if(err){		printf("ERROR: invalid hostname and/or port./n");		return false;	}		return true;}

int main (int argc, char *argv []){    //  Initialize context for talking to tasks    zctx_t *ctx = zctx_new ();    zctx_set_linger (ctx, 100);        //  Get number of nodes N to simulate    //  We need 3 x N x N + 3N file handles    int max_nodes = 10;    int nbr_nodes = 0;    if (argc > 1)        max_nodes = atoi (argv [1]);    int max_iterations = -1;    int nbr_iterations = 0;    if (argc > 2)        max_iterations = atoi (argv [2]);    //  We address nodes as an array of pipes    void **pipes = zmalloc (sizeof (void *) * max_nodes);    //  We will randomly start and stop node threads    while (!zctx_interrupted) {        uint index = randof (max_nodes);        //  Toggle node thread        if (pipes [index]) {            zstr_send (pipes [index], "STOP");            zsocket_destroy (ctx, pipes [index]);            pipes [index] = NULL;            zclock_log ("I: Stopped node (%d running)", --nbr_nodes);        }        else {            pipes [index] = zthread_fork (ctx, node_task, NULL);            zclock_log ("I: Started node (%d running)", ++nbr_nodes);        }        nbr_iterations++;        if (max_iterations > 0 && nbr_iterations >= max_iterations)            break;        //  Sleep ~750 msecs randomly so we smooth out activity        zclock_sleep (randof (500) + 500);    }    zclock_log ("I: Stopped tester (%d iterations)", nbr_iterations);        //  Does not actually terminate properly... :-/    //  zctx_destroy (&ctx);    free (pipes);    return 0;}

static voids_agent_destroy (agent_t **self_p){    assert (self_p);    if (*self_p) {        agent_t *self = *self_p;        zhash_destroy (&self->passwords);        zhash_destroy (&self->whitelist);        zhash_destroy (&self->blacklist);        zcertstore_destroy (&self->certstore);        zsocket_unbind (self->handler, ZAP_ENDPOINT);        zsocket_destroy (self->ctx, self->handler);        free (self);        *self_p = NULL;    }}

void s_mdp_worker_connect_to_broker (mdp_worker_t *self){    if (self->worker)        zsocket_destroy (self->ctx, self->worker);    self->worker = zsocket_new (self->ctx, ZMQ_DEALER);    zmq_connect (self->worker, self->broker);    if (self->verbose)        zclock_log ("I: connecting to broker at %s...", self->broker);    //  Register service with broker    s_mdp_worker_send_to_broker (self, MDPW_READY, self->service, NULL);    //  If liveness hits zero, worker is considered disconnected    self->liveness = HEARTBEAT_LIVENESS;    self->heartbeat_at = zclock_time () + self->heartbeat;}

void add_monitor(rulepackage_t * rpkg, zmsg_t * reply) {  // unconditionally fork a monitor for each line  // they'll die when they get a channel change  int i;  for(i=1; i<4; i++) {     monitorconfig_t * mconf = malloc(sizeof(monitorconfig_t));    mconf->line_id = i;    mconf->source_worker = rpkg->servicename;    mconf->out_socket = rpkg->base_config->portwatcher_endpoint;    mconf->channel = rpkg->channel;    void * monitor_pipe = zthread_fork(rpkg->context, watch_port, (void*)mconf);    send_sync("ping", monitor_pipe);    recv_sync("pong", monitor_pipe);    zsocket_destroy(rpkg->context, monitor_pipe);  }  zmsg_pushstr(reply, "ok");}

intzsocket_test (Bool verbose){    printf (" * zsocket: ");    //  @selftest    zctx_t *ctx = zctx_new ();    assert (ctx);    //  Create a detached thread, let it run    char *interf = "*";    char *domain = "localhost";    int service = 5560;    void *writer = zsocket_new (ctx, ZMQ_PUSH);    assert (writer);    void *reader = zsocket_new (ctx, ZMQ_PULL);    assert (reader);    assert (streq (zsocket_type_str (writer), "PUSH"));    assert (streq (zsocket_type_str (reader), "PULL"));    int rc = zsocket_bind (writer, "tcp://%s:%d", interf, service);    assert (rc == service);    rc = zsocket_connect (reader, "tcp://%s:%d", domain, service);    assert (rc == 0);    zstr_send (writer, "HELLO");    char *message = zstr_recv (reader);    assert (message);    assert (streq (message, "HELLO"));    free (message);    int port = zsocket_bind (writer, "tcp://%s:*", interf);    assert (port >= ZSOCKET_DYNFROM && port <= ZSOCKET_DYNTO);    assert (zsocket_poll (writer, 100) == FALSE);    rc = zsocket_connect (reader, "txp://%s:%d", domain, service);    assert (rc == -1);    zsocket_destroy (ctx, writer);    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}