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

 virtual void run() {     zmq_msg_t msg;     int rc = zmq_msg_init(&msg);     assert(rc == 0);     rc = zmq_msg_recv(&msg, socket_, ZMQ_DONTWAIT);     assert(rc != -1);     if (rc > 0) {         std::cout << id_ << ": Persisting data from member TODO." << std::endl;     } }

/** * Receive a single frame from a socket and store the results in a string. * The frame is not expected to contain a terminating NUL character. * @param socket The socket to receive from * @param str A reference to a string that will be set to the received value * @return -1 on error (--> str value is unchanged). 0 else */static inline int receiveStringFrame(void* socket, std::string& str) {    zmq_msg_t msg;    zmq_msg_init(&msg);    int rc = zmq_msg_recv(&msg, socket, 0);    if (rc == -1) {        return -1;    }    str = std::string((char*) zmq_msg_data(&msg), zmq_msg_size(&msg));    zmq_msg_close(&msg);    return 0;}

int main (void) {    void *context = zmq_ctx_new ();    //  Socket to receive messages on    void *receiver = zmq_socket (context, ZMQ_PULL);    zmq_connect (receiver, "tcp://localhost:5557");    //  Socket to send messages to    void *sender = zmq_socket (context, ZMQ_PUSH);    zmq_connect (sender, "tcp://localhost:5558");    //  Socket for control input    void *controller = zmq_socket (context, ZMQ_SUB);    zmq_connect (controller, "tcp://localhost:5559");    zmq_setsockopt (controller, ZMQ_SUBSCRIBE, "", 0);    //  Process messages from receiver and controller    zmq_pollitem_t items [] = {        { receiver, 0, ZMQ_POLLIN, 0 },        { controller, 0, ZMQ_POLLIN, 0 }    };    //  Process messages from both sockets    while (true) {        zmq_msg_t message;        zmq_poll (items, 2, -1);        if (items [0].revents & ZMQ_POLLIN) {            zmq_msg_init (&message);            zmq_msg_recv (&message, receiver, 0);            //  Do the work            s_sleep (atoi ((char *) zmq_msg_data (&message)));            //  Send results to sink            zmq_msg_init (&message);            zmq_msg_send (&message, sender, 0);            //  Simple progress indicator for the viewer            printf (".");            fflush (stdout);            zmq_msg_close (&message);        }        //  Any waiting controller command acts as 'KILL'        if (items [1].revents & ZMQ_POLLIN)            break;                      //  Exit loop    }    //  Finished    zmq_close (receiver);    zmq_close (sender);    zmq_close (controller);    zmq_ctx_destroy (context);    return 0;}

    void process( zeq::detail::Socket& socket )    {        zmq_msg_t msg;        zmq_msg_init( &msg );        zmq_msg_recv( &msg, socket.socket, 0 );        const std::string address( (const char*)zmq_msg_data( &msg ),                                   zmq_msg_size( &msg ));        _receiver.addConnection( std::string( "tcp://" ) + address );        zmq_msg_send( &msg, socket.socket, 0 );        zmq_msg_close( &msg );    }

static char* safe_recv_from_server (void *socket, int *size) {	zmq_msg_t message;	zmq_msg_init (&message);	*size = zmq_msg_recv (&message, socket, 0);	if (*size == -1)		return NULL;	char *string = malloc (*size + 1);	memcpy (string, zmq_msg_data (&message), *size);	zmq_msg_close (&message);	string [*size] = 0;	return (string);}

static void *Log_internal_thread(void *spec){    zmq_msg_t msg;    int rc = 0;    LogConfig *config = spec;    void *socket = zmq_socket(ZMQ_CTX, ZMQ_SUB);    check(socket, "Could not bind the logging subscribe socket.");    // warning: could cause threading problems if more than one of these    rc = zmq_setsockopt(socket, ZMQ_SUBSCRIBE, "", 0);    check(rc == 0, "Could not subscribe to the logger.");#ifdef ZMQ_LINGER    int opt = 0;    rc = zmq_setsockopt(socket, ZMQ_LINGER, &opt, sizeof(opt));    check(rc == 0, "Could not set the linger option.");#endif    rc = zmq_connect(socket, bdata(config->log_spec));    check(rc == 0, "Could connect to logging endpoint: %s", bdata(config->log_spec));    while(1) {       rc = zmq_msg_init(&msg);       check(rc == 0, "Failed to initialize message.");       rc = zmq_msg_recv(&msg, socket, 0);       if(rc == -1 && errno == ETERM) {           // The ZMQ context has been terminated, must be shutting down.           break;       }       check(rc >= 0, "Failed to receive from the zeromq logging socket");       check(zmq_msg_size(&msg) > 0, "Received poison pill, log thread exiting.");       fprintf(config->log_file, "%.*s", (int)zmq_msg_size(&msg), (char *)zmq_msg_data(&msg));       rc = zmq_msg_close(&msg);       check(rc == 0, "Message close failed.");    }    rc = zmq_close(socket);    check(rc == 0, "Could not close socket");    LogConfig_destroy(config);    return NULL;error:     LogConfig_destroy(config);    // could leak the socket and the msg but not much we can do    return NULL;}

	inline int recv_data(void * socket, function<int(char * data, size_t len)> process_input){		zmq_msg_t msg;		int rc = zmq_msg_init(&msg);		assert(rc == 0);		rc = zmq_msg_recv(&msg, socket, 0);		assert(rc >= 0);		rc = process_input(reinterpret_cast<char*>(zmq_msg_data(&msg)), zmq_msg_size(&msg));		zmq_msg_close(&msg);		return rc;	}

void test_multiple_connects (){    const unsigned int no_of_connects = 10;    void *ctx = zmq_ctx_new ();    assert (ctx);    int rc;    void *connectSocket [no_of_connects];    // Connect first    for (unsigned int i = 0; i < no_of_connects; ++i)    {        connectSocket [i] = zmq_socket (ctx, ZMQ_PUSH);        assert (connectSocket [i]);        rc = zmq_connect (connectSocket [i], "inproc://multiple");        assert (rc == 0);        // Queue up some data        rc = zmq_send_const (connectSocket [i], "foobar", 6, 0);        assert (rc == 6);    }    // Now bind    void *bindSocket = zmq_socket (ctx, ZMQ_PULL);    assert (bindSocket);    rc = zmq_bind (bindSocket, "inproc://multiple");    assert (rc == 0);    for (unsigned int i = 0; i < no_of_connects; ++i)    {        // Read pending message        zmq_msg_t msg;        rc = zmq_msg_init (&msg);        assert (rc == 0);        rc = zmq_msg_recv (&msg, bindSocket, 0);        assert (rc == 6);        void *data = zmq_msg_data (&msg);        assert (memcmp ("foobar", data, 6) == 0);    }    // Cleanup    for (unsigned int i = 0; i < no_of_connects; ++i)    {        rc = zmq_close (connectSocket [i]);        assert (rc == 0);    }    rc = zmq_close (bindSocket);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

int main(void){        void *context = zmq_ctx_new();        void *receiver = zmq_socket(context, ZMQ_PULL);        zmq_connect(receiver, "tcp://localhost:5557");        void *subscriber = zmq_socket(context, ZMQ_SUB);        zmq_connect(subscriber, "tcp://localhost:5556");        zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "10001 ", 6);        zmq_pollitem_t items[] = {                {receiver, 0, ZMQ_POLLIN, 0},                {subscriber, 0, ZMQ_POLLIN, 0},        };        while(1) {                zmq_msg_t message;                zmq_poll(items, 2, -1);                if (items[0].revents & ZMQ_POLLIN) {                        zmq_msg_init(&message);                        zmq_msg_recv(&message, receiver, 0);                        printf("receiver: %s/n", (char *)zmq_msg_data(&message));                        zmq_msg_close(&message);                }                if (items[1].revents & ZMQ_POLLIN) {                        zmq_msg_init(&message);                        zmq_msg_recv(&message, subscriber, 0);                        printf("subscribe: %s/n", (char *)zmq_msg_data(&message));                        zmq_msg_close(&message);                }        }        zmq_close(receiver);        zmq_close(subscriber);        zmq_ctx_destroy(context);        return 0;}

static inline void switch_msg_from_to( void *from, void *to ) {        zmq_msg_t msg;        int more = 1;        while( more ) {                zmq_msg_init( &msg );                int ret = zmq_msg_recv( &msg, from, 0 );                assert( ret != -1 );                more = zmq_msg_more( &msg );                ret = zmq_msg_send( &msg, to, more ? ZMQ_SNDMORE : 0 );                assert( ret != -1 );        }}

static void_zmq2agent_receive_acks (struct _zmq2agent_ctx_s *ctx){	int rc;	zmq_msg_t msg;	do {		zmq_msg_init (&msg);		rc = zmq_msg_recv (&msg, ctx->zagent, ZMQ_DONTWAIT);		if (rc > 0)			_zmq2agent_manage_ack (ctx, &msg);		zmq_msg_close (&msg);	} while (rc >= 0);}

//  Receive 0MQ string from socket and convert into C stringstatic char *s_recv (void *socket) {    zmq_msg_t message;    zmq_msg_init (&message);    int size = zmq_msg_recv (&message, socket, 0);    if (size == -1)        return NULL;    char *string = malloc (size + 1);    memcpy (string, zmq_msg_data (&message), size);    zmq_msg_close (&message);    string [size] = 0;    return (string);}

void ConnectedApp::initConnection(string name_in) {    int rc;    name_client = zmq_socket(context, ZMQ_REQ);    rc = zmq_connect(name_client, "ipc:///tmp/name");    // send a name request    Message m("name");    m.send(name_client);    cout << "ConnectedApp: send name request " << endl << flush;    // poll for a few seconds    zmq_pollitem_t items [] = {        { name_client,  0, ZMQ_POLLIN, 0 }    };    zmq_poll(items, 1, 2 * 1000);    if (items[0].revents & ZMQ_POLLIN) {        // use the received name        zmq_msg_t msg;        zmq_msg_init(&msg);        int r = zmq_msg_recv(&msg, name_client, 0);        cout << "ConnectedApp: got name reply " << r << endl << flush;        Message in_m(&msg);        messages.push_back(in_m.message_string);        zmq_msg_close(&msg);        cout << "got name " << in_m.name << endl;        this->name = in_m.name;        address.str = "ipc:///tmp/" + name;        server = zmq_socket(context, ZMQ_REP);        rc = zmq_bind(server, address.str.c_str());    } else {        // if did not get a reply, use the tcp socket        address.str = "tcp://*:" + ofToString(port_number);        server = zmq_socket(context, ZMQ_REP);        rc = zmq_bind(server, address.str.c_str());        cout << "ConnectedApp: no name reply" << endl << flush;    }}

static void client_loop_frontend (client_state* state, void *frontend){    zmq_msg_t zmessage;    while (1) {        // Process all parts of the message        //printf("RECEIVED  /n");                zmq_msg_init (&zmessage);        zmq_msg_recv (&zmessage, frontend, 0);                int more = zmq_msg_more (&zmessage);                if(more)        {            zmq_msg_close (&zmessage);                  continue;        }                zchat_message* message = zchat_message_deserialize_from_zmq_msg(&zmessage);        if(message == NULL)        {            zchat_log("Message deserialisation error");            zmq_msg_close (&zmessage);            return;        }                if(message->type() == zchat_message_message_type_PONG)        {            //ECHO_2_STR("PONG", message->ShortDebugString().c_str());;            process_pong_message(state, message);        }                if(message->type() == zchat_message_message_type_PING)        {            //ECHO_2_STR("PING", message->ShortDebugString().c_str());            client_state_reset_heartbeat(state);            client_state_set_heartbeat_time(state);            add_pong_message(state);        }                client_state_reset_heartbeat(state);        client_state_set_heartbeat_time(state);        ECHO_2_STR("RECEIVED", message->ShortDebugString().c_str());                zchat_message_destroy(message);        zmq_msg_close (&zmessage);        break; // Last message part    }}

static void * worker_routine (void *context) {    char *query_string, *datos_jugador;    query_string  = NULL;    query_string  = (char *) malloc(10 * sizeof(char));    datos_jugador = NULL;    datos_jugador = (char *) malloc(10 * sizeof(char));    //  Socket to talk to dispatcher    void *receiver = zmq_socket (context, ZMQ_REP);    zmq_connect (receiver, "inproc://example");    while (1) {        zmq_msg_t request;        zmq_msg_init(&request);        zmq_msg_recv(&request,receiver,0);                query_string = (char *)zmq_msg_data(&request);        zmq_msg_close(&request);                printf("%s/n", query_string);        if(strcmp(query_string,"%") == 0){            datos_jugador = "1|1,2,3,4,5,6,7,8,9|0";        }else{            datos_jugador = query_string;        }        printf("%s/n", datos_jugador);        sleep (1);          //  Do some 'work'        //Envio de respuesta al cliente        zmq_msg_t reply;        zmq_msg_init_size(&reply, 23);        memcpy(zmq_msg_data(&reply), datos_jugador, 23);        zmq_msg_send(&reply,receiver,0);                zmq_msg_close(&reply);        char *string = s_recv (receiver);        printf ("Received request: [%s]/n", string);        free (string);        //  Do some 'work'        sleep (1);        //  Send reply back to client        s_send (receiver, "World");    }    zmq_close (receiver);    return NULL;}

    void unblock(const char * sink)    {      zmq_send(zmq_requester, "UBL", 3, ZMQ_SNDMORE);      zmq_send(zmq_requester, sink, strlen(sink), 0);      zmq_msg_t reply;      zmq_msg_init(&reply);      int reply_size = zmq_msg_recv(&reply, zmq_requester, 0);      if(reply_size < 0)      {	std::cout << "adaptor: ublock: reply error " << errno << std::endl;      }     }

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 proccss_sub_msg(uv_poll_t *req, int status, int events){	void * sub = req->data;	assert(sub == g_sub_socket);	if (!(events & UV_READABLE))		return;	while (true) {		int zevents;		size_t zlen = sizeof(zevents);		zmq_getsockopt(sub, ZMQ_EVENTS, &zevents, &zlen);		if (!(zevents & ZMQ_POLLIN))			break;		zmq_msg_t cmd;		zmq_msg_init(&cmd);		zmq_msg_recv(&cmd, sub, 0);		if (!zmq_msg_more(&cmd)) {			zmq_msg_close(&cmd);			return;		}		zmq_msg_t data;		zmq_msg_init(&data);		zmq_msg_recv(&data, sub, 0);		if (!zmq_msg_more(&data)) {			handle_sub_msg(cmd, data);		}		else {			while (zmq_msg_more(&data)) {				zmq_msg_recv(&data, sub, 0);			}		}		zmq_msg_close(&cmd);		zmq_msg_close(&data);	}}

void test_connect_before_bind_pub_sub (){    void *ctx = zmq_ctx_new ();    assert (ctx);    // Connect first    void *connectSocket = zmq_socket (ctx, ZMQ_PUB);    assert (connectSocket);    int rc = zmq_connect (connectSocket, "inproc://cbbps");    assert (rc == 0);    // Queue up some data, this will be dropped    rc = zmq_send_const (connectSocket, "before", 6, 0);    assert (rc == 6);    // Now bind    void *bindSocket = zmq_socket (ctx, ZMQ_SUB);    assert (bindSocket);    rc = zmq_setsockopt (bindSocket, ZMQ_SUBSCRIBE, "", 0);    assert (rc == 0);    rc = zmq_bind (bindSocket, "inproc://cbbps");    assert (rc == 0);    // Wait for pub-sub connection to happen    msleep (SETTLE_TIME);    // Queue up some data, this not will be dropped    rc = zmq_send_const (connectSocket, "after", 6, 0);    assert (rc == 6);    // Read pending message    zmq_msg_t msg;    rc = zmq_msg_init (&msg);    assert (rc == 0);    rc = zmq_msg_recv (&msg, bindSocket, 0);    assert (rc == 6);    void *data = zmq_msg_data (&msg);    assert (memcmp ("after", data, 5) == 0);    // Cleanup    rc = zmq_close (connectSocket);    assert (rc == 0);    rc = zmq_close (bindSocket);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

bool Service::subscribe( const std::string& brokerAddress,                         const Publisher& publisher ){    void* context = zmq_ctx_new();    void* socket = zmq_socket( context, ZMQ_REQ );    const std::string zmqAddress = std::string("tcp://" ) + brokerAddress;    if( zmq_connect( socket, zmqAddress.c_str( )) == -1 )    {        ZEQINFO << "Can't reach connection broker at " << brokerAddress                << std::endl;        zmq_close( socket );        zmq_ctx_destroy( context );        return false;    }    const std::string& address = publisher.getAddress();    zmq_msg_t request;    zmq_msg_init_size( &request, address.size( ));    memcpy( zmq_msg_data( &request ), address.c_str(), address.size( ));    if( zmq_msg_send( &request, socket, 0 ) == -1 )    {        zmq_msg_close( &request );        ZEQINFO << "Can't send connection request " << address << " to "                << brokerAddress << ": " << zmq_strerror( zmq_errno( ))                << std::endl;        return false;    }    zmq_msg_close( &request );    zmq_msg_t reply;    zmq_msg_init( &reply );    if( zmq_msg_recv( &reply, socket, 0 )  == -1 )    {        zmq_msg_close( &reply );        ZEQINFO << "Can't receive connection reply from " << brokerAddress                << std::endl;        return false;    }    const std::string result( (const char*)zmq_msg_data( &reply ),                              zmq_msg_size( &reply ));    zmq_msg_close( &reply );    zmq_close( socket );    zmq_ctx_destroy( context );    return address == std::string( result );}

    bool eof(const char * sink)    {      zmq_send(zmq_requester, "EOF", 3, ZMQ_SNDMORE);      zmq_send(zmq_requester, sink, strlen(sink), 0);      zmq_msg_t reply;      zmq_msg_init(&reply);      int reply_size = zmq_msg_recv(&reply, zmq_requester, 0);      if(reply_size < 0)      {	std::cout << "adaptor: eof: reply error " << errno << std::endl;      }      return reply_size == 0;    }

/** * Receive a single-byte boolean-representing frame. * If the byte is 0, the resulting boolean is expected to be false. * Else, the resulting boolean i * @param socket The socket to receive from * @param str A reference to a string that will be set to the received value * @return < 0 on error. 0 on false result, 1 on true result. */static inline int receiveBooleanFrame(void* socket) {    zmq_msg_t msg;    zmq_msg_init(&msg);    int rc = zmq_msg_recv(&msg, socket, 0);    if (rc == -1) {        return -1;    }    if(zmq_msg_size(&msg) != 1) {        zmq_msg_close(&msg);        return -2; //Frame size mismatch    }    int val = (((char*)zmq_msg_data(&msg))[0] == 0 ? 0 : 1);    zmq_msg_close(&msg);    return val;}

void test_unbind (){    void *ctx = zmq_ctx_new ();    assert (ctx);    // Bind and unbind socket 1    void *bindSocket1 = zmq_socket (ctx, ZMQ_PAIR);    assert (bindSocket1);    int rc = zmq_bind (bindSocket1, "inproc://unbind");    assert (rc == 0);    zmq_unbind (bindSocket1, "inproc://unbind");    assert (rc == 0);    // Bind socket 2    void *bindSocket2 = zmq_socket (ctx, ZMQ_PAIR);    assert (bindSocket2);    rc = zmq_bind (bindSocket2, "inproc://unbind");    assert (rc == 0);    // Now connect    void *connectSocket = zmq_socket (ctx, ZMQ_PAIR);    assert (connectSocket);    rc = zmq_connect (connectSocket, "inproc://unbind");    assert (rc == 0);    // Queue up some data    rc = zmq_send_const (connectSocket, "foobar", 6, 0);    assert (rc == 6);    // Read pending message    zmq_msg_t msg;    rc = zmq_msg_init (&msg);    assert (rc == 0);    rc = zmq_msg_recv (&msg, bindSocket2, 0);    assert (rc == 6);    void *data = zmq_msg_data (&msg);    assert (memcmp ("foobar", data, 6) == 0);    // Cleanup    rc = zmq_close (connectSocket);    assert (rc == 0);    rc = zmq_close (bindSocket1);    assert (rc == 0);    rc = zmq_close (bindSocket2);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

    void setup_pipe(action &a)    {      zmq_send(zmq_requester, "PIP", 3, ZMQ_SNDMORE);      zmq_send(zmq_requester, a.sink.c_str(), a.sink.size(), ZMQ_SNDMORE);      zmq_send(zmq_requester, a.log_id.c_str(), a.log_id.size(), ZMQ_SNDMORE);      zmq_send(zmq_requester, a.data_id.c_str(), a.data_id.size(), 0);      zmq_msg_t reply;      zmq_msg_init(&reply);      int reply_size = zmq_msg_recv(&reply, zmq_requester, 0);      if(reply_size < 0)      {	std::cout << "adaptor: pipe reply error " << errno << std::endl;      }    }

char * tracer_string_recv(void *socket) {  zmq_msg_t msg;  size_t message_size;  char *message;  int rc = zmq_msg_init(&msg);  assert(rc == 0);  rc = zmq_msg_recv(&msg, socket, 0);  assert(rc != -1);  message_size = zmq_msg_size(&msg);  message = (char *)malloc(message_size +1);  memcpy(message, zmq_msg_data(&msg), message_size);  message[message_size] = 0;  zmq_msg_close(&msg);  return message;}

void ConnectedApp::processMessages() {    zmq_msg_t in_msg;    zmq_msg_init(&in_msg);    if (zmq_msg_recv(&in_msg, server, ZMQ_DONTWAIT) != -1) {        // reply to service messages        Message in_m(&in_msg);        if (in_m.name == "ping") {            // 1) ping message            Message m("status");            m.addValue("ok");            m.send(server);        } else if (in_m.name == "input_ports") {            // ports info request            Message m("ports");            for (int i = 0; i < ports_in.size(); i++) {                m.addValue(ports_in[i]->name);            }            m.send(server);        } else if (in_m.name == "output_ports") {            // ports info request            Message m("ports");            for (int i = 0; i < ports_out.size(); i++) {                m.addValue(ports_out[i]->name);            }            m.send(server);        } else {            // 1) ping message            Message m("unknown");            m.send(server);        }        messages.push_back(in_m.message_string);    }    zmq_msg_close(&in_msg);}

static ssize_tngx_zeromq_recvmsg(void *zmq, ngx_event_t *ev, zmq_msg_t *msg){    int  more;    for (;;) {        if (zmq_msg_recv(msg, zmq, ZMQ_DONTWAIT) == -1) {            if (ngx_errno == NGX_EAGAIN) {                ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, 0,                               "zmq_recv: not ready");                ev->ready = 0;                return NGX_AGAIN;            }            if (ngx_errno == NGX_EINTR) {                ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, 0,                               "zmq_recv: interrupted");                ev->ready = 0;                continue;            }            ngx_zeromq_log_error(ev->log, "zmq_msg_recv()");            ev->error = 1;            return NGX_ERROR;        }        break;    }    more = zmq_msg_get(msg, ZMQ_MORE);    if (more == -1) {        ngx_zeromq_log_error(ev->log, "zmq_msg_more()");        ev->error = 1;        return NGX_ERROR;    }    ev->eof = more ? 0 : 1;    ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,                   "zmq_recv: %uz eom:%d", zmq_msg_size(msg), ev->eof);    return zmq_msg_size(msg);}

static int process_msg(void *sock, unsigned char *exit_status){	int ret;	CqReq *req;	zmq_msg_t msg;	void *buf;	size_t len;	ret = zmq_msg_init(&msg);	if (ret < 0)		return -1;	ret = zmq_msg_recv(&msg, sock, 0);	if (ret < 0) {		zmq_msg_close(&msg);		if (errno != EINTR) {			fprintf(stderr, "Failed to receive message/n");			return -1;		}		return -2;	}	len = zmq_msg_size(&msg);	buf = zmq_msg_data(&msg);	printf("Got message (%zu)/n", len);	req = cq_req__unpack(NULL, len, buf);	if (req == NULL) {		fprintf(stderr, "Failed to unpack message/n");		zmq_msg_close(&msg);		return -1;	}	printf("New OP/n");	ret = run(req->command, req->n_args, req->args, req->n_env, req->env, exit_status);	cq_req__free_unpacked(req, NULL);	zmq_msg_close(&msg);	return ret;}

void got_new_subscriber(void * const s){	zmq_msg_t msg;	assert(!zmq_msg_init(&msg));	assert(zmq_msg_recv(&msg, s, 0) >= 0);	if (zmq_msg_size(&msg) < 1)		goto out;		uint8_t * const data = zmq_msg_data(&msg);	if (!data[0])		goto out;		zmq_send_protobuf(zmq_pub, pb_event, &current_pbe, 0);	out:	zmq_msg_close(&msg);}

void test_multiple_threads (){    const unsigned int no_of_threads = 30;    void *ctx = zmq_ctx_new ();    assert (ctx);    int rc;    void *threads [no_of_threads];    // Connect first    for (unsigned int i = 0; i < no_of_threads; ++i)    {        threads [i] = zmq_threadstart (&pusher, ctx);    }    // Now bind    void *bindSocket = zmq_socket (ctx, ZMQ_PULL);    assert (bindSocket);    rc = zmq_bind (bindSocket, "inproc://sink");    assert (rc == 0);    for (unsigned int i = 0; i < no_of_threads; ++i)    {        // Read pending message        zmq_msg_t msg;        rc = zmq_msg_init (&msg);        assert (rc == 0);        rc = zmq_msg_recv (&msg, bindSocket, 0);        assert (rc == 6);        void *data = zmq_msg_data (&msg);        assert (memcmp ("foobar", data, 6) == 0);    }    // Cleanup    for (unsigned int i = 0; i < no_of_threads; ++i)    {        zmq_threadclose (threads [i]);    }    rc = zmq_close (bindSocket);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}