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

static voids_self_switch (self_t *self, zsock_t *input, zsock_t *output){    //  We use the low-level libzmq API for best performance    void *zmq_input = zsock_resolve (input);    void *zmq_output = zsock_resolve (output);    void *zmq_capture = self->capture? zsock_resolve (self->capture): NULL;    zmq_msg_t msg;    zmq_msg_init (&msg);    if (zmq_recvmsg (zmq_input, &msg, 0) == -1)        return;                 //  Nothing to do, probably interrupted            while (true) {        int send_flags = zsocket_rcvmore (zmq_input)? ZMQ_SNDMORE: 0;        if (zmq_capture) {            zmq_msg_t dup;            zmq_msg_init (&dup);            zmq_msg_copy (&dup, &msg);            if (zmq_sendmsg (zmq_capture, &dup, send_flags) == -1)                zmq_msg_close (&dup);        }        if (zmq_sendmsg (zmq_output, &msg, send_flags) == -1) {            zmq_msg_close (&msg);            break;        }        if (zmq_recvmsg (zmq_input, &msg, ZMQ_DONTWAIT) == -1)            break;      //  Presumably EAGAIN    }}

int main (int argc, char *argv []){    //  Create the infrastructure    void *ctx = zmq_init (0);    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_XREP);    assert (sb);    int rc = zmq_bind (sb, "inproc://a");    assert (rc == 0);    void *sc = zmq_socket (ctx, ZMQ_XREQ);    assert (sc);    rc = zmq_connect (sc, "inproc://a");    assert (rc == 0);       //  Send 2-part message.    rc = zmq_send (sc, "A", 1, ZMQ_SNDMORE);    assert (rc == 1);    rc = zmq_send (sc, "B", 1, 0);    assert (rc == 1);    //  Identity comes first.    zmq_msg_t msg;    rc = zmq_msg_init (&msg);    assert (rc == 0);    rc = zmq_recvmsg (sb, &msg, 0);    assert (rc >= 0);    int more;    size_t more_size = sizeof (more);    rc = zmq_getmsgopt (&msg, ZMQ_MORE, &more, &more_size);    assert (rc == 0);    assert (more == 1);    //  Then the first part of the message body.    rc = zmq_recvmsg (sb, &msg, 0);    assert (rc == 1);    more_size = sizeof (more);    rc = zmq_getmsgopt (&msg, ZMQ_MORE, &more, &more_size);    assert (rc == 0);    assert (more == 1);    //  And finally, the second part of the message body.    rc = zmq_recvmsg (sb, &msg, 0);    assert (rc == 1);    more_size = sizeof (more);    rc = zmq_getmsgopt (&msg, ZMQ_MORE, &more, &more_size);    assert (rc == 0);    assert (more == 0);    //  Deallocate the infrastructure.    rc = zmq_close (sc);    assert (rc == 0);    rc = zmq_close (sb);    assert (rc == 0);    rc = zmq_term (ctx);    assert (rc == 0);    return 0 ;}

int router_start(struct router_context *ctx) {    zmq_msg_t msg;    struct router_route *route_info;    char pipe_path[256];    int ret, forwarded = 0;    GList *l;    lock_and_log("process_list_lock", &ctx->process_list_lock);    // Forward messages from peer router    zmq_msg_init(&msg);    ret = zmq_recvmsg(ctx->zmq_sock, &msg, ZMQ_NOBLOCK);    if (ret > 0) {        route_info = zmq_msg_data(&msg);        syslog(LOG_DEBUG, "router_route: message base: %p", route_info);        syslog(LOG_DEBUG, "router_route: received message from forwarder to pid %d", route_info->pid);        for (l = ctx->process_list; l != NULL; l = l->next) {            syslog(LOG_DEBUG, "router_route: trying to %p", l->data);            syslog(LOG_DEBUG, "router_route: trying to %d", ((struct router_process*)l->data)->pid);            if (((struct router_process*)l->data)->pid == route_info->pid) {                syslog(LOG_DEBUG, "router_forward_msg: /tforwarding message to pid %u", route_info->pid);                router_forward_to_proc(((struct router_process*)l->data)->zmq_sock,                                       zmq_msg_data(&msg) + sizeof(struct router_route),                                       zmq_msg_size(&msg) - sizeof(struct router_route));                forwarded += 1;                break;            }        }        if (forwarded == 0)            syslog(LOG_WARNING, "router_route: pid %d not found", route_info->pid);    }    // Forward messages from running processes    for (l = ctx->process_list; l != NULL; l = l->next) {        zmq_msg_t msg;        zmq_msg_init(&msg);        ret = zmq_recvmsg(((struct router_process*)l->data)->zmq_sock, &msg, ZMQ_NOBLOCK);        if (ret > 0) {            syslog(LOG_DEBUG, "router_route: received message from pid %u", ((struct router_process*)l->data)->pid);            router_forward_from_proc(ctx->zmq_sock, &msg, ((struct router_process*)l->data)->pid);            forwarded += 1;        }    }    unlock_and_log("process_list_lock", &ctx->process_list_lock);    return forwarded;}

		static int ForwardMessage(void* srcSocket, void* desSocket, zmq_msg_t& msg)		{			int rc;			int more;			size_t moresz;			while (true) 			{				rc = zmq_recvmsg(srcSocket, &msg, 0);				if (rc < 0)	return -1;				moresz = sizeof more;				rc = zmq_getsockopt(srcSocket, ZMQ_RCVMORE, &more, &moresz);				if (rc < 0)					return -1;				rc = zmq_sendmsg(desSocket, &msg, more? ZMQ_SNDMORE: 0);				if (rc < 0)					return -1;				if (more == 0)					break;			}			return 0;		}

static void copy_msg(void* from, void* to){    zmq_msg_t msg;    int more = 1;    int rc;    while (more)    {        more = 0;                rc = zmq_msg_init(&msg);        assert (rc == 0);        rc = zmq_recvmsg(from, &msg, 0);        assert (rc >= 0);                                size_t size = sizeof more;        rc = zmq_getsockopt(from, ZMQ_RCVMORE, &more, &size);        assert (rc == 0);        int flags = (more ? ZMQ_SNDMORE : 0);        rc = zmq_sendmsg(to, &msg, flags);        assert (rc >= 0);        rc = zmq_msg_close(&msg);        assert (rc == 0);    }}

int zmq_recv (void *s_, void *buf_, size_t len_, int flags_){    zmq_msg_t msg;    int rc = zmq_msg_init (&msg);    errno_assert (rc == 0);    int nbytes = zmq_recvmsg (s_, &msg, flags_);    if (unlikely (nbytes < 0)) {        int err = errno;        rc = zmq_msg_close (&msg);        errno_assert (rc == 0);        errno = err;        return -1;    }    //  At the moment an oversized message is silently truncated.    //  TODO: Build in a notification mechanism to report the overflows.    size_t to_copy = size_t (nbytes) < len_ ? size_t (nbytes) : len_;    memcpy (buf_, zmq_msg_data (&msg), to_copy);    rc = zmq_msg_close (&msg);    errno_assert (rc == 0);    return nbytes;}

CAMLprim value recv_stub(value socket, value rcv_option) {    CAMLparam2 (socket, rcv_option);    CAMLlocal1 (message);    void *sock = Socket_val(socket)->wrapped;    zmq_msg_t request;    int result = zmq_msg_init (&request);    stub_raise_if (result == -1);    caml_release_runtime_system();    result = zmq_recvmsg(sock, &request, Int_val(rcv_option));    caml_acquire_runtime_system();    stub_raise_if (result == -1);    size_t size = zmq_msg_size (&request);    if (size == 0) {        message = EMPTY_STRING;    } else {        message = caml_alloc_string(size);        memcpy (String_val(message), zmq_msg_data (&request), size);    }    result = zmq_msg_close(&request);    stub_raise_if (result == -1);    CAMLreturn (message);}

int CZmqThread::block_rcv_socket(QString & s){    QString msg;    int rc;    int64_t more;    size_t more_size = sizeof (more);    s="";    do {        rc = zmq_recvmsg (m_socket, &m_msg, ZMQ_DONTWAIT);        if (rc < 0) {            if ( zmq_errno () ==  EAGAIN ){                return (1);            }            printf ("error in zmq_rcv : %s/n", zmq_strerror (errno));            return(-1);        }else{            s +=(QString::fromLocal8Bit((char *)zmq_msg_data (&m_msg), (int)zmq_msg_size (&m_msg) ) );            rc = zmq_getsockopt (m_socket, ZMQ_RCVMORE, &more, &more_size);            if (rc != 0) {                printf ("error in zmq_getsockopt : %s/n", zmq_strerror (errno));                return(-1);            }        }    } while ( more );    return (0);}

        inline void monitor (monitor_t* mon)        {            zmq_event_t event;            int rc;            assert(mon);            void *s = zmq_socket (ctxptr, ZMQ_PAIR);            assert (s);            rc = zmq_connect (s, monaddr.c_str());            assert (rc == 0);            while (true) {                    zmq_msg_t msg;                    zmq_msg_init (&msg);                    rc = zmq_recvmsg (s, &msg, 0);                    if (rc == -1 && zmq_errno() == ETERM) break;                    assert (rc != -1);                    memcpy (&event, zmq_msg_data (&msg), sizeof (event));                    switch (event.event) {                    case ZMQ_EVENT_CONNECTED:                            mon->on_event_connected(event.data.connected.addr);                            break;                    case ZMQ_EVENT_CONNECT_DELAYED:                            mon->on_event_connect_delayed(event.data.connect_delayed.addr);                            break;                    case ZMQ_EVENT_CONNECT_RETRIED:                            mon->on_event_connect_retried(event.data.connect_retried.addr);                            break;                    case ZMQ_EVENT_LISTENING:                            mon->on_event_listening(event.data.listening.addr);                            break;                    case ZMQ_EVENT_BIND_FAILED:                            mon->on_event_bind_failed(event.data.bind_failed.addr);                            break;                    case ZMQ_EVENT_ACCEPTED:                            mon->on_event_accepted(event.data.accepted.addr);                            break;                    case ZMQ_EVENT_ACCEPT_FAILED:                            mon->on_event_accept_failed(event.data.accept_failed.addr);                            break;                    case ZMQ_EVENT_CLOSED:                        mon->on_event_closed(event.data.closed.addr);                            break;                    case ZMQ_EVENT_CLOSE_FAILED:                            mon->on_event_close_failed(event.data.close_failed.addr);                            break;                    case ZMQ_EVENT_DISCONNECTED:                            mon->on_event_disconnected(event.data.disconnected.addr);                            break;                    default:                            mon->on_event_unknown(event.event);                            break;                    }                    zmq_msg_close (&msg);            }            zmq_close (s);        }

intzsocket_wait (void *zocket){    zmq_msg_t msg;    zmq_msg_init (&msg);    int rc = zmq_recvmsg (zocket, &msg, 0);    return rc == -1? -1: 0;}

int main(int argc, char **argv) {  (void)argc;  (void)argv;  auto *context = zmq_ctx_new();  rpc_socket = zmq_socket(context, ZMQ_PAIR);  int rc = zmq_bind(rpc_socket, "tcp://127.0.0.1:9999");  if (rc == 0) {    std::cout << "Listening on: 127.0.0.1:9999" << std::endl;  }  while (true) {    zmq_msg_t header_zmsg;    zmq_msg_init(&header_zmsg);    zmq_recvmsg(rpc_socket, &header_zmsg, 0);    messages::header header;    msgpack::unpacked header_unpacked;    msgpack::unpack(&header_unpacked,                    static_cast<char *>(zmq_msg_data(&header_zmsg)),                    zmq_msg_size(&header_zmsg));    header_unpacked.get().convert(header);    zmq_msg_t body_zmsg;    zmq_msg_init(&body_zmsg);    zmq_recvmsg(rpc_socket, &body_zmsg, 0);    msgpack::unpacked body_unpacked;    msgpack::unpack(&body_unpacked,                    static_cast<char *>(zmq_msg_data(&body_zmsg)),                    zmq_msg_size(&body_zmsg));    execute_rpc_request(static_cast<rpc_call_name>(header.get<0>()),                        body_unpacked.get());    zmq_msg_close(&header_zmsg);    zmq_msg_close(&body_zmsg);  }  zmq_close(rpc_socket);  zmq_ctx_destroy(context);  return 0;}

 inline bool recv (message_t *msg_, int flags_ = 0) {     int nbytes = zmq_recvmsg (ptr, &(msg_->msg), flags_);     if (nbytes >= 0)         return true;     if (zmq_errno () == EAGAIN)         return false;     throw error_t (); }

static void *worker (void *ctx_)#endif{    void *s;    int rc;    int i;    zmq_msg_t msg;    s = zmq_socket (ctx_, ZMQ_REP);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        exit (1);    }    rc = zmq_connect (s, "inproc://lat_test");    if (rc != 0) {        printf ("error in zmq_connect: %s/n", zmq_strerror (errno));        exit (1);    }    rc = zmq_msg_init (&msg);    if (rc != 0) {        printf ("error in zmq_msg_init: %s/n", zmq_strerror (errno));        exit (1);    }    for (i = 0; i != roundtrip_count; i++) {        rc = zmq_recvmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_recvmsg: %s/n", zmq_strerror (errno));            exit (1);        }        rc = zmq_sendmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_sendmsg: %s/n", zmq_strerror (errno));            exit (1);        }    }    rc = zmq_msg_close (&msg);    if (rc != 0) {        printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));        exit (1);    }    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        exit (1);    }#if defined ZMQ_HAVE_WINDOWS    return 0;#else    return NULL;#endif}

intzsocket_wait (void *zocket){    zmq_msg_t msg;    zmq_msg_init (&msg);    if (zmq_recvmsg (zocket, &msg, 0) == -1)        return -1;    else {        zmq_msg_close (&msg);        return 0;    }}

//  Receive 0MQ string from socket and convert into C stringstatic char *s_recv (void *socket) {  zmq_msg_t message;  zmq_msg_init (&message);  zmq_recvmsg (socket, &message, 0);  // return (NULL);  int size = zmq_msg_size (&message);  char *string = malloc (size + 1);  memcpy (string, zmq_msg_data (&message), size);  zmq_msg_close (&message);  string [size] = 0;  printf("s: %s/n",string);  return (string);}

Ret get_reply(void *z_sock){  zmq_msg_t msg;  zmq_msg_init(&msg);  zmq_recvmsg(z_sock, &msg, 0);  msgpack::unpacked reply_unpacked;  msgpack::unpack(reply_unpacked,    static_cast<char*>(zmq_msg_data(&msg)), zmq_msg_size(&msg));  zmq_msg_close(&msg);  Ret reply_data;  reply_unpacked.get().convert(reply_data);  return reply_data;}

char *zstr_recv_nowait (void *zocket){    assert (zocket);    zmq_msg_t message;    zmq_msg_init (&message);    if (zmq_recvmsg (zocket, &message, ZMQ_DONTWAIT) < 0)        return NULL;    size_t size = zmq_msg_size (&message);    char *string = (char *) malloc (size + 1);    memcpy (string, zmq_msg_data (&message), size);    zmq_msg_close (&message);    string [size] = 0;    return string;}

static int pcore_recv_identity (pb_core_t *core, pb_identity_t *identity){	zmq_msg_t msg;	int rc, more;	zmq_msg_init (&msg);	rc = zmq_recvmsg (core->sock, &msg, 0);	if (rc < 0)	{		pb_log (core, PBYTE_WARN,		        "Recieve identity failed: %s",		        zmq_strerror(zmq_errno()));		zmq_msg_close(&msg);		return -1;	}	more = zmq_msg_more (&msg);	if (more != 1)	{		pb_log (core, PBYTE_WARN,		        "No more data after identity");		zmq_msg_close(&msg);		return -2;	}	if (zmq_msg_size (&msg) >= PBYTE_IDENTITY_LEN)	{		pb_log (core, PBYTE_WARN,		        "Recieve message failed. "		        "Large identity");		zmq_msg_close(&msg);		return -3;	}	// pb_print_buff (&core->logger, PBYTE_INFO,	//                "incomming identity",	//                zmq_msg_data(msg),	//                zmq_msg_size(msg));	memcpy (identity->identity, zmq_msg_data (&msg), zmq_msg_size (&msg));	identity->identity_len = zmq_msg_size (&msg);	zmq_msg_close(&msg);	return 0;}

int init_server(void *ctx){    char addr[32];    /* send out init signal to main server  and get the table number */     snprintf(addr, 32, "tcp://%s:%d", GlobalArgs.server_address, GlobalArgs.port+1);    void *skt = socket_connect(ctx, ZMQ_REQ, addr);    if (!skt){	syslog(LOG_CRIT,"INITSERVER: unable to get req skt");	exit(1);    }    char *str = strdup(init_str);    send_msg_data(skt, str, strlen(str)+1, free_fn, NULL);    /* wait for response */    zmq_msg_t msg;    zmq_msg_init(&msg);    time_t curr_time;    time(&curr_time);    int err = 0;    /* wait until error is recieved or until time alloted is up */    do {	err = zmq_recvmsg(skt, &msg, ZMQ_NOBLOCK);	sleep(1);    } while (err && time(NULL) <= curr_time + 10);    if (err){	syslog(LOG_CRIT,"INITSERVER: no response from main server - timed out");	err = -1;    }    uint8_t tn= 0;        if (err == 0){	memcpy(&tn, zmq_msg_data(&msg), sizeof(uint8_t));    }    zmq_msg_close(&msg);    /* set static global variable */    table_number = tn;    zmq_close(skt);    syslog(LOG_DEBUG, "INITSERVER: recived table number %u, init complete", tn);    return err;}

char *zstr_recv (void *source){    assert (source);    void *handle = zsock_resolve (source);    zmq_msg_t message;    zmq_msg_init (&message);    if (zmq_recvmsg (handle, &message, 0) < 0)        return NULL;    size_t size = zmq_msg_size (&message);    char *string = (char *) malloc (size + 1);    memcpy (string, zmq_msg_data (&message), size);    zmq_msg_close (&message);    string [size] = 0;    return string;}

char *zstr_recv_nowait (void *dest){    assert (dest);    void *handle = zsock_resolve (dest);    zmq_msg_t message;    zmq_msg_init (&message);    if (zmq_recvmsg (handle, &message, ZMQ_DONTWAIT) < 0)        return NULL;    size_t size = zmq_msg_size (&message);    char *string = (char *) malloc (size + 1);    if (string) {        memcpy (string, zmq_msg_data (&message), size);        string [size] = 0;    }    zmq_msg_close (&message);    return string;}

bool ZMQSocket::recv(int64_t flags, String& msg) {  zmq_msg_t message;  if (zmq_msg_init(&message) != 0) {    throwExceptionClassZMQErr(s_ZMQSocketExceptionClass, "Failed to initialize message structure: {}", errno);  }  int rc = zmq_recvmsg(socket->z_socket, &message, flags);  int err = errno;  if (rc == -1) {    zmq_msg_close(&message);    if (err == EAGAIN) {      return false;    }    throwExceptionClassZMQErr(s_ZMQSocketExceptionClass, "Failed to recieve message: {}", err);  }  msg = String((char*)zmq_msg_data(&message), zmq_msg_size(&message), CopyString);  zmq_msg_close(&message);  return true;}

uint64_t recv_stat (void *sock, bool last){    uint64_t res;    zmq_msg_t stats_msg;    int rc = zmq_msg_init (&stats_msg);    assert (rc == 0);    rc = zmq_recvmsg (sock, &stats_msg, 0);    assert (rc == sizeof(uint64_t));    memcpy(&res, zmq_msg_data(&stats_msg), zmq_msg_size(&stats_msg));    rc = zmq_msg_close (&stats_msg);    assert (rc == 0);    int more;    size_t moresz = sizeof more;    rc = zmq_getsockopt (sock, ZMQ_RCVMORE, &more, &moresz);    assert (rc == 0);    assert ((last && !more) || (!last && more));    return res;}

/** * Issue a read on the socket. */static zmq_msg_t* do_read(JNIEnv *env, jobject obj, zmq_msg_t *message, int flags){    void *s = get_socket (env, obj, 1);    int rc = zmq_msg_init (message);    int err = zmq_errno();    if (rc != 0) {        raise_exception (env, err);        return NULL;    }#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)    rc = zmq_recvmsg (s, message, flags);#else    rc = zmq_recv (s, message, flags);#endif    err = zmq_errno();    if (rc < 0 && err == EAGAIN) {        rc = zmq_msg_close (message);        err = zmq_errno();        if (rc != 0) {            raise_exception (env, err);            return NULL;        }        return NULL;    }    if (rc < 0) {        raise_exception (env, err);        rc = zmq_msg_close (message);        err = zmq_errno();        if (rc != 0) {            raise_exception (env, err);            return NULL;        }        return NULL;    }        return message;}

// worker thread functionstatic void* worker_routine (void *context){	// socket to talk to dispatcher	void* receiver = zmq_socket(context, ZMQ_REP); 	if (!receiver) 	{		dzlog_error("create socket to talk to dispatcher failed, err: %s", zmq_strerror(errno));		return NULL;	}	if (zmq_connect(receiver, "inproc://workers") != 0)	{		dzlog_error("worker thread connect worker_socket failed, err: %s", zmq_strerror(errno));		return NULL;	}	dzlog_info("worker thread run ...");	while (1) 	{		zmq_msg_t request;		zmq_msg_init(&request);		zmq_recvmsg(receiver, &request, 0); 		printf("Recv Hello/n"); 		zmq_msg_close(&request);		sleep(1);				zmq_msg_t reply;		zmq_msg_init_size(&reply, 5);		memcpy(zmq_msg_data (&reply), "World", 5); 		zmq_sendmsg(receiver, &reply, 0); 		zmq_msg_close(&reply);	}	zmq_close (receiver);	return NULL;}

// This main thread simply starts several clients, and a server, and then// waits for the server to finish.//int main (void){    void *context = zmq_init (1);    // Backend socket talks to workers over inproc    void *pipein = zmq_socket (context, ZMQ_DEALER);    zsocket_bind (pipein, "inproc://pipe");    // Frontend socket talks to clients over TCP    void *pipeout = zmq_socket (context, ZMQ_ROUTER);    zsocket_connect (pipeout, "inproc://pipe");    zstr_send (pipein, "ABC");        int rc;    zmq_msg_t msg;    zmq_msg_init (&msg);    rc = zmq_recvmsg (pipeout, &msg, 0);    if (rc < 0) {        printf ("E: error, failed/n");        exit (1);    }    int rcvmore;    size_t type_size = sizeof (int);    rc = zmq_getsockopt (pipeout, ZMQ_RCVMORE, &rcvmore, &type_size);    if (rc)        puts (strerror (errno));    assert (rc == 0);    int rcvlabel;    zmq_getsockopt (pipeout, ZMQ_RCVLABEL, &rcvlabel, &type_size);    assert (rc == 0);    printf ("Got first message part: size=%zd more=%d label=%d/n",        zmq_msg_size (&msg), rcvmore, rcvlabel);    return 0;}

int kill_server(){    char addr[32];    void *ctx = main_ctx;    snprintf(addr, 32, "tcp://%s:%d", GlobalArgs.server_address, GlobalArgs.port + 1);    void *skt = socket_connect(ctx, ZMQ_REQ, addr);    if (!skt){	syslog(LOG_CRIT,"KILLSERVER: unable to get req skt");	return -1;    }    uint8_t tn = table_number;    table_number = 0;    char *tmpstr = strdup(kill_str);    sendmore_msg_data(skt, tmpstr, strlen(tmpstr)+1, free_fn, NULL);    send_msg_vsm(skt, &tn, sizeof(uint8_t));    time_t curr_time;    time(&curr_time);    zmq_msg_t msg;    zmq_msg_init(&msg);    int err = 0;    do {	err = zmq_recvmsg(skt, &msg, ZMQ_NOBLOCK);	sleep(1);    } while (err && time(NULL) < curr_time + 10 );    if (err){	syslog(LOG_ERR, "KILLSERVER: no ack recieved - ignore");	err = -1;    }    zmq_close(skt);    syslog(LOG_DEBUG,"KILLSERVER: server killed");    return err;}

static char *s_recv(void *socket) {	zmq_msg_t message;	zmq_msg_init(&message);#if ZMQ_VERSION_MAJOR == 2	if (zmq_recv(socket, &message, 0) != 0) {		perror("zmq_recv");		return (NULL);	}#elif ZMQ_VERSION_MAJOR == 3	if (zmq_recvmsg(socket, &message, 0) == -1) {		perror("zmq_recvmsg");		return (NULL);	}#else# error "unsupported ZeroMQ major version"#endif	int size = zmq_msg_size(&message);	char *string = malloc(size + 1);	memcpy(string, zmq_msg_data(&message), size);	zmq_msg_close(&message);	string[size] = 0;	return (string);}

/* aux function to worker threads for sending the results */static int send_results(void *skt, uint8_t threadnb, uint32_t id, float cs){    syslog(LOG_DEBUG,"SEND: send thr = %u, id = %u, cs = %f", threadnb, id, cs);    sendmore_msg_vsm(skt, &threadnb, sizeof(uint8_t));    sendmore_msg_vsm(skt, &id, sizeof(uint32_t));    send_msg_vsm(skt, &cs, sizeof(float));    int err =0;    zmq_msg_t msg;    zmq_msg_init(&msg);    time_t curr_time;    time(&curr_time);    do {      err = zmq_recvmsg(skt, &msg, ZMQ_NOBLOCK);      sleep(1);    } while (err && time(NULL) < curr_time + 10 );    if (!err){      syslog(LOG_DEBUG,"SEND: reply recieved, msg size = %d", zmq_msg_size(&msg));    } else {      syslog(LOG_DEBUG,"SEND: no reply recieved");    }    return 0;}

static zmq_msg_t*read_msg(JNIEnv *env, void *socket, zmq_msg_t *msg, int flags){    int rc = zmq_msg_init (msg);    if (rc != 0) {        raise_exception (env, zmq_errno());        return NULL;    }#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)    rc = zmq_recvmsg (socket, msg, flags);#else    rc = zmq_recv (socket, msg, flags);#endif    int err = zmq_errno();    if (rc < 0 && err == EAGAIN) {        rc = zmq_msg_close (msg);        err = zmq_errno();        if (rc != 0) {            raise_exception (env, err);            return NULL;        }        return NULL;    }    if (rc < 0) {        raise_exception (env, err);        rc = zmq_msg_close (msg);        err = zmq_errno();        if (rc != 0) {            raise_exception (env, err);            return NULL;        }        return NULL;    }    return msg;}