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

int jsc_notify_status_obj (flux_t h, jsc_handler_obj_f func, void *d){    int rc = -1;    cb_pair_t *c = NULL;     jscctx_t *ctx = NULL;     if (!func)         goto done;    if (reg_newjob_hdlr (h, new_job_cb) == -1) {        flux_log (h, LOG_ERR, "jsc_notify_status: reg_newjob_hdlr failed");        goto done;    }     ctx = getctx (h);    c = (cb_pair_t *) xzmalloc (sizeof(*c));    c->cb = func;    c->arg = d;     if (zlist_append (ctx->callbacks, c) < 0)         goto done;     zlist_freefn (ctx->callbacks, c, free, true);       rc = 0;done:    return rc;}

static voidhandle_request (client_t *self){    const char *service_name = mdp_msg_service(self->message);    if(strstr(service_name, "mmi.")) {        handle_mmi(self, service_name);        return;    }    // Create a fresh instance of mdp_msg_t to append to the list of requests.    mdp_msg_t *msg = mdp_msg_new();    // routing id, messageid, service, body    mdp_msg_set_routing_id(msg, mdp_msg_routing_id(self->message));    mdp_msg_set_id(msg, mdp_msg_id(self->message));    mdp_msg_set_service(msg, service_name);    zmsg_t *body = mdp_msg_get_body(self->message);    mdp_msg_set_body(msg, &body);    service_t *service = s_service_require(self->server, service_name);    zlist_append(service->requests, msg);    s_service_dispatch(service);}

void send_frames_at_server(zhash_t *frames, void *worker,  enum SEND_TYPE type, int n, ...) {    char *key;    va_list valist;    int i;    va_start(valist, n);    zlist_t *names = zlist_new();    for(i=0; i < n; i++ ) {        key = va_arg(valist, char *);        zframe_t *frame = (zframe_t *)zhash_lookup(frames, key);        assert(frame!=NULL);         assert(zframe_is(frame));        zlist_append(names, key);        if( i == n-1 && type==SEND_FINAL)  {            zframe_send(&frame, worker, ZFRAME_REUSE);        } else            zframe_send(&frame, worker, ZFRAME_REUSE + ZFRAME_MORE);    }    va_end(valist);    if(DEBUG_MODE) print_out_hash_in_order(frames, names);    zlist_purge(names);    zlist_destroy(&names);}

static ints_collector (zloop_t *loop, zmq_pollitem_t *poller, void *args){    clonesrv_t *self = (clonesrv_t *) args;    kvmsg_t *kvmsg = kvmsg_recv (poller->socket);    if (kvmsg) {        if (self->master) {            kvmsg_set_sequence (kvmsg, ++self->sequence);            kvmsg_send (kvmsg, self->publisher);            int ttl = atoi (kvmsg_get_prop (kvmsg, "ttl"));            if (ttl)                kvmsg_set_prop (kvmsg, "ttl",                    "%" PRId64, zclock_time () + ttl * 1000);            kvmsg_store (&kvmsg, self->kvmap);            zclock_log ("I: publishing update=%d", (int) self->sequence);        }        else {            //  If we already got message from master, drop it, else            //  hold on pending list            if (s_was_pending (self, kvmsg))                kvmsg_destroy (&kvmsg);            else                zlist_append (self->pending, kvmsg);        }    }    return 0;}

static voidpeering_raise (peering_t *self){    vocket_t *vocket = self->vocket;    driver_t *driver = self->driver;    if (driver->verbose)        zclock_log ("I: (tcp) bring up peering to %s", self->address);    if (!self->alive) {        self->alive = TRUE;        zlist_append (vocket->live_peerings, self);        //  Send ZMTP handshake, which is an empty message        zmq_msg_t msg;        zmq_msg_init_size (&msg, 0);        s_queue_output (self, &msg, FALSE);        //  If we can now route to peerings, start reading from msgpipe        if (zlist_size (vocket->live_peerings) == vocket->min_peerings) {            //  Ask reactor to start monitoring vocket's msgpipe pipe            zmq_pollitem_t item = { vocket->msgpipe, 0, ZMQ_POLLIN, 0 };            zloop_poller (driver->loop, &item, s_vocket_input, vocket);        }    }}

voidzcertstore_insert (zcertstore_t *self, zcert_t **cert_p){    zlist_append (self->cert_list, *cert_p);    zhash_insert (self->cert_hash, zcert_public_txt (*cert_p), *cert_p);    *cert_p = NULL;             //  We own this now}

zlist_t *get_update (uint64_t from_state){     printf("[ST] GET_UPDATE/n");    zlist_t *filemeta_list = zlist_new ();        DIR *dir;    struct dirent *ent;    if ((dir = opendir ("./syncfolder")) != NULL) {        /* print all the files and directories within directory */        while ((ent = readdir (dir)) != NULL) {            if (strcmp (ent->d_name, ".") != 0 && strcmp (ent->d_name, "..") != 0) {                struct stat st;                stat(ent->d_name, &st);                zs_fmetadata_t *fmetadata = zs_fmetadata_new ();                zs_fmetadata_set_path (fmetadata, "%s", ent->d_name);                zs_fmetadata_set_size (fmetadata, st.st_size);                zs_fmetadata_set_operation (fmetadata, ZS_FILE_OP_UPD);                zs_fmetadata_set_timestamp (fmetadata, st.st_ctime);                zs_fmetadata_set_checksum (fmetadata, 0x3312AFFDE12);                zlist_append(filemeta_list, fmetadata);            }        }        closedir (dir);    }     if (zlist_size (filemeta_list) > 0) {        return filemeta_list;    }    else {        return NULL;    }}

intzloop_poller (zloop_t *self, zmq_pollitem_t *item, zloop_fn handler, void *arg){    assert (self);    if (!item->socket && !item->fd)        return -1;    if (item->socket)        if (streq (zsocket_type_str (item->socket), "UNKNOWN"))            return -1;    s_poller_t *poller = s_poller_new (item, handler, arg);    if (poller) {        if (zlist_append (self->pollers, poller))            return -1;        self->dirty = true;        if (self->verbose)            zclock_log ("I: zloop: register %s poller (%p, %d)",                item->socket? zsocket_type_str (item->socket): "FD",                item->socket, item->fd);        return 0;    }    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_unbounded (remote);    if (zsock_connect (remote, "%s", endpoint)) {        zsys_warning ("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);}

static voids_win32_populate_entry (zdir_t *self, WIN32_FIND_DATA *entry){    if (entry->cFileName [0] == '.')        ; //  Skip hidden files    else    //  If we have a subdirectory, go load that    if (entry->dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) {        zdir_t *subdir = zdir_new (entry->cFileName, self->path);        zlist_append (self->subdirs, subdir);    }    else {        //  Add file entry to directory list        zfile_t *file = zfile_new (self->path, entry->cFileName);        zlist_append (self->files, file);    }}

void wait_addqueue (waitqueue_t *q, wait_t *w){    assert (q->magic == WAITQUEUE_MAGIC);    assert (w->magic == WAIT_MAGIC);    if (zlist_append (q->q, w) < 0)        oom ();    w->usecount++;}

intzmsg_add (zmsg_t *self, zframe_t *frame){    assert (self);    assert (frame);    self->content_size += zframe_size (frame);    return zlist_append (self->frames, frame);}

static int defer_enqueue (zlist_t **l, flux_msg_t *msg){    if ((!*l && !(*l = zlist_new ())) || zlist_append (*l, msg) < 0) {        errno = ENOMEM;        return -1;    }    return 0;}

voidzmsg_addmem (zmsg_t *self, const void *src, size_t size){    assert (self);    zframe_t *frame = zframe_new (src, size);    self->content_size += size;    zlist_append (self->frames, frame);}

//  Process message from pipestatic voidserver_control_message (server_t *self){    zmsg_t *msg = zmsg_recv (self->pipe);    char *method = zmsg_popstr (msg);    if (streq (method, "BIND")) {        char *endpoint = zmsg_popstr (msg);        self->port = zsocket_bind (self->router, endpoint);        zstr_sendf (self->pipe, "%d", self->port);        free (endpoint);    }    else    if (streq (method, "PUBLISH")) {        char *location = zmsg_popstr (msg);        char *alias = zmsg_popstr (msg);        mount_t *mount = mount_new (location, alias);        zlist_append (self->mounts, mount);                  free (location);        free (alias);    }    else    if (streq (method, "SET ANONYMOUS")) {        char *enabled_string = zmsg_popstr (msg);        long enabled = atoi (enabled_string);        free (enabled_string);        //  Enable anonymous access without a config file                           zconfig_put (self->config, "security/anonymous", enabled? "1" :"0");    }    else    if (streq (method, "CONFIG")) {        char *config_file = zmsg_popstr (msg);        zconfig_destroy (&self->config);        self->config = zconfig_load (config_file);        if (self->config)            server_apply_config (self);        else {            printf ("E: cannot load config file '%s'/n", config_file);            self->config = zconfig_new ("root", NULL);        }        free (config_file);    }    else    if (streq (method, "SETOPTION")) {        char *path = zmsg_popstr (msg);        char *value = zmsg_popstr (msg);        zconfig_put (self->config, path, value);        server_config_self (self);        free (path);        free (value);    }    else    if (streq (method, "STOP")) {        zstr_send (self->pipe, "OK");        self->stopped = true;    }    free (method);    zmsg_destroy (&msg);}

struct subprocess * subprocess_create (struct subprocess_manager *sm){    int fds[2];    int saved_errno;    struct subprocess *p = xzmalloc (sizeof (*p));    memset (p, 0, sizeof (*p));    p->childfd = -1;    p->parentfd = -1;    fda_zero (p->child_fda);    p->sm = sm;    if (!(p->zhash = zhash_new ())     || hooks_table_init (p) < 0) {        errno = ENOMEM;        goto error;    }    p->pid = (pid_t) -1;    p->refcount = 1;    if (socketpair (PF_LOCAL, SOCK_STREAM | SOCK_CLOEXEC, 0, fds) < 0)        goto error;    p->childfd = fds[0];    p->parentfd = fds[1];    p->started = 0;    p->running = 0;    p->exited = 0;    p->completed = 0;    if (!(p->zio_in = zio_pipe_writer_create ("stdin", (void *) p)))        goto error;    if (!(p->zio_out = zio_pipe_reader_create ("stdout", NULL, (void *) p)))        goto error;    if (!(p->zio_err = zio_pipe_reader_create ("stderr", NULL, (void *) p)))        goto error;    zio_set_send_cb (p->zio_out, output_handler);    zio_set_send_cb (p->zio_err, output_handler);    if (zlist_append (sm->processes, (void *)p) < 0) {        errno = ENOMEM;        goto error;    }    if (sm->reactor) {        zio_reactor_attach (p->zio_in, sm->reactor);        zio_reactor_attach (p->zio_err, sm->reactor);        zio_reactor_attach (p->zio_out, sm->reactor);    }    return (p);error:    saved_errno = errno;    subprocess_destroy (p);    errno = saved_errno;    return (NULL);}

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}

static int backlog_append (flux_msg_handler_t *w, flux_msg_t **msg){    if (!w->backlog && !(w->backlog = zlist_new ()))        oom ();    if (zlist_append (w->backlog, *msg) < 0)        oom ();    *msg = NULL;    return 0;}

static voidclient_store_chunk (client_t *self, zchunk_t **chunk_p){    zchunk_t *chunk = *chunk_p;    assert (chunk);    zlist_append (self->queue, chunk);    self->pending += zchunk_size (chunk);    *chunk_p = NULL;}

// Distribute all backlog messages while are workers available.// Failed-to-send messages are put back in the backlog.voidbroker_check_backlog (broker_t *self){    while (zlist_size (self->backlog) && zlist_size (self->executor_lb)) {        zmsg_t *backlog_msg = zlist_pop (self->backlog);        if (0 != broker_send_to_executor (self, backlog_msg))            zlist_append (self->backlog, backlog_msg);    }}

static int client_send_nocopy (client_t *c, flux_msg_t **msg){    if (zlist_append (c->outqueue, *msg) < 0) {        errno = ENOMEM;        return -1;    }    *msg = NULL;    return client_send_try (c);}

static zsync_node_t *zsync_node_new (){    int rc;    zsync_node_t *self = (zsync_node_t *) zmalloc (sizeof (zsync_node_t));       self->ctx = zctx_new ();    assert (self->ctx);    self->zyre = zyre_new (self->ctx);      assert (self->zyre);        // Obtain permanent UUID    self->own_uuid = zuuid_new ();    if (zsys_file_exists (UUID_FILE)) {        // Read uuid from file        zfile_t *uuid_file = zfile_new (".", UUID_FILE);        int rc = zfile_input (uuid_file);    // open file for reading                assert (rc == 0);         zchunk_t *uuid_chunk = zfile_read (uuid_file, 16, 0);        assert (zchunk_size (uuid_chunk) == 16);    // make sure read succeeded        zuuid_set (self->own_uuid, zchunk_data (uuid_chunk));        zfile_destroy (&uuid_file);    } else {        // Write uuid to file        zfile_t *uuid_file = zfile_new (".", UUID_FILE);        rc = zfile_output (uuid_file); // open file for writing        assert (rc == 0);        zchunk_t *uuid_bin = zchunk_new ( zuuid_data (self->own_uuid), 16);        rc = zfile_write (uuid_file, uuid_bin, 0);        assert (rc == 0);        zfile_destroy (&uuid_file);    }        // Obtain peers and states    self->peers = zlist_new ();    if (zsys_file_exists (PEER_STATES_FILE)) {        zhash_t *peer_states = zhash_new ();        int rc = zhash_load (peer_states, PEER_STATES_FILE);        assert (rc == 0);        zlist_t *uuids = zhash_keys (peer_states);        char *uuid = zlist_first (uuids);        while (uuid) {            char * state_str = zhash_lookup (peer_states, uuid);            uint64_t state;            sscanf (state_str, "%"SCNd64, &state);            zlist_append (self->peers, zsync_peer_new (uuid, state));             uuid = zlist_next (uuids);        }    }        self->zyre_peers = zhash_new ();    self->terminated = false;    return self;}

void *thread (void *arg){    thd_t *t = arg;    char *key, *fence_name = NULL;    int i, flags = 0;    struct timespec t0;    uint32_t rank;    flux_future_t *f;    flux_kvs_txn_t *txn;    if (!(t->h = flux_open (NULL, 0))) {        log_err ("%d: flux_open", t->n);        goto done;    }    if (flux_get_rank (t->h, &rank) < 0) {        log_err ("%d: flux_get_rank", t->n);        goto done;    }    for (i = 0; i < count; i++) {        if (!(txn = flux_kvs_txn_create ()))            log_err_exit ("flux_kvs_txn_create");        key = xasprintf ("%s.%"PRIu32".%d.%d", prefix, rank, t->n, i);        if (fopt)            fence_name = xasprintf ("%s-%d", prefix, i);        if (sopt)            monotime (&t0);        if (flux_kvs_txn_pack (txn, 0, key, "i", 42) < 0)            log_err_exit ("%s", key);        if (nopt && (i % nopt_divisor) == 0)            flags |= FLUX_KVS_NO_MERGE;        else            flags = 0;        if (fopt) {            if (!(f = flux_kvs_fence (t->h, flags, fence_name,                                                   fence_nprocs, txn))                    || flux_future_get (f, NULL) < 0)                log_err_exit ("flux_kvs_fence");            flux_future_destroy (f);        } else {            if (!(f = flux_kvs_commit (t->h, flags, txn))                    || flux_future_get (f, NULL) < 0)                log_err_exit ("flux_kvs_commit");            flux_future_destroy (f);        }        if (sopt && zlist_append (t->perf, ddup (monotime_since (t0))) < 0)            oom ();        free (key);        free (fence_name);        flux_kvs_txn_destroy (txn);    }done:    if (t->h)        flux_close (t->h);    return NULL;}

/*PUB tcp://*:9003<<  [ "(?stock:[A-Z][A-Z0-9]+)" ]    [ timestamp :f64            ]    [ price     :f64            ]*/int main(int argc, const char* argv[]){  SetConsoleTitle(L"tickz.server");  // initialize random number generator  srand( (unsigned int)time(NULL) );  // initialize stock data  tick_t msft = tick_new("MSFT", 41.78);  tick_t aapl = tick_new("AAPL", 95.35);  tick_t goog = tick_new("GOOG",571.09);  tick_t yhoo = tick_new("YHOO", 34.53);  tick_t bbry = tick_new("BBRY", 10.90);    zlist_t *stocks = zlist_new();  zlist_append(stocks,&msft);  zlist_append(stocks,&aapl);  zlist_append(stocks,&goog);  zlist_append(stocks,&yhoo);  zlist_append(stocks,&bbry);    // set up publisher  zctx_t *ctx = zctx_new();  void *pub = zsocket_new(ctx,ZMQ_PUB);  zsocket_bind(pub,"tcp://*:9003");  // set up main loop  zloop_t *loop = zloop_new();  zloop_data_t loopdata;  loopdata.stocks = stocks;  loopdata.socket = pub;  // every 500 ms, update the stocks and publish the new data  int timer = zloop_timer(loop,500,0,onloop,&loopdata); //TOOD: take delay as input  zloop_start(loop); //NOTE: CTRL+C will cleanly interrupt the infinite loop    // clean up  zctx_destroy(&ctx);  zlist_destroy(&stocks);  return 0;}

/* This test is to make sure that deferred responses are handled in order. * Arrange for module to source 10K sequenced responses.  Messages 5000-5499 * are "put back" on the handle using flux_putmsg().  We ensure that * the 10K messages are nonetheless received in order. */void test_putmsg (flux_t *h, uint32_t nodeid){    flux_future_t *f;    const char *json_str;    const int count = 10000;    const int defer_start = 5000;    const int defer_count = 500;    json_object *in = Jnew ();    json_object *out = NULL;    int seq, myseq = 0;    zlist_t *defer = zlist_new ();    bool popped = false;    flux_msg_t *z;    if (!defer)        oom ();    Jadd_int (in, "count", count);    if (!(f = flux_rpc (h, "req.nsrc", Jtostr (in), FLUX_NODEID_ANY,                                                      FLUX_RPC_NORESPONSE)))        log_err_exit ("%s", __FUNCTION__);    flux_future_destroy (f);    do {        flux_msg_t *msg = flux_recv (h, FLUX_MATCH_ANY, 0);        if (!msg)            log_err_exit ("%s", __FUNCTION__);        if (flux_response_decode (msg, NULL, &json_str) < 0)            log_msg_exit ("%s: decode", __FUNCTION__);        if (!json_str            || !(out = Jfromstr (json_str))            || !Jget_int (out, "seq", &seq))            log_msg_exit ("%s: decode - payload", __FUNCTION__);        Jput (out);        if (seq >= defer_start && seq < defer_start + defer_count && !popped) {            if (zlist_append (defer, msg) < 0)                oom ();            if (seq == defer_start + defer_count - 1) {                while ((z = zlist_pop (defer))) {                    if (flux_requeue (h, z, FLUX_RQ_TAIL) < 0)                        log_err_exit ("%s: flux_requeue", __FUNCTION__);                    flux_msg_destroy (z);                }                popped = true;            }            continue;        }        if (seq != myseq)            log_msg_exit ("%s: expected %d got %d", __FUNCTION__, myseq, seq);        myseq++;        flux_msg_destroy (msg);    } while (myseq < count);    zlist_destroy (&defer);    Jput (in);}

intzmsg_addstr (zmsg_t *self, const char *string){    assert (self);    assert (zmsg_is (self));    assert (string);    self->content_size += strlen (string);    zlist_append (self->frames, zframe_new (string, strlen (string)));    return 0;}

intzmsg_addmem (zmsg_t *self, const void *src, size_t size){    assert (self);    zframe_t *frame = zframe_new (src, size);    if (frame) {        self->content_size += size;        return zlist_append (self->frames, frame);    }    else        return -1;}

void list_timer_cb (struct ev_loop *loop, ev_timer *w, int revents){    static int i = 100;    zlist_t *l = w->data;    if (--i == 0) {        ev_break (loop, EVBREAK_ALL);    } else {        zmsg_t *zmsg;        if (!(zmsg = zmsg_new ()) || zlist_append (l, zmsg) < 0)            oom ();    }}

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

intzmsg_append (zmsg_t *self, zframe_t **frame_p){    assert (self);    assert (zmsg_is (self));    assert (frame_p);    zframe_t *frame = *frame_p;    *frame_p = NULL;            //  We now own frame    self->content_size += zframe_size (frame);    return zlist_append (self->frames, frame);}