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

static void send_response(void *sock, int internal_status, int exit_status){	void *buf;	int len;	CqRep rep = CQ_REP__INIT;	zmq_msg_t msg;	zmq_msg_init(&msg);	rep.exit_status = exit_status;	rep.internal_status = internal_status;	len = cq_rep__get_packed_size(&rep);	buf = malloc(len);	cq_rep__pack(&rep, buf);	zmq_msg_init_data(&msg, buf, len, free_buf, NULL);	zmq_msg_send(&msg, sock, 0);	zmq_msg_close(&msg);}

/** * 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);    int rc = zmq_msg_init (message);    if (rc != 0) {        raise_exception (env, zmq_errno());        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    int 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;}

static inline int handler_recv_parse(Handler *handler, HandlerParser *parser){    check(handler->running, "Called while handler wasn't running, that's not good.");    zmq_msg_t *inmsg = calloc(sizeof(zmq_msg_t), 1);    int rc = 0;    check_mem(inmsg);    rc = zmq_msg_init(inmsg);    check(rc == 0, "Failed to initialize message.");    taskstate("recv");    rc = mqrecv(handler->recv_socket, inmsg, ZMQ_NOBLOCK);    check(rc == 0, "Receive on handler socket failed.");    check(handler->running, "Received shutdown notification, goodbye.");    rc = HandlerParser_execute(parser, zmq_msg_data(inmsg), zmq_msg_size(inmsg));    check(rc == 1, "Failed to parse message from handler.");    check(parser->target_count > 0, "Message sent had 0 targets: %.*s",            (int)zmq_msg_size(inmsg), (char *)zmq_msg_data(inmsg));    debug("Parsed message with %d targets, first: %d, uuid: %s, and body: %d",            (int)parser->target_count, parser->targets[0],            bdata(parser->uuid), blength(parser->body));    zmq_msg_close(inmsg);    free(inmsg);    return 0;error:    if(inmsg) {        zmq_msg_close(inmsg);        free(inmsg);    }    return -1;}

int main(int argc, char *argv[]){    if (argc < 3) return EXIT_FAILURE;    int M = atoi(argv[1]);    int N = atoi(argv[2]);    printf("M: %d, N: %d/n", M, N);    void *ctx = zmq_init(1);    void *b = zmq_socket(ctx, ZMQ_PAIR);    zmq_connect(b, "tcp://localhost:4444");    zmq_msg_t msg;    int val[M];    long long start_time = sc_time();    int i;    for (i=0; i<N; i++) {        int *buf = (int *)malloc(M * sizeof(int));        memset(val, i, M * sizeof(int));        memcpy(buf, val, M * sizeof(int));        zmq_msg_init_data(&msg, buf, M * sizeof(int), _dealloc, NULL);        zmq_send(b, &msg, 0);        zmq_msg_close(&msg);        zmq_msg_init(&msg);        zmq_recv(b, &msg, 0);        memcpy(val, (int *)zmq_msg_data(&msg), zmq_msg_size(&msg));        zmq_msg_close(&msg);    }    long long end_time = sc_time();    printf("zmq_a: Time elapsed: %f sec/n", sc_time_diff(start_time, end_time));    zmq_close(b);    zmq_term(ctx);    return EXIT_SUCCESS;}

int zmq::pair_t::xsend (zmq_msg_t *msg_, int flags_){    if (outpipe == NULL || !outpipe_alive) {        errno = EAGAIN;        return -1;    }    if (!outpipe->write (msg_)) {        outpipe_alive = false;        errno = EAGAIN;        return -1;    }    if (!(flags_ & ZMQ_SNDMORE))        outpipe->flush ();    //  Detach the original message from the data buffer.    int rc = zmq_msg_init (msg_);    zmq_assert (rc == 0);    return 0;}

int main (void) {    setup_test_environment ();    void *ctx = zmq_ctx_new ();    assert (ctx);        void *stream = zmq_socket (ctx, ZMQ_STREAM);    assert (stream);    void *dealer = zmq_socket (ctx, ZMQ_DEALER);    assert (dealer);        int rc = zmq_bind (stream, "tcp://127.0.0.1:5555");    assert (rc >= 0);    rc = zmq_connect (dealer, "tcp://127.0.0.1:5555");    assert (rc >= 0);    zmq_send (dealer, "", 0, 0);        zmq_msg_t ident, empty;    zmq_msg_init (&ident);    rc = zmq_msg_recv (&ident, stream, 0);    assert (rc >= 0);    rc = zmq_msg_init_data (&empty, (void *) "", 0, NULL, NULL);    assert (rc >= 0);        rc = zmq_msg_send (&ident, stream, ZMQ_SNDMORE);    assert (rc >= 0);    rc = zmq_msg_close (&ident);    assert (rc >= 0);        rc = zmq_msg_send (&empty, stream, 0);    assert (rc >= 0);        //  This close used to fail with Bad Address    rc = zmq_msg_close (&empty);    assert (rc >= 0);        close_zero_linger (dealer);    close_zero_linger (stream);    zmq_ctx_term (ctx);}

int main (int argc, char const *argv[]) {  void* context = zmq_ctx_new();  //create a SUB socket  void* subscriber = zmq_socket(context, ZMQ_SUB);  const char* filter;    if(argc > 1) {    filter = argv[1];  } else {    filter = "Company1|";  }  printf("Collecting stock information from the server./n");  int conn = zmq_connect(subscriber, "tcp://localhost:4040");  // must set a subscription for SUB socket  conn = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter));  int i = 0;  for(i = 0; i < 10; i++) {    zmq_msg_t reply;    zmq_msg_init(&reply);    // receive the message, previous message is deallocated    zmq_msg_recv(&reply, subscriber, 0);        int length = zmq_msg_size(&reply);    char* value = malloc(length + 1);    memcpy(value, zmq_msg_data(&reply), length);    zmq_msg_close(&reply);    printf("%s/n", value);        free(value);  }  zmq_close(subscriber);  zmq_ctx_destroy(context);  return 0;}

void Executor::work(){    void *socket = zmq_socket(vcontext, ZMQ_REP);    zmq_msg_t zMsg;    char *cstring = NULL;    size_t msgSize = 0;        zmq_connect(socket, "inproc://storageWorkers");        while(vstop==false)    {        zmq_msg_init(&zMsg);        zmq_msg_recv(&zMsg, socket, 0);        msgSize = zmq_msg_size(&zMsg);        cstring = new char[msgSize+1];        memcpy(cstring, zmq_msg_data(&zMsg), msgSize);        zmq_msg_close(&zMsg);        cstring[msgSize] = '/0';        std::string str(cstring);        delete cstring;        Message* msg = vmsgFactory->parse(str);        Action *action = vactionFactory->parse(*msg);        delete msg;        msg = action->execute();        delete action;        msgSize = msg->toString().size()+1;        zmq_msg_init_size(&zMsg, msgSize);        memcpy(zmq_msg_data(&zMsg), msg->toString().c_str(), msgSize);        delete msg;        zmq_msg_send(&zMsg, socket, 0);        zmq_msg_close(&zMsg);    }    zmq_close(socket);}    

int main (void){    void *context = zmq_init (1);    //  This is where the weather server sits    void *frontend = zmq_socket (context, ZMQ_XSUB);    zmq_connect (frontend, "tcp://192.168.55.210:5556");    //  This is our public endpoint for subscribers    void *backend = zmq_socket (context, ZMQ_XPUB);    zmq_bind (backend, "tcp://10.1.1.0:8100");    //  Subscribe on everything    zmq_setsockopt (frontend, ZMQ_SUBSCRIBE, "", 0);    //  Shunt messages out to our own subscribers    while (1) {        while (1) {            zmq_msg_t message;            int64_t more;            //  Process all parts of the message            zmq_msg_init (&message);            zmq_recv (frontend, &message, 0);            size_t more_size = sizeof (more);            zmq_getsockopt (frontend, ZMQ_RCVMORE, &more, &more_size);            zmq_send (backend, &message, more? ZMQ_SNDMORE: 0);            zmq_msg_close (&message);            if (!more)                break;      //  Last message part        }    }    //  We don't actually get here but if we did, we'd shut down neatly    zmq_close (frontend);    zmq_close (backend);    zmq_term (context);    return 0;}

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

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

static void subscriber_thread_main (void *pvoid){    const proxy_hwm_cfg_t *cfg = (proxy_hwm_cfg_t *) pvoid;    const int idx = cfg->thread_idx;    void *subsocket = zmq_socket (cfg->context, ZMQ_SUB);    assert (subsocket);    set_hwm (subsocket);    TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (subsocket, ZMQ_SUBSCRIBE, 0, 0));    TEST_ASSERT_SUCCESS_ERRNO (      zmq_connect (subsocket, cfg->backend_endpoint[idx]));    //  Receive message_count messages    uint64_t rxsuccess = 0;    bool success = true;    while (success) {        zmq_msg_t msg;        int rc = zmq_msg_init (&msg);        assert (rc == 0);        rc = zmq_msg_recv (&msg, subsocket, 0);        if (rc != -1) {            TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));            rxsuccess++;        }        if (rxsuccess == message_count)            break;    }    //  Cleanup    zmq_close (subsocket);    printf ("subscriber thread ended/n");}

int zmq::xsub_t::xsend (zmq_msg_t *msg_, int options_){    size_t size = zmq_msg_size (msg_);    unsigned char *data = (unsigned char*) zmq_msg_data (msg_);    //  Malformed subscriptions are dropped silently.    if (size >= 1) {        //  Process a subscription.        if (*data == 1)            subscriptions.add (data + 1, size - 1);        //  Process an unsubscription. Invalid unsubscription is ignored.        if (*data == 0)            subscriptions.rm (data + 1, size - 1);    }    int rc = zmq_msg_close (msg_);    zmq_assert (rc == 0);    rc = zmq_msg_init (msg_);    zmq_assert (rc == 0);    return 0;}

void input_reaper(void * insock) {	while(1) {		zmq_msg_t input;		zmq_msg_init(&input);		if(zmq_msg_recv(&input, insock, ZMQ_DONTWAIT) == -1) {			if(errno == EAGAIN)				break;			else if(errno == EINTR)				continue;			syslog(LOG_ERR, "Error receiving message from interface: %s",				zmq_strerror(errno));			continue;		}		if(insock == pullsock)			process_pull_msg(&input);		else if(insock == reqsock)			process_req_msg(&input);		zmq_msg_close(&input);	}}

void Van::monitor() {  VLOG(1) << "starting monitoring...";  void *s = CHECK_NOTNULL(zmq_socket(context_, ZMQ_PAIR));  CHECK(!zmq_connect(s, "inproc://monitor"));  while (true) {    zmq_msg_t msg;    zmq_msg_init(&msg);    if (zmq_msg_recv(&msg, s, 0) == -1) {      if (errno == EINTR)        continue;      break;    }    uint8_t *data = (uint8_t *)zmq_msg_data(&msg);    int event = *(reinterpret_cast<uint16_t *>(data));    int value = *(reinterpret_cast<uint32_t *>(data + 2));    if (event == ZMQ_EVENT_DISCONNECTED) {      auto &manager = PostOffice::getInstance().manager();      if (isScheduler()) {        Lock l(fdToNodeIdMu_);        if (fdToNodeId_.find(value) == fdToNodeId_.end()) {          LOG(WARNING) << "cannot find the node id for Fd = " << value;          continue;        }        manager.nodeDisconnected(fdToNodeId_[value]);      } else {        manager.nodeDisconnected(scheduler_.id());      }    }    if (event == ZMQ_EVENT_MONITOR_STOPPED) {      break;    }  }  zmq_close(s);  VLOG(1) << "monitor stopped";}

void test_routing_id (){    //  Create the infrastructure    void *sc = test_context_socket (ZMQ_DEALER);    TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sc, "inproc://routing_id"));    void *sb = test_context_socket (ZMQ_ROUTER);    TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (sb, "inproc://routing_id"));    //  Send 2-part message.    TEST_ASSERT_EQUAL_INT (      1, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (sc, "A", 1, ZMQ_SNDMORE)));    TEST_ASSERT_EQUAL_INT (      1, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (sc, "B", 1, 0)));    //  Routing id comes first.    zmq_msg_t msg;    TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));    TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, sb, 0));    TEST_ASSERT_EQUAL_INT (1, zmq_msg_more (&msg));    //  Then the first part of the message body.    TEST_ASSERT_EQUAL_INT (      1, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, sb, 0)));    TEST_ASSERT_EQUAL_INT (1, zmq_msg_more (&msg));    //  And finally, the second part of the message body.    TEST_ASSERT_EQUAL_INT (      1, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, sb, 0)));    TEST_ASSERT_EQUAL_INT (0, zmq_msg_more (&msg));    //  Deallocate the infrastructure.    test_context_socket_close (sc);    test_context_socket_close (sb);}

void ZGWServer::pullThreadFunc(){    LOG_INFO << "PULL线程启动, TID: " << muduo::CurrentThread::tid();    assert( NULL != zmqContext_ );    std::string pull_endpoint = ini_.GetValue("backend", "pull_service", "");    if( pull_endpoint.size() == 0 )    {        LOG_ERROR << "未设置PULL socket的地址";        return;    }    void* pullSocket = zmq_socket(zmqContext_, ZMQ_PULL);    assert( NULL != pullSocket );    int rc = zmq_bind(pullSocket, pull_endpoint.c_str());    assert( rc == 0 );    LOG_INFO << "绑定PULL Socket成功，地址: " << pull_endpoint;    while(true)    {        zmq_msg_t msg_t;        rc = zmq_msg_init(&msg_t);        assert( 0 == rc );        rc = zmq_msg_recv(&msg_t, pullSocket, 0);        if( rc == -1 )        {            LOG_ERROR << "PULL线程接收消息失败, errno: " << errno;        }        else        {            responseMsg(msg_t);        }        zmq_msg_close(&msg_t);    }}

static ssize_tngx_zeromq_recv_part(void *zmq, ngx_event_t *rev, u_char *buf, size_t size){    zmq_msg_t  zmq_msg;    ssize_t    n;    if (zmq_msg_init(&zmq_msg) == -1) {        ngx_zeromq_log_error(rev->log, "zmq_msg_init()");        return NGX_ERROR;    }    n = ngx_zeromq_recvmsg(zmq, rev, &zmq_msg);    if (n < 0) {        goto done;    }    if ((size_t) n > size) {        ngx_log_error(NGX_LOG_ALERT, rev->log, 0,                      "zmq_recv: ZeroMQ message part too big (%uz) to fit"                      " into buffer (%uz)", n, size);        n = NGX_ERROR;        goto done;    }    ngx_memcpy(buf, zmq_msg_data(&zmq_msg), n);done:    if (zmq_msg_close(&zmq_msg) == -1) {        ngx_zeromq_log_error(rev->log, "zmq_msg_close()");        return NGX_ERROR;    }    return n;}

void rpc_call_accept_response(rpc_call_t *rpc) {  zmq_msg_t response;  int rc;  printf("rpc_call_accept_response/n");  rc = zmq_msg_init(&response);  rc = zmq_recv(rpc->socket, &response, ZMQ_NOBLOCK);  insist_return(rc == 0 || errno == EAGAIN, (void)(0),                "zmq_recv: expected success or EAGAIN, got errno %d:%s",                errno, strerror(errno));  if (rc == -1 && errno == EAGAIN) {    /* nothing to do, would block */    zmq_msg_close(&response);    return;  }  rpc_call_handle_response(rpc, &response);  zmq_msg_close(&response);  /* Free the 'rpc' call */  rpc_call_free(rpc);} /* rpc_call_accept_response */

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

//  Receives all message parts from socket, prints neatlyvoid s_dump (void *socket){    puts ("----------------------------------------");    while (1) {        //  Process all parts of the message        zmq_msg_t message;        zmq_msg_init (&message);        zmq_recv (socket, &message, 0);        //  Dump the message as text or binary        char *data = zmq_msg_data (&message);        int size = zmq_msg_size (&message);        int is_text = 1;        int char_nbr;        for (char_nbr = 0; char_nbr < size; char_nbr++)            if ((unsigned char) data [char_nbr] < 32            ||  (unsigned char) data [char_nbr] > 127)                is_text = 0;        printf ("[%03d] ", size);        for (char_nbr = 0; char_nbr < size; char_nbr++) {            if (is_text)                printf ("%c", data [char_nbr]);            else                printf ("%02X", (unsigned char) data [char_nbr]);        }        printf ("/n");        int64_t more;           //  Multipart detection        size_t more_size = sizeof (more);        zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);        zmq_msg_close (&message);        if (!more)            break;      //  Last message part    }}

int zmq::rep_t::xsend (zmq_msg_t *msg_, int flags_){    if (!waiting_for_reply) {        errno = EFSM;        return -1;    }    //  TODO: Implement this once queue limits are in-place. If the reply    //  overloads the buffer, connection should be torn down.    zmq_assert (reply_pipe->check_write (zmq_msg_size (msg_)));    //  Push message to the selected pipe.    reply_pipe->write (msg_);    reply_pipe->flush ();    waiting_for_reply = false;    reply_pipe = NULL;    //  Detach the message from the data buffer.    int rc = zmq_msg_init (msg_);    zmq_assert (rc == 0);    return 0;}

voids_recv_seq (void *socket, ...){    zmq_msg_t msg;    zmq_msg_init (&msg);    int more;    size_t more_size = sizeof(more);    va_list ap;    va_start (ap, socket);    const char * data = va_arg (ap, const char *);    while (true) {        int rc = zmq_msg_recv (&msg, socket, 0);        assert (rc != -1);        if (!data)            assert (zmq_msg_size (&msg) == 0);        else            assert (strcmp (data, (const char *)zmq_msg_data (&msg)) == 0);        data = va_arg (ap, const char *);        bool end = data == SEQ_END;        rc = zmq_getsockopt (socket, ZMQ_RCVMORE, &more, &more_size);        assert (rc == 0);        assert (!more == end);        if (end)            break;    }    va_end (ap);    zmq_msg_close (&msg);}

static inline int handler_recv_parse(Handler *handler, HandlerParser *parser){    zmq_msg_t *inmsg = NULL;    check(handler->running, "Called while handler wasn't running, that's not good.");    inmsg = calloc(sizeof(zmq_msg_t), 1);    int rc = 0;    check_mem(inmsg);    rc = zmq_msg_init(inmsg);    check(rc == 0, "Failed to initialize message.");    taskstate("recv");    rc = mqrecv(handler->recv_socket, inmsg, 0);    check(rc == 0, "Receive on handler socket failed.");    check(handler->running, "Handler marked as not running.");    rc = HandlerParser_execute(parser, zmq_msg_data(inmsg), zmq_msg_size(inmsg));    check(rc == 1, "Failed to parse message from handler.");    check(parser->target_count > 0, "Message sent had 0 targets: %.*s",            (int)zmq_msg_size(inmsg), (char *)zmq_msg_data(inmsg));    zmq_msg_close(inmsg);    free(inmsg);    return 0;error:    if(inmsg) {        zmq_msg_close(inmsg);        free(inmsg);    }    return -1;}

static int luazmq_skt_recv (lua_State *L) {  zsocket *skt = luazmq_getsocket(L);  zmq_msg_t msg;  int flags = luaL_optint(L,2,0);  int ret = zmq_msg_init(&msg);  if(-1 == ret) return luazmq_fail(L, skt);  ret = zmq_msg_recv(&msg, skt->skt, flags);  if(-1 == ret){    zmq_msg_close(&msg);    return luazmq_fail(L, skt);  }  lua_pushlstring(L, zmq_msg_data(&msg), zmq_msg_size(&msg));  if( zmq_msg_more(&msg) ){    skt->flags |= LUAZMQ_FLAG_MORE;    lua_pushboolean(L, 1);  }  else{    skt->flags &= ~LUAZMQ_FLAG_MORE;    lua_pushboolean(L, 0);  }  zmq_msg_close(&msg);  return 2;}

static int luazmq_skt_recvx (lua_State *L) {  zsocket *skt = luazmq_getsocket(L);  zmq_msg_t msg;  int flags = luaL_optint(L,2,0);  int i = 0;  lua_settop(L, 1);  while(1){    int ret = zmq_msg_init(&msg);    if(-1 == ret){      ret = luazmq_fail(L, skt);      {int j;for(j = ret; j >= 0; --j){        lua_insert(L, 1);      }}      return ret + i;    }    ret = zmq_msg_recv(&msg, skt->skt, flags);    if(-1 == ret){      zmq_msg_close(&msg);      ret = luazmq_fail(L, skt);      {int j;for(j = ret; j >= 0; --j){        lua_insert(L, 1);      }}      return ret + i;    }    i++;    lua_checkstack(L, i);    lua_pushlstring(L, zmq_msg_data(&msg), zmq_msg_size(&msg));    ret = zmq_msg_more(&msg);    zmq_msg_close(&msg);    if(!ret) break;  }  return i;}

static int luazmq_skt_sendv_impl(lua_State *L, int flags) {  zsocket *skt = luazmq_getsocket(L);  size_t i, size = 0, top = lua_gettop(L);  zmq_msg_t msg;  int ret;  for(i = 2; i<=top; ++i){    size_t s;    luaL_checklstring(L,i,&s);    size += s;  }  if (0 == size){    ret = zmq_msg_init(&msg);  }  else {    ret = zmq_msg_init_size(&msg, size);  }  if(-1 == ret) return luazmq_fail(L, skt);  {    size_t pos;    for(pos = 0, i = 2; i<=top; ++i){      const char *data = luaL_checklstring(L,i,&size);      memcpy((char*)zmq_msg_data(&msg) + pos, data, size);      pos += size;    }  }  ret = zmq_msg_send(&msg, skt->skt, flags);  zmq_msg_close(&msg);  if(-1 == ret) return luazmq_fail(L, skt);  return luazmq_pass(L);}

void pair_callback(int _, short __, void *pull){	zmq_msg_t msg;	uint32_t events;	size_t len;	len = sizeof(events);	zmq_getsockopt(pull, ZMQ_EVENTS, &events, &len);	if ( events & ZMQ_POLLIN )	{			while (1)		{			zmq_msg_init(&msg);			if ( zmq_recv(pull, &msg, ZMQ_NOBLOCK) == -1 )			{				zmq_msg_close(&msg);				break;			}			printf("recv: %s/n", (char *)zmq_msg_data(&msg));			zmq_msg_close(&msg);		}	}}

/* 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_recv(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;}