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

zframe_t *zmsg_last (zmsg_t *self){    assert (self);    zframe_t *frame = (zframe_t *) zlist_first (self->frames);    while (frame) {        zframe_t *next = (zframe_t *) zlist_next (self->frames);        if (!next)            break;        frame = next;    }    return frame;}

static voids_timer_remove (zloop_t *self, int timer_id){    s_timer_t *timer = (s_timer_t *) zlist_first (self->timers);    while (timer) {        if (timer->timer_id == timer_id) {            zlist_remove (self->timers, timer);            free (timer);            break;        }        timer = (s_timer_t *) zlist_next (self->timers);    }}

static zsync_peer_t *zsync_node_peers_lookup (zsync_node_t *self, char *uuid){    assert (self);    zsync_peer_t *peer = zlist_first (self->peers);    while (peer) {        if (streq (zsync_peer_uuid (peer), uuid)) {            return peer;        }        peer = zlist_next (self->peers);    }    return NULL;}

static int invoke_cbs (flux_t *h, int64_t j, json_object *jcb, int errnum){    int rc = 0;    cb_pair_t *c = NULL;    jscctx_t *ctx = getctx (h);    for (c = zlist_first (ctx->callbacks); c; c = zlist_next (ctx->callbacks)) {        if (c->cb (jcb, c->arg, errnum) < 0) {            flux_log (h, LOG_ERR, "callback returns an error");            rc = -1;        }    }    return rc;}

END_TEST//  --------------------------------------------------------------------------/// Try to _pop () two succeesding listsSTART_TEST(test_msg_pop_l_double){    sam_selftest_introduce ("test_msg_pop_l_double");    sam_selftest_introduce ("test_msg_get_l_double");    zmsg_t *zmsg = zmsg_new ();    if (zmsg_pushstr (zmsg, "two") ||        zmsg_pushstr (zmsg, "1")   ||        zmsg_pushstr (zmsg, "one") ||        zmsg_pushstr (zmsg, "1")) {        ck_abort_msg ("could not build zmsg");    }    sam_msg_t *msg = sam_msg_new (&zmsg);    ck_assert_int_eq (sam_msg_size (msg), 4);    zlist_t *list1, *list2;    int rc = sam_msg_pop (msg, "ll", &list1, &list2);    ck_assert_int_eq (rc, 0);    ck_assert_int_eq (sam_msg_size (msg), 0);    ck_assert_int_eq (zlist_size (list1), 1);    ck_assert_int_eq (zlist_size (list2), 1);    ck_assert_str_eq (zlist_first (list1), "one");    ck_assert_str_eq (zlist_first (list2), "two");    zlist_destroy (&list1);    zlist_destroy (&list2);    sam_msg_destroy (&msg);}

static voids_worker_ready (worker_t *self, zlist_t *workers){    worker_t *worker = (worker_t *) zlist_first (workers);    while (worker) {        if (streq (self->identity, worker->identity)) {            zlist_remove (workers, worker);            s_worker_destroy (&worker);            break;        }        worker = (worker_t *) zlist_next (workers);    }    zlist_append (workers, self);}

//  If message was already on pending list, remove it and return true,//  else return false.static ints_was_pending (clonesrv_t *self, kvmsg_t *kvmsg){    kvmsg_t *held = (kvmsg_t *) zlist_first (self->pending);    while (held) {        if (memcmp (kvmsg_uuid (kvmsg),                    kvmsg_uuid (held), sizeof (uuid_t)) == 0) {            zlist_remove (self->pending, held);            return true;        }        held = (kvmsg_t *) zlist_next (self->pending);    }    return false;}

voidpass_update (char *sender, zlist_t *fmetadata) {    printf ("[ST] PASS_UPDATE from %s: %"PRId64"/n", sender, zlist_size (fmetadata));    uint64_t size = 0;    zlist_t *paths = zlist_new ();    zs_fmetadata_t *meta = zlist_first (fmetadata);    while (meta) {        zlist_append (paths, zs_fmetadata_path (meta));        size += zs_fmetadata_size (meta);        meta = zlist_next (fmetadata);    }    zsync_agent_send_request_files (agent, sender, paths, size);}

halutils_err_e disp_table_remove_all (disp_table_t *self){    assert (self);    zlist_t *hash_keys = zhash_keys (self->table_h);    void * table_item = zlist_first (hash_keys);    for (; table_item; table_item = zlist_next (hash_keys)) {        zhash_delete (self->table_h, (char *) table_item);    }    zlist_destroy (&hash_keys);    return HALUTILS_SUCCESS;}

const char *flux_kvsitr_next (flux_kvsitr_t *itr){    const char *name = NULL;    if (itr) {        if (itr->reset)            name = zlist_first (itr->keys);        else            name = zlist_next (itr->keys);        if (name)            itr->reset = false;    }    return name;}

static int subprocess_run_hooks (struct subprocess *p, zlist_t *hooks){    subprocess_cb_f fn = zlist_first (hooks);    int rc = 0;    subprocess_ref (p);    while (fn) {        if ((rc = fn (p)) < 0)            goto done;        fn = zlist_next (hooks);    }done:    subprocess_unref (p);    return (rc);}

voidzcertstore_fprint (zcertstore_t *self, FILE *file){    if (self->location)        fprintf (file, "Certificate store at %s:/n", self->location);    else        fprintf (file, "Certificate store/n");    zcert_t *cert = (zcert_t *) zlist_first (self->cert_list);    while (cert) {        zcert_fprint (cert, file);        cert = (zcert_t *) (zcert_t *) zlist_next (self->cert_list);    }}

voidzctx_destroy (zctx_t **self_p){    assert (self_p);    if (*self_p) {        zctx_t *self = *self_p;        while (zlist_size (self->sockets))            zctx__socket_destroy (self, zlist_first (self->sockets));        zlist_destroy (&self->sockets);        if (self->main && self->context)            zmq_term (self->context);        free (self);        *self_p = NULL;    }}

static voidmount_sub_purge (mount_t *self, client_t *client){    sub_t *sub = (sub_t *) zlist_first (self->subs);    while (sub) {        if (sub->client == client) {            sub_t *next = (sub_t *) zlist_next (self->subs);            zlist_remove (self->subs, sub);            sub_destroy (&sub);            sub = next;        }        else            sub = (sub_t *) zlist_next (self->subs);    }}

static bool subscription_match (client_t *c, const flux_msg_t *msg){    subscription_t *sub;    const char *topic;    if (flux_msg_get_topic (msg, &topic) < 0)        return false;    sub = zlist_first (c->subscriptions);    while (sub) {        if (!strncmp (topic, sub->topic, strlen (sub->topic)))            return true;        sub = zlist_next (c->subscriptions);    }    return false;}

voidzloop_poller_set_tolerant (zloop_t *self, zmq_pollitem_t *item){    assert (self);    assert (item->socket || item->fd);    //  Find matching poller(s) and mark as tolerant    s_poller_t *poller = (s_poller_t *) zlist_first (self->pollers);    while (poller) {        if ((item->socket && item->socket == poller->item.socket)        ||  (item->fd     && item->fd     == poller->item.fd))            poller->tolerant = true;                    poller = (s_poller_t *) zlist_next (self->pollers);    }}

static order_t *s_engine_match_sell_order (engine_t *self){    assert (self);    order_t *order = (order_t *) zlist_first (self->sell_orders);    order_t *best_order = order;    while (order) {        if (order->price < best_order->price)            best_order = order;        order = (order_t *) zlist_next (self->sell_orders);    }    return best_order;}

static char *zsync_node_zyre_uuid (zsync_node_t *self, char *sender){    assert (self);    assert (sender);    zlist_t *keys = zhash_keys (self->zyre_peers);    char *key = zlist_first (keys);    while (key) {        zsync_peer_t *peer = zhash_lookup (self->zyre_peers, key);        if (peer && streq (sender, zsync_peer_uuid (peer))) {           return key;        }        key = zlist_next (keys);    }    return NULL;}

//  Apply configuration tree://   * apply client configuration//   * print any echo items in top-level sections//   * apply sections that match methodsstatic voidclient_apply_config (client_t *self){    //  Apply echo commands and class methods    fmq_config_t *section = fmq_config_child (self->config);    while (section) {        fmq_config_t *entry = fmq_config_child (section);        while (entry) {            if (streq (fmq_config_name (entry), "echo"))                zclock_log (fmq_config_value (entry));            entry = fmq_config_next (entry);        }        if (streq (fmq_config_name (section), "subscribe")) {            char *path = fmq_config_resolve (section, "path", "?");            //  Store subscription along with any previous ones                                   //  Check we don't already have a subscription for this path                          self->sub = (sub_t *) zlist_first (self->subs);                                       while (self->sub) {                                                                       if (streq (path, self->sub->path))                                                        return;                                                                           self->sub = (sub_t *) zlist_next (self->subs);                                    }                                                                                     //  Subscription path must start with '/'                                             //  We'll do better error handling later                                              assert (*path == '/');                                                                                                                                                      //  New subscription, store it for later replay                                       char *inbox = fmq_config_resolve (self->config, "client/inbox", ".inbox");            self->sub = sub_new (self, inbox, path);                                              zlist_append (self->subs, self->sub);                                             }        else        if (streq (fmq_config_name (section), "set_inbox")) {            char *path = fmq_config_resolve (section, "path", "?");            fmq_config_path_set (self->config, "client/inbox", path);        }        else        if (streq (fmq_config_name (section), "set_resync")) {            long enabled = atoi (fmq_config_resolve (section, "enabled", ""));            //  Request resynchronization from server                                          fmq_config_path_set (self->config, "client/resync", enabled? "1" :"0");        }        section = fmq_config_next (section);    }    client_config_self (self);}

static voidstore_client_subscription (server_t *self, client_t *client){    //  Find mount point with longest match to subscription             char *path = fmq_msg_path (client->request);                                                                                            mount_t *check = (mount_t *) zlist_first (self->mounts);            mount_t *mount = check;                                             while (check) {                                                         //  If check->alias is prefix of path and alias is                  //  longer than current mount then we have a new mount              if (strncmp (path, check->alias, strlen (check->alias)) == 0        &&  strlen (check->alias) > strlen (mount->alias))                      mount = check;                                                  check = (mount_t *) zlist_next (self->mounts);                  }                                                                   mount_sub_store (mount, client, client->request);               }

/* Received an event message from broker. * Find all subscribers and deliver. */static void event_cb (flux_t h, flux_msg_watcher_t *w,                      const flux_msg_t *msg, void *arg){    ctx_t *ctx = arg;    client_t *c;    c = zlist_first (ctx->clients);    while (c) {        if (subscription_match (c, msg)) {            if (client_send (c, msg) < 0) { /* FIXME handle errors */                flux_log (h, LOG_ERR, "%s: client_send %s: %s",                          __FUNCTION__, zuuid_str (c->uuid), strerror (errno));                errno = 0;            }        }        c = zlist_next (ctx->clients);    }}

voidzloop_reader_end (zloop_t *self, zsock_t *sock){    assert (self);    assert (sock);    s_reader_t *reader = (s_reader_t *) zlist_first (self->readers);    while (reader) {        if (reader->sock == sock) {            zlist_remove (self->readers, reader);            free (reader);            self->need_rebuild = true;        }        reader = (s_reader_t *) zlist_next (self->readers);    }    if (self->verbose)        zsys_debug ("zloop: cancel %s reader", zsock_type_str (sock));}

static void cleanup (void){    const struct cleaner *c;    pthread_mutex_lock(&mutex);    if ( ! cleanup_list || cleaner_pid != getpid())        goto out;    c = zlist_first(cleanup_list);    while (c){        if (c && c->fun){            c->fun(c);        }        c = zlist_next(cleanup_list);    }    zlist_destroy(&cleanup_list);    cleanup_list = NULL;out:    pthread_mutex_unlock(&mutex);}

static voidclient_apply_config (client_t *self){    //  Get standard client configuration    self->heartbeat = atoi (        fmq_config_resolve (self->config, "client/heartbeat", "1")) * 1000;    //  Apply echo commands and class methods    fmq_config_t *section = fmq_config_child (self->config);    while (section) {        fmq_config_t *entry = fmq_config_child (section);        while (entry) {            if (streq (fmq_config_name (entry), "echo"))                puts (fmq_config_value (entry));            entry = fmq_config_next (entry);        }        if (streq (fmq_config_name (section), "subscribe")) {            char *path = fmq_config_resolve (section, "path", "?");            //  Store subscription along with any previous ones                     //  Check we don't already have a subscription for this path            sub_t *sub = (sub_t *) zlist_first (self->subs);                        while (sub) {                                                               if (streq (path, sub->path))                                                return;                                                             sub = (sub_t *) zlist_next (self->subs);                            }                                                                       //  Subscription path must start with '/'                               //  We'll do better error handling later                                assert (*path == '/');                                                                                                                          //  New subscription, so store it for later replay                      sub = sub_new (self, path);                                             zlist_append (self->subs, sub);                                                                                                                 //  If we're connected, then also send to server                        if (self->connected) {                                                      fmq_msg_path_set (self->request, path);                                 self->next_event = subscribe_event;                                 }                                                                   }        section = fmq_config_next (section);    }}

voidzctx_destroy (zctx_t **self_p){    assert (self_p);    if (*self_p) {        zctx_t *self = *self_p;        while (zlist_size (self->sockets)) {            void *socket = zlist_first (self->sockets);            zsockopt_set_linger (socket, self->linger);            zmq_close (socket);            zlist_remove (self->sockets, socket);        }        zlist_destroy (&self->sockets);        if (self->main && self->context)            zmq_term (self->context);        free (self);        *self_p = NULL;    }}

rsRetVal outputCZMQ(uchar* msg, instanceData* pData) {	DEFiRet;	/* if auth or socket is missing, initialize */	if((NULL == pData->sock) || (NULL == pData->authActor)) {		CHKiRet(initCZMQ(pData));	}	/* if we have a ZMQ_PUB sock and topics, send once per topic */	if (pData->sockType == ZMQ_PUB && pData->topics) {		char *topic = zlist_first(pData->topics);		while (topic) {			if (pData->topicFrame) {				int rc = zstr_sendx(pData->sock, topic, (char*)msg, NULL);				if (rc != 0) {					pData->sendError = true;					ABORT_FINALIZE(RS_RET_SUSPENDED);				}			} 			else {				int rc = zstr_sendf(pData->sock, "%s%s", topic, (char*)msg);				if (rc != 0) {					pData->sendError = true;					ABORT_FINALIZE(RS_RET_SUSPENDED);				}			}			topic = zlist_next(pData->topics);		}	} 	/* otherwise do a normal send */	else {		int rc = zstr_send(pData->sock, (char*)msg);		if (rc != 0) {			pData->sendError = true;			ABORT_FINALIZE(RS_RET_SUSPENDED);		}	}finalize_it:	RETiRet;}

static void freectx (void *arg){    ctx_t *ctx = arg;    zlist_t *keys = zhash_keys (ctx->ping_requests);    char *key = zlist_first (keys);    while (key) {        zmsg_t *zmsg = zhash_lookup (ctx->ping_requests, key);        zmsg_destroy (&zmsg);        key = zlist_next (keys);    }    zlist_destroy (&keys);    zhash_destroy (&ctx->ping_requests);    zmsg_t *zmsg;    while ((zmsg = zlist_pop (ctx->clog_requests)))        zmsg_destroy (&zmsg);    zlist_destroy (&ctx->clog_requests);    free (ctx);}

intztns_append_list (ztns_t *self, zlist_t *data, ztns_list_foreach_fn *fn){    ztns_t *items = ztns_new ();    assert (items);    if (!items)        return -1;    void * item = zlist_first (data);    while (item) {        int rc = fn (items, item);        if (0 != rc) {            ztns_destroy (&items);            return rc;        }        item = zlist_next (data);    }    int rc = s_append (self, items->data, ']');    ztns_destroy (&items);    return rc;}

voidzinterface_test(bool verbose){    printf(" * zinterface: ");    zlist_t *interfaces = zinterface_list();    assert(interfaces);    if (verbose) {        printf("Len: %zu/n", zlist_size(interfaces));        zinterface_t *iface = zlist_first(interfaces);        while (iface) {            printf ("%s/t%s/t%s/t%s/n", zinterface_name(iface), zinterface_address(iface),                zinterface_netmask(iface), zinterface_broadcast(iface));            iface = zlist_next(interfaces);        }    }    zlist_destroy(&interfaces);    printf("OK/n");}