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

void *blitScreen(void *) {  void *context = zmq_ctx_new ();  void *requester;  requester = zmq_socket (context, ZMQ_REQ);  zmq_connect(requester, "tcp://mensadisplay:5556");  // zmq_connect(requester, "tcp://192.168.178.147:5570");  // zmq_connect(requester, "tcp://localhost:5556");  while(1) {    zmq_msg_t msg;    zmq_msg_init_size(&msg, sizeof(msgBlit) + WIDTH * HEIGHT);    memcpy(zmq_msg_data(&msg), screen, sizeof(msgBlit) + WIDTH * HEIGHT);    zmq_msg_send(&msg, requester, ZMQ_SNDMORE);    zmq_msg_init_size(&msg, 0);    zmq_msg_send(&msg, requester, 0);    zmq_recv(requester, NULL, 0, 0);  }  zmq_close(requester);  zmq_ctx_destroy(context);  return nullptr;}

int send_int_array(const int arr[], size_t count, role *r, const char *label){  int rc = 0;  zmq_msg_t msg;  size_t size = sizeof(int) * count;#ifdef __DEBUG__  fprintf(stderr, " --> %s ", __FUNCTION__);#endif  if (label != NULL) {#ifdef __DEBUG__    fprintf(stderr, "{label: %s}", label);#endif    zmq_msg_t msg_label;    char *buf_label = (char *)calloc(sizeof(char), strlen(label));    memcpy(buf_label, label, strlen(label));    zmq_msg_init_data(&msg_label, buf_label, strlen(label), _dealloc, NULL);    switch (r->type) {      case SESSION_ROLE_P2P:        rc = zmq_msg_send(r->p2p->ptr, &msg_label, ZMQ_SNDMORE);        break;      case SESSION_ROLE_GRP:        rc = zmq_msg_send(r->grp->out->ptr, &msg_label, ZMQ_SNDMORE);        break;    default:        fprintf(stderr, "%s: Unknown endpoint type: %d/n", __FUNCTION__, r->type);    }    zmq_msg_close(&msg_label);  }  int *buf = (int *)malloc(size);  memcpy(buf, arr, size);  zmq_msg_init_data(&msg, buf, size, _dealloc, NULL);  switch (r->type) {    case SESSION_ROLE_P2P:      rc = zmq_msg_send(r->p2p->ptr, &msg, 0);      break;    case SESSION_ROLE_GRP:#ifdef __DEBUG__      fprintf(stderr, "bcast -> %s(%d endpoints) ", r->grp->name, r->grp->nendpoint);#endif      rc = zmq_msg_send(r->grp->out->ptr, &msg, 0);      break;    default:      fprintf(stderr, "%s: Unknown endpoint type: %d/n", __FUNCTION__, r->type);  }  zmq_msg_close(&msg);  if (rc != 0) perror(__FUNCTION__); #ifdef __DEBUG__  fprintf(stderr, "./n");#endif  return rc;}

int main (void) {  void *context = zmq_ctx_new ();  void *requester;  requester = zmq_socket (context, ZMQ_REQ);  zmq_connect(requester, "tcp://mensadisplay:5556");  // zmq_connect(requester, "tcp://192.168.178.147:5570");  // zmq_connect(requester, "tcp://localhost:5556");  msgBlit *snow = reinterpret_cast<msgBlit *>(malloc(sizeof(msgBlit) + WIDTH * HEIGHT));  snow->cmd = CMD_BLIT;  snow->x = 0;  snow->y = 0;  snow->w = WIDTH;  snow->h = HEIGHT;  int on = 0;  while(1) {    zmq_msg_t msg;    zmq_msg_init_size(&msg, sizeof(msgBlit) + WIDTH * HEIGHT);    int d = 0;    switch(on) {      case 0: d = 0; break;      case 1: d = 0; break;      case 2: d = 0; break;      case 3: d = 0; break;      case 4: d = 255; break;      case 5: d = 255; break;      case 6: d = 0; break;      case 7: d = 255; break;    }    for(int i = 0; i < WIDTH * HEIGHT; ++i) {      snow->data[i] = d;    }    on = (on + 1) % 8;    memcpy(zmq_msg_data(&msg), snow, sizeof(msgBlit) + WIDTH * HEIGHT);    zmq_msg_send(&msg, requester, ZMQ_SNDMORE);    zmq_msg_init_size(&msg, 0);    zmq_msg_send(&msg, requester, 0);    zmq_recv(requester, NULL, 0, 0);  }  zmq_close(requester);  zmq_ctx_destroy(context);  return 0;}

void* pub_worker(void* arg){    void *ctx = arg;    assert(ctx);    void *qskt = zmq_socket(ctx, ZMQ_REP);    assert(qskt);    int rc = zmq_connect(qskt, "inproc://querys");    assert(rc == 0);    void *pubskt = zmq_socket(ctx, ZMQ_PUB);    assert(pubskt);    rc = zmq_bind(pubskt, "inproc://publish");    assert(rc == 0);    uint8_t cmd;    uint32_t nb;    zmq_msg_t topic_msg, cmd_msg, nb_msg, resp_msg;    zmq_msg_init_data(&topic_msg, topic_str, strlen(topic_str) , NULL, NULL);    fprintf(stdout,"WORKER: ready to receive messages/n");    //  NOTE: do NOT reuse this example code, It's broken.    //  e.g. topic_msg will be invalid the second time through    while (1){    zmq_msg_send(pubskt, &topic_msg, ZMQ_SNDMORE);    zmq_msg_init(&cmd_msg);    zmq_msg_recv(qskt, &cmd_msg, 0);    memcpy(&cmd, zmq_msg_data(&cmd_msg), sizeof(uint8_t));    zmq_msg_send(pubskt, &cmd_msg, ZMQ_SNDMORE);    zmq_msg_close(&cmd_msg);    fprintf(stdout, "received cmd %u/n", cmd);    zmq_msg_init(&nb_msg);    zmq_msg_recv(qskt, &nb_msg, 0);    memcpy(&nb, zmq_msg_data(&nb_msg), sizeof(uint32_t));    zmq_msg_send(pubskt, &nb_msg, 0);    zmq_msg_close(&nb_msg);    fprintf(stdout, "received nb %u/n", nb);    zmq_msg_init_size(&resp_msg, sizeof(uint8_t));    memset(zmq_msg_data(&resp_msg), 0, sizeof(uint8_t));    zmq_msg_send(qskt, &resp_msg, 0);    zmq_msg_close(&resp_msg);    }    return NULL;}

void AsyncJobRouter::forwardToJob(uint64_t apid,            zmq_msg_t* routingFrame,            zmq_msg_t* delimiterFrame,            zmq_msg_t* headerFrame) {    void* outSock = processSocketMap[apid];    zmq_msg_send(routingFrame, outSock, ZMQ_SNDMORE);    zmq_msg_send(delimiterFrame, outSock, ZMQ_SNDMORE);    //Only send MORE flag for the header frame if there are more  frames to follow    zmq_msg_send(headerFrame, outSock,        (socketHasMoreFrames(processorInputSocket) ? ZMQ_SNDMORE : 0));    //Proxy the rest of the message (if any)    proxyMultipartMessage(processorInputSocket, outSock);}

int main(void){        void *context = zmq_ctx_new();        void *frontend = zmq_socket(context, ZMQ_ROUTER);        void *backend = zmq_socket(context, ZMQ_DEALER);        zmq_bind(frontend, "tcp://*:5559");        zmq_bind(backend, "tcp://*:5560");        zmq_pollitem_t items[] = {                {frontend, 0, ZMQ_POLLIN, 0},                {backend, 0, ZMQ_POLLIN, 0},        };        while (1) {                zmq_msg_t message;                int more;                zmq_poll(items, 2, -1);                if (items[0].revents & ZMQ_POLLIN) {                        while (1) {                                zmq_msg_init(&message);                                zmq_msg_recv(&message, frontend, 0);                                printf("frontend recv: %s/n", (char *)zmq_msg_data(&message));                                size_t more_size = sizeof(more);                                zmq_getsockopt(frontend, ZMQ_RCVMORE, &more, &more_size);                                zmq_msg_send(&message, backend, more ? ZMQ_SNDMORE : 0);                                zmq_msg_close(&message);                                if (!more)                                        break;                        }                }                if (items[1].revents & ZMQ_POLLIN) {                        while (1) {                                zmq_msg_init (&message);                                zmq_msg_recv(&message, backend, 0);                                printf("backend: %s/n", (char *)zmq_msg_data(&message));                                size_t more_size = sizeof(more);                                zmq_getsockopt(backend, ZMQ_RCVMORE, &more, &more_size);                                zmq_msg_send(&message, frontend, more ? ZMQ_SNDMORE : 0);                                zmq_msg_close(&message);                                if (!more)                                        break;                        }                }        }        zmq_close(frontend);        zmq_close(backend);        zmq_ctx_destroy(context);        return 0;}

int main (void) {    //  Prepare our context and sockets    void *context = zmq_ctx_new ();    void *frontend = zmq_socket (context, ZMQ_ROUTER);    void *backend  = zmq_socket (context, ZMQ_DEALER);    zmq_bind (frontend, "tcp://*:5559");    zmq_bind (backend,  "tcp://*:5560");    //  Initialize poll set    zmq_pollitem_t items [] = {        { frontend, 0, ZMQ_POLLIN, 0 },        { backend,  0, ZMQ_POLLIN, 0 }    };    //  Switch messages between sockets    while (1) {        zmq_msg_t message;        zmq_poll (items, 2, -1);        if (items [0].revents & ZMQ_POLLIN) {            while (1) {                //  Process all parts of the message                zmq_msg_init (&message);                zmq_msg_recv (&message, frontend, 0);                zmq_msg_send (&message, backend,                    zmq_msg_more (&message)? ZMQ_SNDMORE: 0);                zmq_msg_close (&message);                if (!zmq_msg_more (&message))                    break;      //  Last message part            }        }        if (items [1].revents & ZMQ_POLLIN) {            while (1) {                //  Process all parts of the message                zmq_msg_init (&message);                zmq_msg_recv (&message, backend, 0);                zmq_msg_send (&message, frontend,                    zmq_msg_more (&message)? ZMQ_SNDMORE: 0);                zmq_msg_close (&message);                if (!zmq_msg_more (&message))                    break;      //  Last message part            }        }    }    //  We never get here but clean up anyhow    zmq_close (frontend);    zmq_close (backend);    zmq_ctx_destroy (context);    return 0;}

int main (void){    int major, minor, patch;    zmq_version (&major, &minor, &patch);    printf ("Current 0MQ version is %d.%d.%d/n", major, minor, patch);	    void *context = zmq_ctx_new ();    //  Socket to talk to clients    void *responder = zmq_socket (context, ZMQ_REP);    zmq_bind (responder, "tcp://*:5555");    while (1) {        //  Wait for next request from client	char *message = s_recv(responder);        printf ("Received %s/n", message);        //  Do some 'work'        sleep (1);        //  Send reply back to client        zmq_msg_t reply;        zmq_msg_init_size (&reply, 5);        memcpy (zmq_msg_data (&reply), "World", 5);        zmq_msg_send (&reply, responder, 0);        zmq_msg_close (&reply);    }    //  We never get here but if we did, this would be how we end    zmq_close (responder);    zmq_ctx_destroy (context);    return 0;}

void test_req_message_format (void *ctx){    void *req = zmq_socket (ctx, ZMQ_REQ);    assert (req);    void *router = zmq_socket (ctx, ZMQ_ROUTER);    assert (router);    int rc = zmq_bind (req, bind_address);    assert (rc == 0);    size_t len = MAX_SOCKET_STRING;    rc = zmq_getsockopt (req, ZMQ_LAST_ENDPOINT, connect_address, &len);    assert (rc == 0);    rc = zmq_connect (router, connect_address);    assert (rc == 0);    // Send a multi-part request.    s_send_seq (req, "ABC", "DEF", SEQ_END);    zmq_msg_t msg;    zmq_msg_init (&msg);    // Receive peer identity    rc = zmq_msg_recv (&msg, router, 0);    assert (rc != -1);    assert (zmq_msg_size (&msg) > 0);    zmq_msg_t peer_id_msg;    zmq_msg_init (&peer_id_msg);    zmq_msg_copy (&peer_id_msg, &msg);    int more = 0;    size_t more_size = sizeof (more);    rc = zmq_getsockopt (router, ZMQ_RCVMORE, &more, &more_size);    assert (rc == 0);    assert (more);    // Receive the rest.    s_recv_seq (router, 0, "ABC", "DEF", SEQ_END);    // Send back a single-part reply.    rc = zmq_msg_send (&peer_id_msg, router, ZMQ_SNDMORE);    assert (rc != -1);    s_send_seq (router, 0, "GHI", SEQ_END);    // Receive reply.    s_recv_seq (req, "GHI", SEQ_END);    rc = zmq_msg_close (&msg);    assert (rc == 0);    rc = zmq_msg_close (&peer_id_msg);    assert (rc == 0);    close_zero_linger (req);    close_zero_linger (router);    // Wait for disconnects.    msleep (SETTLE_TIME);}

 void reply() {   zmq_msg_t reply;   zmq_msg_init_size (&reply, 0);   zmq_msg_send (&reply, zmq_responder, 0);   zmq_msg_close (&reply); }

int main (void){    void *context = zmq_ctx_new ();    //  Socket to talk to server    printf ("Connecting to hello world server.../n");    void *requester = zmq_socket (context, ZMQ_REQ);    zmq_connect (requester, "tcp://localhost:5555");    int request_nbr;    for (request_nbr = 0; request_nbr != 10; request_nbr++) {        zmq_msg_t request;        zmq_msg_init_size (&request, 5);        memcpy (zmq_msg_data (&request), "Hello", 5);        printf ("Sending Hello %d.../n", request_nbr);        zmq_msg_send (&request, requester, 0);        zmq_msg_close (&request);        zmq_msg_t reply;        zmq_msg_init (&reply);        zmq_msg_recv (&reply, requester, 0);        printf ("Received World %d/n", request_nbr);        zmq_msg_close (&reply);    }    zmq_close (requester);    zmq_ctx_destroy (context);    return 0;}

SEXP R_zmq_msg_send(SEXP R_rmsg, SEXP R_socket, SEXP R_flags){	int C_rmsg_length = LENGTH(R_rmsg);	int C_ret = -1, C_errno, C_flags = INTEGER(R_flags)[0];	void *C_socket = R_ExternalPtrAddr(R_socket);	zmq_msg_t msg;	if(C_socket != NULL){		C_ret = zmq_msg_init_size(&msg, C_rmsg_length);		if(C_ret == -1){			C_errno = zmq_errno();			REprintf("R_zmq_msg_init_size errno: %d strerror: %s/n",				C_errno, zmq_strerror(C_errno));		}		memcpy(zmq_msg_data(&msg), RAW(R_rmsg), C_rmsg_length);		C_ret = zmq_msg_send(&msg, C_socket, C_flags);		if(C_ret == -1){			C_errno = zmq_errno();			REprintf("R_zmq_msg_send errno: %d strerror: %s/n",				C_errno, zmq_strerror(C_errno));		}		C_ret = zmq_msg_close(&msg);		if(C_ret == -1){			C_errno = zmq_errno();			REprintf("R_zmq_msg_close errno: %d strerror: %s/n",				C_errno, zmq_strerror(C_errno));		}	} else{		REprintf("R_zmq_send: C_socket is not available./n");	}	return(R_NilValue);} /* End of R_zmq_msg_send(). */

int main(void) {	void* context = zmq_ctx_new();	// Socket to talk to clients	void* responder = zmq_socket(context, ZMQ_REP);	zmq_bind(responder, "tcp://*:5555");	printf("Starting Hello World server on port 5555/n");	// Run server forever	while(1) {		// Wait for the next request from server		zmq_msg_t request;		zmq_msg_init (&request);		zmq_msg_recv (&request, responder, 0);		printf("Received Hello/n");		zmq_msg_close (&request);		// Do some work		sleep(1);		// Send reply back to client		zmq_msg_t reply;		zmq_msg_init_size(&reply, 6);		memcpy(zmq_msg_data(&reply), "World!", 6);		zmq_msg_send(&reply, responder, 0);		zmq_msg_close(&reply);	}	// We never get here, but clean up for style	zmq_close(responder);	zmq_ctx_destroy(context);	return 0;}

int main (int argc, char const *argv[]) {    void* context = zmq_ctx_new();  //create a PUB socket  void* publisher = zmq_socket(context, ZMQ_PUB);  int conn = zmq_bind(publisher, "tcp://*:4040");  conn = zmq_bind(publisher, "ipc://stock.ipc");    const char* companies[3] = {"Company1", "Company10", "Company101"};  for(;;) {    int price = count % 17;    int which_company = count % 3;    int index = strlen(companies[which_company]);    char update[64];    sprintf(update, "%s| %d", companies[which_company], price);    zmq_msg_t message;    zmq_msg_init_size(&message, index);    memcpy(zmq_msg_data(&message), update, index);    zmq_msg_send(&message, publisher, 0);    zmq_msg_close(&message);    count++;  }  zmq_close(publisher);  zmq_ctx_destroy(context);    return 0;}

int main (void){    void *context = zmq_ctx_new ();    //  Socket to talk to clients    void *responder = zmq_socket (context, ZMQ_REP);    zmq_bind (responder, "tcp://*:5555");    while (1) {        //  Wait for next request from client        zmq_msg_t request;        zmq_msg_init (&request);        zmq_msg_recv (&request, responder, 0);        printf ("Received Hello/n");        zmq_msg_close (&request);        //  Do some 'work'        sleep (1);        //  Send reply back to client        zmq_msg_t reply;        zmq_msg_init_size (&reply, 5);        memcpy (zmq_msg_data (&reply), "World", 5);        zmq_msg_send (&reply, responder, 0);        zmq_msg_close (&reply);    }    //  We never get here but if we did, this would be how we end    zmq_close (responder);    zmq_ctx_destroy (context);    return 0;}

static int Lzmq_send(lua_State *L){    zmq_ptr *s = luaL_checkudata(L, 1, MT_ZMQ_SOCKET);    size_t msg_size;    const char *data = luaL_checklstring(L, 2, &msg_size);    int flags = luaL_optint(L, 3, 0);    zmq_msg_t msg;    if(zmq_msg_init_size(&msg, msg_size) != 0) {        return Lzmq_push_error(L);    }    memcpy(zmq_msg_data(&msg), data, msg_size);    int rc = zmq_msg_send(&msg, s->ptr, flags);    if(zmq_msg_close(&msg) != 0) {        return Lzmq_push_error(L);    }    if (rc == -1) {        return Lzmq_push_error(L);    }    lua_pushboolean(L, 1);    return 1;}

int main (int argc, char const *argv[]) {    void* context = zmq_ctx_new();  printf("Client Starting..../n");  void* request = zmq_socket(context, ZMQ_REQ);  zmq_connect(request, "tcp://localhost:4040");  int count = 0;  for(;;) {    zmq_msg_t req;    zmq_msg_init_size(&req, 5);    memcpy(zmq_msg_data(&req), "hello", 5);    printf("Sending: hello - %d/n", count);    zmq_msg_send(&req, request, 0);    zmq_msg_close(&req);        zmq_msg_t reply;    zmq_msg_init(&reply);    zmq_msg_recv(&reply, request, 0);    printf("Received: hello - %d/n", count);    zmq_msg_close(&reply);    count++;  }  zmq_close(request);  zmq_ctx_destroy(context);    return 0;}

int Log_request(Connection *conn, int status, int size){    zmq_msg_t msg;    if(LOG_SOCKET == NULL)         return 0;    bstring log_data = make_log_message(conn->req, conn->remote, conn->rport, status, size);    check_mem(log_data);    int rc = zmq_msg_init_data(&msg, bdata(log_data), blength(log_data),            free_log_msg, log_data);    check(rc == 0, "Could not craft message for log message '%s'.", bdata(log_data));        rc = zmq_msg_send(&msg, LOG_SOCKET, 0);    check(rc >= 0, "Could not send log message to socket.");    log_data = NULL; // that way errors from above can clean the log_data    rc = zmq_msg_close(&msg);    check(rc == 0, "Failed to close message object.");    return 0;error:    bdestroy(log_data);    zmq_msg_close(&msg);    return -1;}

int main(void){    void* zmq_ctx = zmq_ctx_new();    void* zmq_sock = zmq_socket(zmq_ctx, ZMQ_REP);    zmq_connect(zmq_sock, "tcp://localhost:5560");    int seq = 0;    while (1)    {        printf("/nloop seq: %d/n", seq++);   //     char buf[MSG_SIZE + 1] = {0};        int more = 1;        zmq_msg_t message;        while (1)        {            zmq_msg_init(&message);            zmq_msg_recv(&message, zmq_sock, 0);            printf("recv/send: size = %d/n", zmq_msg_size(&message));            uint32_t more_size = sizeof(more);            zmq_getsockopt(zmq_sock, ZMQ_RCVMORE, &more, &more_size);            zmq_msg_send(&message, zmq_sock, more ? ZMQ_SNDMORE : 0);            zmq_msg_close(&message);            if (!more)                break;        }        sleep(1);    }    zmq_close(zmq_sock);    zmq_ctx_destroy(zmq_ctx);    return 0;}

intzmailer_msg_send (zmailer_msg_t *self, zsock_t *output){    assert (self);    assert (output);    if (zsock_type (output) == ZMQ_ROUTER)        zframe_send (&self->routing_id, output, ZFRAME_MORE + ZFRAME_REUSE);    size_t frame_size = 2 + 1;          //  Signature and message ID    switch (self->id) {        case ZMAILER_MSG_MAIL:            frame_size += 2;            //  version            frame_size += 1 + strlen (self->from);            frame_size += 4;            if (self->to)                frame_size += strlen (self->to);            frame_size += 4;            if (self->subject)                frame_size += strlen (self->subject);            frame_size += 4;            if (self->request)                frame_size += strlen (self->request);            break;    }    //  Now serialize message into the frame    zmq_msg_t frame;    zmq_msg_init_size (&frame, frame_size);    self->needle = (byte *) zmq_msg_data (&frame);    PUT_NUMBER2 (0xAAA0 | 0);    PUT_NUMBER1 (self->id);    size_t nbr_frames = 1;              //  Total number of frames to send    switch (self->id) {        case ZMAILER_MSG_MAIL:            PUT_NUMBER2 (1);            PUT_STRING (self->from);            if (self->to) {                PUT_LONGSTR (self->to);            }            else                PUT_NUMBER4 (0);    //  Empty string            if (self->subject) {                PUT_LONGSTR (self->subject);            }            else                PUT_NUMBER4 (0);    //  Empty string            if (self->request) {                PUT_LONGSTR (self->request);            }            else                PUT_NUMBER4 (0);    //  Empty string            break;    }    //  Now send the data frame    zmq_msg_send (&frame, zsock_resolve (output), --nbr_frames? ZMQ_SNDMORE: 0);    return 0;}

int main(){    void *context = zmq_ctx_new();    void *responder = zmq_socket(context, ZMQ_REP);    zmq_bind(responder, "tcp://*:5555");    while(1) {        zmq_msg_t request;        zmq_msg_init(&request);        zmq_msg_recv(&request, responder, 0);        char *recvmsg ;        int recvsize = zmq_msg_size(&request);        //memset(recvmsg, 0, recvsize);        memcpy(recvmsg, (char *)zmq_msg_data(&request), recvsize);        recvmsg[recvsize] = '/0';        printf("Received request :[%s]/n", recvmsg);        zmq_msg_close(&request);        sleep(1);        zmq_msg_t reply;        //zmq_msg_init_size(&reply, 5);        //memcpy((void*)zmq_msg_data(&reply), "World", 5);        zmq_msg_init_data(&reply, "World", 5, NULL, NULL);        zmq_msg_send(&reply, responder, 0);        zmq_msg_close(&reply);    }    zmq_close(responder);    zmq_ctx_destroy(context);    return 0;}

DWORD WINAPI MyThreadFunction( LPVOID lpParam ){	ThreadData *pData = (ThreadData*)lpParam ;   void *requester = zmq_socket (context, ZMQ_REQ);     zmq_connect (requester, pData->connect_url); // "tcp://localhost:5555"	 pData->thread_running = true;	 while (pData->thread_running)	 {		 if (!MyQueue.empty()) {			 std::string command = MyQueue.front();			 MyQueue.pop();			 		  zmq_msg_t request;		  zmq_msg_init_size (&request, command.length());		  memcpy (zmq_msg_data (&request), command.c_str() , command.length());          zmq_msg_send (&request, requester, 0);          zmq_msg_close (&request);         zmq_msg_t reply;           zmq_msg_init (&reply);          zmq_msg_recv (&reply, requester, 0);//          printf ("Received World %d/n", request_nbr);          zmq_msg_close (&reply);		 } else {			 // Be nice and sleep			 Sleep(1);		 }	 }	zmq_close (requester);	 return 0; }

int main(int argc, char* argv[]){  int i;  void* ctx = zmq_ctx_new();  void* s = zmq_socket(ctx, ZMQ_REQ);  zmq_connect(s, "tcp://localhost:5555");  for (i = 0; i < 10; ++i) {    zmq_msg_t req, rep;    zmq_msg_init_size(&req, 128);    sprintf((char*)zmq_msg_data(&req),       "Hello, world! current count[%d]", i);    zmq_msg_send(&req, s, 0);    zmq_msg_close(&req);    zmq_msg_init(&rep);    zmq_msg_recv(&rep, s, 0);    fprintf(stdout, "reply message : %s/n", (char*)zmq_msg_data(&rep));    zmq_msg_close(&rep);    __sleep(100);  }  zmq_close(s);  zmq_ctx_destroy(ctx);  return 0;}

static int luazmq_msg_send(lua_State *L){  zmessage *zmsg = luazmq_getmessage(L);  zsocket  *zskt = luazmq_getsocket_at(L, 2);  int flags = luaL_optint(L, 3, 0);  int err = zmq_msg_send(&zmsg->msg, zskt->skt, flags);  if(-1 == err)return luazmq_fail(L, zskt);  return luazmq_pass(L);}

static int luazmq_skt_send_msg (lua_State *L) {  zsocket *skt  = luazmq_getsocket(L);  zmessage *msg = luazmq_getmessage_at(L,2);  int flags = luaL_optint(L,3,0);  int ret = zmq_msg_send(&msg->msg, skt->skt, flags);  if(-1 == ret) return luazmq_fail(L, skt);  return luazmq_pass(L);}

// insert employeeInsertResponse *test_insert__insert(void *socket, const uint64_t id, const char *name, const Employee__Gender gender, const time_t date_of_birth, const double bonus, const double salary, const double sales){    assert(socket);    assert(name);    // fill protobuf message    char emp_name[EMPLOYEE_NAME_SIZE] = { 0 };    Employee employee;    employee__init(&employee);    employee.name = emp_name;    InsertRequest insert_request;    insert_request__init(&insert_request);    insert_request.employee = &employee;    insert_request.employee->id = id;    strcpy(insert_request.employee->name, name);    insert_request.employee->gender = gender;    insert_request.employee->date_of_birth = date_of_birth;    insert_request.employee->last_10_annual_bonus = bonus;    insert_request.employee->last_10_annual_salary = salary;    insert_request.employee->last_10_annual_sales = sales;    // get protobuf message data    size_t data_req_size = insert_request__get_packed_size(&insert_request) + 1;    unsigned char *data_req = (unsigned char *)malloc(data_req_size);    assert(data_req);    *data_req = MSG_INSERT_REQUEST;    // write protobuf message data    insert_request__pack(&insert_request, data_req + 1);    // set zmq message data    zmq_msg_t msg_req, msg_res;    assert(zmq_msg_init_size(&msg_req, data_req_size) == 0);    memcpy(zmq_msg_data(&msg_req), data_req, data_req_size);    free(data_req);    // send    int msg_req_size = zmq_msg_size(&msg_req);    assert(zmq_msg_send(&msg_req, socket, 0) == msg_req_size);    // receive    assert(zmq_msg_init(&msg_res) == 0);    int received = zmq_msg_recv(&msg_res, socket, 0);    assert(received > 0);    // get response data    unsigned char *data_res = zmq_msg_data(&msg_res);    size_t data_res_size = zmq_msg_size(&msg_res);    assert(data_res);    assert(data_res[0] == MSG_INSERT_RESPONSE);    assert(data_res_size > 0);    // unpack protobuf message    return (insert_response__unpack(NULL, data_res_size - 1, data_res + 1));}

 inline bool send (message_t &msg_, int flags_ = 0) {     int nbytes = zmq_msg_send (&(msg_.msg), ptr, flags_);     if (nbytes >= 0)         return true;     if (zmq_errno () == EAGAIN)         return false;     throw error_t (); }

void parse_notifications(char *buff, ssize_t len, void* worker){		zmq_msg_t message;	zmq_msg_init_size (&message, len);	memcpy (zmq_msg_data (&message), buff, len);	int rc = zmq_msg_send (&message, worker, 0);	assert(rc != -1);	zmq_msg_close (&message);}

static int safe_send_to_proxy(void *socket, char *string, size_t len) {	zmq_msg_t message;	zmq_msg_init_size (&message, len);	memcpy (zmq_msg_data (&message), string, len);	int rc = zmq_msg_send (&message, socket, 0);	assert(rc != -1);	zmq_msg_close (&message);	return (rc);}

void HardwareServer::sendImage(HardwareImage &image){    std::string data = image.encode();    zmq_msg_t msg;    zmq_msg_init_size(&msg, data.size());    memcpy(zmq_msg_data(&msg), data.c_str(), image.data.size());    zmq_msg_send(&msg, imagesServer, 0);    zmq_msg_close(&msg);}