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

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

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 (void){    s_version_assert (2, 1);    //  Prepare our context and sockets    void *context = zmq_init (1);    void *frontend = zmq_socket (context, ZMQ_XREP);    void *backend  = zmq_socket (context, ZMQ_XREP);    zmq_bind (frontend, "tcp://*:5555");    //  For clients    zmq_bind (backend,  "tcp://*:5556");    //  For workers    //  Queue of available workers    int available_workers = 0;    char *worker_queue [MAX_WORKERS];    while (1) {        zmq_pollitem_t items [] = {            { backend,  0, ZMQ_POLLIN, 0 },            { frontend, 0, ZMQ_POLLIN, 0 }        };        //  Poll frontend only if we have available workers        if (available_workers)            zmq_poll (items, 2, -1);        else            zmq_poll (items, 1, -1);        //  Handle worker activity on backend        if (items [0].revents & ZMQ_POLLIN) {            zmsg_t *zmsg = zmsg_recv (backend);            //  Use worker address for LRU routing            assert (available_workers < MAX_WORKERS);            worker_queue [available_workers++] = zmsg_unwrap (zmsg);            //  Return reply to client if it's not a READY            if (strcmp (zmsg_address (zmsg), "READY") == 0)                zmsg_destroy (&zmsg);            else                zmsg_send (&zmsg, frontend);        }        if (items [1].revents & ZMQ_POLLIN) {            //  Now get next client request, route to next worker            zmsg_t *zmsg = zmsg_recv (frontend);            //  REQ socket in worker needs an envelope delimiter            zmsg_wrap (zmsg, worker_queue [0], "");            zmsg_send (&zmsg, backend);            //  Dequeue and drop the next worker address            free (worker_queue [0]);            DEQUEUE (worker_queue);            available_workers--;        }    }    //  We never exit the main loop    return 0;}

static int event(zloop_t *loop, zmq_pollitem_t *item, void *arg){	if (interrupt)		return -1;	zmsg_t *msg = zmsg_recv(dealer);	zframe_t *payload = zmsg_pop(msg);	zmsg_destroy(&msg);	msgpack_unpacked object;	msgpack_unpacked_init(&object);		if (msgpack_unpack_next(&object, (char*)zframe_data(payload), zframe_size(payload) , NULL))	{		//zclock_log("message");		//msgpack_object_print(stdout, object.data);		char *command = (char*)m_lookup(object.data, "command");		if (command) {			//zclock_log("command: %s", command);			if (streq(command, "exception")) {			    failed++;			}			if (streq(command, "result")) {				success++;			}			free(command);		}	}	msgpack_unpacked_destroy(&object);	zframe_destroy(&payload);	return 0;}

zmsg_t *curve_server_recv (curve_server_t *self){    assert (self);    zmsg_t *msg = zmsg_recv (self->data);    return msg;}

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

staticint actor_command(zloop_t *loop, zsock_t *socket, void *callback_data){    static size_t ticks = 0;    int rc = 0;    subscriber_state_t *state = callback_data;    zmsg_t *msg = zmsg_recv(socket);    if (msg) {        char *cmd = zmsg_popstr(msg);        if (streq(cmd, "$TERM")) {            fprintf(stderr, "[D] subscriber: received $TERM command/n");            rc = -1;        } else if (streq(cmd, "tick")) {            printf("[I] subscriber: %5zu messages "                   "(gap_size: %zu, no_info: %zu, dev_zero: %zu, blocks: %zu, drops: %zu)/n",                   state->message_count, state->message_gap_size, state->meta_info_failures,                   state->messages_dev_zero, state->message_blocks, state->message_drops);            state->message_count = 0;            state->message_gap_size = 0;            state->meta_info_failures = 0;            state->messages_dev_zero = 0;            state->message_drops = 0;            if (++ticks % HEART_BEAT_INTERVAL == 0)                device_tracker_reconnect_stale_devices(state->tracker);        } else {            fprintf(stderr, "[E] subscriber: received unknown actor command: %s/n", cmd);        }        free(cmd);        zmsg_destroy(&msg);    }    return rc;}

zmsg_t *curve_client_recv (curve_client_t *self){    assert (self);    zmsg_t *msg = zmsg_recv (self->data);    return msg;}

static voidsomeactor_recv_api (someactor_t *self){//  Get the whole message of the pipe in one go    zmsg_t *request = zmsg_recv (self->pipe);    if (!request)       return;        //  Interrupted    char *command = zmsg_popstr (request);    if (streq (command, "START"))        zsock_signal (self->pipe, someactor_start (self));    else    if (streq (command, "STOP"))        zsock_signal (self->pipe, someactor_stop (self));    else    if (streq (command, "VERBOSE")) {        self->verbose = true;        zsock_signal (self->pipe, 0);    }    else    if (streq (command, "$TERM"))        //  The $TERM command is send by zactor_destroy() method        self->terminated = true;    else {        zsys_error ("invalid command '%s'", command);        assert (false);    }}

int main(int argc, char *argv[]){    if (argc < 2) {        printf("syntax: %s <endpoint>/n", argv[0]);        exit(EXIT_SUCCESS);    }    zctx_t *ctx = zctx_new();    void *server = zsocket_new(ctx, ZMQ_REP);    zsocket_bind(server, argv[1]);    printf("Server is ready at %s/n", argv[1]);    while (TRUE) {        zmsg_t *msg = zmsg_recv(server);        if (!msg) break;        zmsg_send(&msg, server);    }    if (zctx_interrupted) {        printf("context interrupted/n");    }    zctx_destroy(&ctx);    return 0;}

intzsock_wait (void *self){    assert (self);    //  A signal is a message containing one frame with our 8-byte magic     //  value. If we get anything else, we discard it and continue to look    //  for the signal message    while (true) {        zmsg_t *msg = zmsg_recv (self);        if (!msg)            return -1;        if (zmsg_size (msg) == 1        &&  zmsg_content_size (msg) == 8) {            zframe_t *frame = zmsg_first (msg);            int64_t signal_value = *((int64_t *) zframe_data (frame));            if ((signal_value & 0xFFFFFFFFFFFFFF00L) == 0x7766554433221100L) {                zmsg_destroy (&msg);                return signal_value & 255;            }        }        zmsg_destroy (&msg);    }    return -1;}

static voidecho_actor (zsock_t *pipe, void *args){    //  Do some initialization    assert (streq ((char *) args, "Hello, World"));    zsock_signal (pipe, 0);    bool terminated = false;    while (!terminated) {        zmsg_t *msg = zmsg_recv (pipe);        if (!msg)            break;              //  Interrupted        char *command = zmsg_popstr (msg);        //  All actors must handle $TERM in this way        if (streq (command, "$TERM"))            terminated = true;        else        //  This is an example command for our test actor        if (streq (command, "ECHO"))            zmsg_send (&msg, pipe);        else {            puts ("E: invalid message to actor");            assert (false);        }        free (command);        zmsg_destroy (&msg);    }}

/** *  * @param foundId * @param foundReply * @return  */bool BoomStick::ReadFromReadySocket(std::string& foundId, std::string& foundReply) {   if (0 == mUtilizedThread) {      mUtilizedThread = pthread_self();   } else {      CHECK(pthread_self() == mUtilizedThread);   }   if (!mChamber) {      LOG(WARNING) << "Invalid socket";      return false;   }   bool success = false;   zmsg_t* msg = zmsg_recv(mChamber);   if (!msg) {      foundReply = zmq_strerror(zmq_errno());   } else if (zmsg_size(msg) == 2) {      char* msgChar;      msgChar = zmsg_popstr(msg);      foundId = msgChar;      free(msgChar);      msgChar = zmsg_popstr(msg);      foundReply = msgChar;      free(msgChar);      success = true;   } else {      foundReply = "Malformed reply, expecting 2 parts";   }   if (msg) {      zmsg_destroy(&msg);   }   return success;}

static void zmqreader (flux_reactor_t *r, flux_watcher_t *w,                       int revents, void *arg){    void *sock = flux_zmq_watcher_get_zsock (w);    static int count = 0;    if (revents & FLUX_POLLERR) {        fprintf (stderr, "%s: FLUX_POLLERR is set/n", __FUNCTION__);        goto error;    }    if (revents & FLUX_POLLIN) {        zmsg_t *zmsg = zmsg_recv (sock);        if (!zmsg) {            fprintf (stderr, "%s: zmsg_recv: %s/n",                     __FUNCTION__, strerror (errno));            goto error;        }        zmsg_destroy (&zmsg);        count++;        if (count == zmqwriter_msgcount)            flux_watcher_stop (w);    }    return;error:    flux_reactor_stop_error (r);}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    broker_t *self = s_broker_new (verbose);    s_broker_bind (self, "tcp://*:5555");    //  Get and process messages forever or until interrupted    while (TRUE) {        zmq_pollitem_t items [] = {            { self->socket,  0, ZMQ_POLLIN, 0 } };        int rc = zmq_poll (items, 1, HEARTBEAT_INTERVAL * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Interrupted        //  Process next input message, if any        if (items [0].revents & ZMQ_POLLIN) {            zmsg_t *msg = zmsg_recv (self->socket);            if (!msg)                break;          //  Interrupted            if (self->verbose) {                zclock_log ("I: received message:");                zmsg_dump (msg);            }            zframe_t *sender = zmsg_pop (msg);            zframe_t *empty  = zmsg_pop (msg);            zframe_t *header = zmsg_pop (msg);            if (zframe_streq (header, MDPC_CLIENT))                s_client_process (self, sender, msg);            else            if (zframe_streq (header, MDPW_WORKER))                s_worker_process (self, sender, msg);            else {                zclock_log ("E: invalid message:");                zmsg_dump (msg);                zmsg_destroy (&msg);            }            zframe_destroy (&sender);            zframe_destroy (&empty);            zframe_destroy (&header);        }        //  Disconnect and delete any expired workers        //  Send heartbeats to idle workers if needed        if (zclock_time () > self->heartbeat_at) {            s_broker_purge_workers (self);            worker_t *worker = (worker_t *) zlist_first (self->waiting);            while (worker) {                s_worker_send (self, worker, MDPW_HEARTBEAT, NULL, NULL);                worker = (worker_t *) zlist_next (self->waiting);            }            self->heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;        }    }    if (zctx_interrupted)        printf ("W: interrupt received, shutting down.../n");    s_broker_destroy (&self);    return 0;}

zre_msg_t *zre_msg_recv (void *input){    assert (input);    zmsg_t *msg = zmsg_recv (input);    return zre_msg_decode (&msg, zsocket_type (input));}

// Handle input from worker, on backendint s_handle_backend(zloop_t *loop, zmq_pollitem_t *poller, void *arg){	// Use worker identity for load-balancing	lbbroker_t *self = (lbbroker_t *)arg;	zmsg_t *msg = zmsg_recv(self->backend);	if (msg) {		zframe_t *identity = zmsg_unwrap(msg);		zlist_append(self->workers, identity);		// Enable reader on frontend if we went from 0 to 1 workers		if (zlist_size(self->workers) == 1) {			zmq_pollitem_t poller = { self->frontend, 0, ZMQ_POLLIN };			zloop_poller(loop, &poller, s_handle_frontend, self);		}		// Forward message to client if it's not a READY		zframe_t *frame = zmsg_first(msg);		if (memcmp(zframe_data(frame), WORKER_READY, strlen(WORKER_READY)) == 0) {			zmsg_destroy(&msg);		} else {			zmsg_send(&msg, self->frontend);		}	}	return 0;}

void agent_control_message (agent_t *self){    zmsg_t *msg = zmsg_recv (self->control);    char *command = zmsg_pop (msg);    if (strcmp (command, "CONNECT") == 0) {        char *endpoint = zmsg_pop (msg);        printf ("I: connecting to %s.../n", endpoint);        int rc = zmq_connect (self->router, endpoint);        assert (rc == 0);        server_t *server = server_new (endpoint);        zhash_insert (self->servers, endpoint, server);        zhash_freefn (self->servers, endpoint, s_server_free);        zlist_append (self->actives, server);        server->ping_at = s_clock () + PING_INTERVAL;        server->expires = s_clock () + SERVER_TTL;        free (endpoint);    }    else    if (strcmp (command, "REQUEST") == 0) {        assert (!self->request);    //  Strict request-reply cycle        //  Prefix request with sequence number and empty envelope        char sequence_text [10];        sprintf (sequence_text, "%u", ++self->sequence);        zmsg_push (msg, sequence_text);        //  Take ownership of request message        self->request = msg;        msg = NULL;        //  Request expires after global timeout        self->expires = s_clock () + GLOBAL_TIMEOUT;    }    free (command);    zmsg_destroy (&msg);}

staticint actor_command(zloop_t *loop, zsock_t *socket, void *arg){    int rc = 0;    watchdog_state_t *state = arg;    zmsg_t *msg = zmsg_recv(socket);    if (msg) {        char *cmd = zmsg_popstr(msg);        if (streq(cmd, "$TERM")) {            state->received_term_cmd = true;            // fprintf(stderr, "[D] watchdog[0]: received $TERM command/n");            rc = -1;        }        else if (streq(cmd, "tick")) {            if (verbose)                printf("[I] watchdog: credit: %d/n", state->credit);            state->credit = CREDIT;        } else {            fprintf(stderr, "[E] watchdog[0]: received unknown actor command: %s/n", cmd);        }        free(cmd);        zmsg_destroy(&msg);    }    return rc;}

void agent_router_message (agent_t *self){    zmsg_t *reply = zmsg_recv (self->router);    //  Frame 0 is server that replied    char *endpoint = zmsg_pop (reply);    server_t *server = (server_t *) zhash_lookup (self->servers, endpoint);    assert (server);    free (endpoint);    if (!server->alive) {        zlist_append (self->actives, server);        server->alive = 1;    }    server->ping_at = s_clock () + PING_INTERVAL;    server->expires = s_clock () + SERVER_TTL;    //  Frame 1 may be sequence number for reply    if (zmsg_parts (reply) > 1    &&  atoi (zmsg_address (reply)) == self->sequence) {        free (zmsg_pop (reply));        zmsg_push (reply, "OK");        zmsg_send (&reply, self->control);        zmsg_destroy (&self->request);    }    zmsg_destroy (&reply);}

static void* worker_routine(void* arg){  zmsg_t* msg;  zframe_t* frame;  zctx_t* ctx = zctx_new();  void* worker = zsocket_new(ctx, ZMQ_REQ);  zsocket_connect(worker, "ipc://%s-localbe.ipc", self);  frame = zframe_new(WORKER_READY, 1);  zframe_send(&frame, worker, 0);  while (1) {    msg = zmsg_recv(worker);    if (!msg)      break;    zframe_print(zmsg_last(msg), "Worker: ");    zframe_reset(zmsg_last(msg), "OK", 2);    zmsg_send(&msg, worker);  }  zctx_destroy(&ctx);  return NULL;}

zmsg_t *zre_interface_recv (zre_interface_t *self){    assert (self);    zmsg_t *msg = zmsg_recv (self->pipe);    return msg;}

intzstr_recvx (void *source, char **string_p, ...){    assert (source);    void *handle = zsock_resolve (source);    zmsg_t *msg = zmsg_recv (handle);    if (!msg)        return -1;    //  Filter a signal that may come from a dying actor    if (zmsg_signal (msg) >= 0) {        zmsg_destroy (&msg);        return -1;    }    int count = 0;    va_list args;    va_start (args, string_p);    while (string_p) {        *string_p = zmsg_popstr (msg);        string_p = va_arg (args, char **);        count++;    }    va_end (args);    zmsg_destroy (&msg);    return count;}

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

/* ================ upload_data() ================ */int upload_data(zsock_t *sock, const char *key, const char *data, uint32_t data_size){    /* ---------------- Send Message ---------------- */    zmsg_t *upload_msg = create_action_message(MSG_ACTION_PUT);    message_add_key_data(upload_msg, key, data, data_size);    zmsg_send(&upload_msg, sock);    /* ---------------- Receive Message ---------------- */    zmsg_t *recv_msg = zmsg_recv(sock);    if ( recv_msg == NULL ){        return -2;    }    /*zmsg_print(recv_msg);*/    int rc = 0;    if (message_check_status(recv_msg, MSG_STATUS_WORKER_ACK) == 0 ){        /*info_log("Return MSG_STATUS_WORKER_ACK. key=%s", key);*/        rc = 0;    } else if ( message_check_status(recv_msg, MSG_STATUS_WORKER_ERROR) == 0 ){        error_log("Return MSG_STATUS_WORKER_ERROR. key=%s", key);        rc = -1;    }    zmsg_destroy(&recv_msg);    return rc;}

static ints_agent_handle_data (agent_t *self){    //  First frame is client address (hashkey)    //  If caller sends unknown client address, we discard the message    //  For testing, we'll abort in this case, since it cannot happen    //  The assert disappears when we start to timeout clients...    zmsg_t *request = zmsg_recv (self->data);    char *hashkey = zmsg_popstr (request);    client_t *client = (client_t *) zhash_lookup (self->clients, hashkey);    free (hashkey);    if (client) {        //  Encrypt and send all frames of request        //  Each frame is a full ZMQ message with identity frame        while (zmsg_size (request)) {            zframe_t *cleartext = zmsg_pop (request);            if (zmsg_size (request))                zframe_set_more (cleartext, 1);            zframe_t *encrypted = curve_codec_encode (client->codec, &cleartext);            if (encrypted) {                zframe_send (&client->address, self->router, ZFRAME_MORE + ZFRAME_REUSE);                zframe_send (&encrypted, self->router, 0);            }            else                client_set_exception (client);        }    }    zmsg_destroy (&request);    return 0;}

/* ================ handle_pullin_on_client_pipe() ================ */int handle_pullin_on_client_pipe(zloop_t *loop, zsock_t *pipe, void *user_data){    client_t *client = (client_t*)user_data;    if ( over_actors >= total_actors ){        zloop_reader_end(loop, pipe);        return -1;    }    zmsg_t *msg = zmsg_recv(pipe);    if ( msg == NULL ){        zloop_reader_end(loop, pipe);        return -1;    }    /*zmsg_print(msg);*/    if ( message_check_status(msg, MSG_STATUS_ACTOR_OVER) == 0 ){        over_actors++;        info_log("Actor %d over! (%d/%d)", client->id, over_actors, total_actors);    }    zmsg_destroy(&msg);    return 0;}

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

/* ================ delete_data() ================ */int delete_data(zsock_t *sock, const char *key){    /* ---------------- Send Message ---------------- */    zmsg_t *delete_msg = create_action_message(MSG_ACTION_DEL);    message_add_key_data(delete_msg, key, "", 0);    zmsg_send(&delete_msg, sock);    /* ---------------- Receive Message ---------------- */    zmsg_t *recv_msg = zmsg_recv(sock);    if ( recv_msg == NULL ){        return -2;    }    zmsg_print(recv_msg);    int rc = -1;    if (message_check_status(recv_msg, MSG_STATUS_WORKER_NOTFOUND) == 0 ){        warning_log("Not Found. key=%s", key);        rc = 0;    } else if ( message_check_status(recv_msg, MSG_STATUS_WORKER_ERROR) == 0 ){        error_log("Return MSG_STATUS_WORKER_ERROR. key=%s", key);        rc = -1;    }    zmsg_destroy(&recv_msg);    return rc;}