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

// Send multiple messages.// TODO: this function has no man page//// If flag bit ZMQ_SNDMORE is set the vector is treated as// a single multi-part message, i.e. the last message has// ZMQ_SNDMORE bit switched off.//int zmq_sendiov (void *s_, iovec *a_, size_t count_, int flags_){    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {        errno = ENOTSOCK;        return -1;    }    if (unlikely (count_ <= 0 || !a_)) {        errno = EINVAL;        return -1;    }    int rc = 0;    zmq_msg_t msg;    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;    for (size_t i = 0; i < count_; ++i) {        rc = zmq_msg_init_size (&msg, a_[i].iov_len);        if (rc != 0) {            rc = -1;            break;        }        memcpy (zmq_msg_data (&msg), a_[i].iov_base, a_[i].iov_len);        if (i == count_ - 1)            flags_ = flags_ & ~ZMQ_SNDMORE;        rc = s_sendmsg (s, &msg, flags_);        if (unlikely (rc < 0)) {           int err = errno;           int rc2 = zmq_msg_close (&msg);           errno_assert (rc2 == 0);           errno = err;           rc = -1;           break;        }    }    return rc;}

int main (int argc, char *argv []){    const char *connect_to;    int message_count;    int message_size;    void *ctx;    void *s;    int rc;    int i;    zmq_msg_t msg;    if (argc != 4) {        printf ("usage: remote_thr <connect-to> <message-size> "            "<message-count>/n");        return 1;    }    connect_to = argv [1];    message_size = atoi (argv [2]);    message_count = atoi (argv [3]);    ctx = zmq_init (1);    if (!ctx) {        printf ("error in zmq_init: %s/n", zmq_strerror (errno));        return -1;    }    s = zmq_socket (ctx, ZMQ_PUB);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        return -1;    }    //  Add your socket options here.    //  For example ZMQ_RATE, ZMQ_RECOVERY_IVL and ZMQ_MCAST_LOOP for PGM.    rc = zmq_connect (s, connect_to);    if (rc != 0) {        printf ("error in zmq_connect: %s/n", zmq_strerror (errno));        return -1;    }    for (i = 0; i != message_count; i++) {        rc = zmq_msg_init_size (&msg, message_size);        if (rc != 0) {            printf ("error in zmq_msg_init_size: %s/n", zmq_strerror (errno));            return -1;        }#if defined ZMQ_MAKE_VALGRIND_HAPPY        memset (zmq_msg_data (&msg), 0, message_size);#endif        rc = zmq_sendmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_sendmsg: %s/n", zmq_strerror (errno));            return -1;        }        rc = zmq_msg_close (&msg);        if (rc != 0) {            printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));            return -1;        }    }    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_term (ctx);    if (rc != 0) {        printf ("error in zmq_term: %s/n", zmq_strerror (errno));        return -1;    }    return 0;}

void test_client_server (){    void *ctx = zmq_ctx_new ();    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_SERVER);    assert (sb);    pre_allocate_sock(sb, "127.0.0.1", "5560");    int rc = zmq_bind (sb, "tcp://127.0.0.1:5560");    assert (rc == 0);    void *sc = zmq_socket (ctx, ZMQ_CLIENT);    assert (sc);    rc = zmq_connect (sc, "tcp://127.0.0.1:5560");    assert (rc == 0);    zmq_msg_t msg;    rc = zmq_msg_init_size (&msg, 1);    assert (rc == 0);    char *data = (char *) zmq_msg_data (&msg);    data [0] = 1;    rc = zmq_msg_send (&msg, sc, ZMQ_SNDMORE);    assert (rc == -1);    rc = zmq_msg_send (&msg, sc, 0);    assert (rc == 1);    rc = zmq_msg_init (&msg);    assert (rc == 0);    rc = zmq_msg_recv (&msg, sb, 0);    assert (rc == 1);    uint32_t routing_id = zmq_msg_routing_id (&msg);    assert (routing_id != 0);    rc = zmq_msg_close (&msg);    assert (rc == 0);    rc = zmq_msg_init_size (&msg, 1);    assert (rc == 0);    data = (char *)zmq_msg_data (&msg);    data[0] = 2;    rc = zmq_msg_set_routing_id (&msg, routing_id);    assert (rc == 0);    rc = zmq_msg_send (&msg, sb, ZMQ_SNDMORE);    assert (rc == -1);    rc = zmq_msg_send (&msg, sb, 0);    assert (rc == 1);    rc = zmq_msg_recv (&msg, sc, 0);    assert (rc == 1);    routing_id = zmq_msg_routing_id (&msg);    assert (routing_id == 0);    rc = zmq_msg_close (&msg);    assert (rc == 0);    rc = zmq_close (sc);    assert (rc == 0);    rc = zmq_close (sb);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);}

int main (void){    setup_test_environment();    void *ctx = zmq_ctx_new ();    assert (ctx);    void *server = zmq_socket (ctx, ZMQ_SERVER);    void *client = zmq_socket (ctx, ZMQ_CLIENT);    int rc;    rc = zmq_bind (server, "tcp://127.0.0.1:5560");    assert (rc == 0);    rc = zmq_connect (client, "tcp://127.0.0.1:5560");    assert (rc == 0);    zmq_msg_t msg;    rc = zmq_msg_init_size(&msg,1);    assert (rc == 0);    char * data = (char *)zmq_msg_data(&msg);    data[0] = 1;    rc = zmq_msg_send(&msg, client, 0);    assert (rc == 1);    rc = zmq_msg_recv(&msg, server, 0);    assert (rc == 1);        uint32_t routing_id = zmq_msg_get_routing_id(&msg);    assert(routing_id != 0);    rc = zmq_msg_close(&msg);    assert (rc == 0);    rc = zmq_msg_init_size (&msg, 1);    assert (rc == 0);        data = (char *)zmq_msg_data(&msg);    data[0] = 2;    rc = zmq_msg_set_routing_id(&msg, routing_id);    assert (rc == 0);        rc = zmq_msg_send(&msg, server, 0);    assert (rc == 1);    rc = zmq_msg_recv(&msg, client, 0);    assert (rc == 1);    rc = zmq_close (server);    assert (rc == 0);    rc = zmq_close (client);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0 ;}

static voidlog_send(struct evkeyvalq *output_headers, struct evbuffer *res_buf,         thd_data *thd, struct evkeyvalq *get_args){  uint64_t millisec;  int threshold, limit;  const char *callback, *types, *query, *client_id, *target_name,             *learn_target_name;  parse_keyval(get_args, &query, &types, &client_id, &target_name,               &learn_target_name, &callback, &millisec, &threshold, &limit);  /* send data to learn client */  if (thd->zmq_sock && millisec && client_id && query && learn_target_name) {    char c;    size_t l;    msgpack_packer pk;    msgpack_sbuffer sbuf;    int cnt, submit_flag = 0;    msgpack_sbuffer_init(&sbuf);    msgpack_packer_init(&pk, &sbuf, msgpack_sbuffer_write);    cnt = 4;    if (types && !strcmp(types, "submit")) {      cnt++;      types = NULL;      submit_flag = 1;    }    msgpack_pack_map(&pk, cnt);    c = 'i';    msgpack_pack_raw(&pk, 1);    msgpack_pack_raw_body(&pk, &c, 1);    l = strlen(client_id);    msgpack_pack_raw(&pk, l);    msgpack_pack_raw_body(&pk, client_id, l);    c = 'q';    msgpack_pack_raw(&pk, 1);    msgpack_pack_raw_body(&pk, &c, 1);    l = strlen(query);    msgpack_pack_raw(&pk, l);    msgpack_pack_raw_body(&pk, query, l);    c = 's';    msgpack_pack_raw(&pk, 1);    msgpack_pack_raw_body(&pk, &c, 1);    msgpack_pack_uint64(&pk, millisec);    c = 'l';    msgpack_pack_raw(&pk, 1);    msgpack_pack_raw_body(&pk, &c, 1);    l = strlen(learn_target_name);    msgpack_pack_raw(&pk, l);    msgpack_pack_raw_body(&pk, learn_target_name, l);    if (submit_flag) {      c = 't';      msgpack_pack_raw(&pk, 1);      msgpack_pack_raw_body(&pk, &c, 1);      msgpack_pack_true(&pk);    }    {      zmq_msg_t msg;      if (!zmq_msg_init_size(&msg, sbuf.size)) {        memcpy((void *)zmq_msg_data(&msg), sbuf.data, sbuf.size);        if (zmq_send(thd->zmq_sock, &msg, 0)) {          print_error("zmq_send() error");        }        zmq_msg_close(&msg);      }    }    msgpack_sbuffer_destroy(&sbuf);  }  /* make result */  {    int content_length;    if (callback) {      evhttp_add_header(output_headers,                        "Content-Type", "text/javascript; charset=UTF-8");      content_length = strlen(callback);      evbuffer_add(res_buf, callback, content_length);      evbuffer_add(res_buf, "(", 1);      content_length += suggest_result(res_buf, types, query, target_name,                                       threshold, limit,                                       &(thd->cmd_buf), thd->ctx) + 3;      evbuffer_add(res_buf, ");", 2);    } else {      evhttp_add_header(output_headers,                        "Content-Type", "application/json; charset=UTF-8");      content_length = suggest_result(res_buf, types, query, target_name,                                      threshold, limit,                                      &(thd->cmd_buf), thd->ctx);    }    if (content_length >= 0) {      char num_buf[16];      snprintf(num_buf, 16, "%d", content_length);      evhttp_add_header(output_headers, "Content-Length", num_buf);    }//.........这里部分代码省略.........

//.........这里部分代码省略.........        }        /* we set the data format... but we don't have any content to sent? */        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): checking format to log");        if (NULL == clecf->data_lengths) {            ngx_log_error(NGX_LOG_ERR, log, 0, "log_zmq: handler(): no format to log");            continue;        }        /* we set the endpoint... but we don't have any valid endpoint? */        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): checking endpoint to log");        if (NULL == clecf->endpoint_lengths) {            ngx_log_error(NGX_LOG_ERR, log, 0, "log_zmq: handler(): no endpoint to log");            continue;        }        /* process all data variables and write them back to the data values */        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): script data");        if (NULL == ngx_http_script_run(r, &data, clecf->data_lengths->elts, 0, clecf->data_values->elts)) {            ngx_log_error(NGX_LOG_ERR, log, 0, "log_zmq: handler(): error script data");            continue;        }        /* process all endpoint variables and write them back the the endpoint values */        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): script endpoint");        if (NULL == ngx_http_script_run(r, &endpoint, clecf->endpoint_lengths->elts, 0, clecf->endpoint_values->elts)) {            ngx_log_error(NGX_LOG_ERR, log, 0, "log_zmq: handler(): error script endpoint");            continue;        }        /* yes, we must go on */        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): logging to server");        /* no data */        if (0 == data.len) {            ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): no message to log");            continue;        }        /* serialize to the final message format */        serializer = &log_zmq_serialize;        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): serializing message");        if (NGX_ERROR == (*serializer)(pool, &endpoint, &data, &zmq_data)) {            ngx_log_error(NGX_LOG_ERR, log, 0, "log_zmq: handler(): error serializing message");            ngx_pfree(pool, zmq_data.data);            continue;        }        /* no context? we dont create any */        if (NULL == clecf->ctx) {            ngx_log_error(NGX_LOG_ERR, log, 0, "log_zmq: handler(): no context");            continue;        }        clecf->ctx->log = log;        rc = 1; /* we should have a rc = 0 after this call */        /* create zmq context if needed */        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): verify ZMQ context");        if ((NULL == clecf->ctx->zmq_context) && (0 == clecf->ctx->ccreated)) {            ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): creating context");            rc = zmq_create_ctx(clecf);            if (rc != 0) {                ngx_log_error(NGX_LOG_INFO, log, 0, "log_zmq: handler(): error creating context");                continue;            }        }        /* open zmq socket if needed */        ngx_log_debug0(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): verify ZMQ socket");        if (NULL == clecf->ctx->zmq_socket && 0 == clecf->ctx->screated) {            ngx_log_debug0(NGX_LOG_INFO, log, 0, "log_zmq: handler(): creating socket");            rc = zmq_create_socket(pool, clecf);            if (rc != 0) {                ngx_log_error(NGX_LOG_INFO, log, 0, "log_zmq: handler(): error creating socket");                continue;            }        }        /* initialize zmq message */        zmq_msg_init_size(&query, zmq_data.len);        ngx_memcpy(zmq_msg_data(&query), zmq_data.data, zmq_data.len);        if (zmq_msg_send(&query, clecf->ctx->zmq_socket, 0) >= 0) {            ngx_log_debug1(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): message sent: %V", &zmq_data);        } else {            ngx_log_debug1(NGX_LOG_DEBUG_HTTP, log, 0, "log_zmq: handler(): message not sent: %V", &zmq_data);        }        /* free all for the next iteration */        zmq_msg_close(&query);        ngx_pfree(pool, zmq_data.data);    }    return NGX_OK;}

int main (int argc, char *argv []){    const char *connect_to;    int roundtrip_count;    size_t message_size;    void *ctx;    void *s;    int rc;    int i;    zmq_msg_t msg;    void *watch;    unsigned long elapsed;    double latency;    if (argc != 4) {        printf ("usage: remote_lat <connect-to> <message-size> "            "<roundtrip-count>/n");        return 1;    }    connect_to = argv [1];    message_size = atoi (argv [2]);    roundtrip_count = atoi (argv [3]);    ctx = zmq_init (1, 1, 0);    if (!ctx) {        printf ("error in zmq_init: %s/n", zmq_strerror (errno));        return -1;    }    s = zmq_socket (ctx, ZMQ_REQ);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_connect (s, connect_to);    if (rc != 0) {        printf ("error in zmq_connect: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_msg_init_size (&msg, message_size);    if (rc != 0) {        printf ("error in zmq_msg_init_size: %s/n", zmq_strerror (errno));        return -1;    }    memset (zmq_msg_data (&msg), 0, message_size);    watch = zmq_stopwatch_start ();    for (i = 0; i != roundtrip_count; i++) {        rc = zmq_send (s, &msg, 0);        if (rc != 0) {            printf ("error in zmq_send: %s/n", zmq_strerror (errno));            return -1;        }        rc = zmq_recv (s, &msg, 0);        if (rc != 0) {            printf ("error in zmq_recv: %s/n", zmq_strerror (errno));            return -1;        }        if (zmq_msg_size (&msg) != message_size) {            printf ("message of incorrect size received/n");            return -1;        }    }    elapsed = zmq_stopwatch_stop (watch);    rc = zmq_msg_close (&msg);    if (rc != 0) {        printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));        return -1;    }    latency = (double) elapsed / (roundtrip_count * 2);    printf ("message size: %d [B]/n", (int) message_size);    printf ("roundtrip count: %d/n", (int) roundtrip_count);    printf ("average latency: %.3f [us]/n", (double) latency);    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_term (ctx);    if (rc != 0) {        printf ("error in zmq_term: %s/n", zmq_strerror (errno));        return -1;    }    return 0;}

static void publisher_thread_main (void *pvoid){    const proxy_hwm_cfg_t *cfg = (proxy_hwm_cfg_t *) pvoid;    const int idx = cfg->thread_idx;    int optval;    int rc;    void *pubsocket = zmq_socket (cfg->context, ZMQ_XPUB);    assert (pubsocket);    set_hwm (pubsocket);    optval = 1;    TEST_ASSERT_SUCCESS_ERRNO (      zmq_setsockopt (pubsocket, ZMQ_XPUB_NODROP, &optval, sizeof (optval)));    optval = 1;    TEST_ASSERT_SUCCESS_ERRNO (      zmq_setsockopt (pubsocket, ZMQ_SNDTIMEO, &optval, sizeof (optval)));    TEST_ASSERT_SUCCESS_ERRNO (      zmq_connect (pubsocket, cfg->frontend_endpoint[idx]));    //  Wait before starting TX operations till 1 subscriber has subscribed    //  (in this test there's 1 subscriber only)    char buffer[32] = {};    rc = TEST_ASSERT_SUCCESS_ERRNO (      zmq_recv (pubsocket, buffer, sizeof (buffer), 0));    if (rc != 1) {        printf ("invalid response length: expected 1, received %d", rc);        exit (1);    }    if (buffer[0] != 1) {        printf ("invalid response value: expected 1, received %d",                (int) buffer[0]);        exit (1);    }    zmq_msg_t msg_orig;    rc = zmq_msg_init_size (&msg_orig, message_size);    assert (rc == 0);    memset (zmq_msg_data (&msg_orig), 'A', zmq_msg_size (&msg_orig));    uint64_t send_count = 0;    while (send_count < message_count) {        zmq_msg_t msg;        zmq_msg_init (&msg);        rc = zmq_msg_copy (&msg, &msg_orig);        assert (rc == 0);        //  Send the message to the socket        rc = zmq_msg_send (&msg, pubsocket, 0);        if (rc != -1) {            send_count++;        } else {            TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));        }    }    zmq_close (pubsocket);    printf ("publisher thread ended/n");}

ret_tcherokee_handler_zeromq_read_post (cherokee_handler_zeromq_t *hdl){	zmq_msg_t				message;	int						re;	ret_t					ret;	cherokee_buffer_t	   *post = &HANDLER_THREAD(hdl)->tmp_buf1;	cherokee_buffer_t	   *out  = &HANDLER_THREAD(hdl)->tmp_buf2;	cherokee_connection_t  *conn = HANDLER_CONN(hdl);	/* Check for the post info	 */	if (! conn->post.has_info) {		conn->error_code = http_bad_request;		return ret_error;	}	cherokee_buffer_clean (post);	ret = cherokee_post_read (&conn->post, &conn->socket, post);	switch (ret) {	case ret_ok:		cherokee_connection_update_timeout (conn);		break;	case ret_eagain:		ret = cherokee_thread_deactive_to_polling (HANDLER_THREAD(hdl),												   HANDLER_CONN(hdl),												   conn->socket.socket,												   FDPOLL_MODE_READ, false);		if (ret != ret_ok) {			return ret_error;		} else {			return ret_eagain;		}	default:		conn->error_code = http_bad_request;		return ret_error;	}	TRACE (ENTRIES, "Post contains: '%s'/n", post->buf);	re = cherokee_post_read_finished (&conn->post);	ret = re ? ret_ok : ret_eagain;	if (hdl->encoder != NULL) {		cherokee_buffer_clean(out);		if (ret == ret_ok) {			cherokee_encoder_flush(hdl->encoder, post, out);		} else {			cherokee_encoder_encode(hdl->encoder, post, out);		}		post = out;	}		cherokee_buffer_add_buffer(&hdl->output, post);		if (ret == ret_ok) {		cherokee_buffer_t	 			*tmp   = &HANDLER_THREAD(hdl)->tmp_buf1;		cherokee_handler_zeromq_props_t *props = HANDLER_ZEROMQ_PROPS(hdl);		zmq_msg_t envelope;		zmq_msg_t message;		cuint_t len;		if ((cherokee_buffer_is_empty (&conn->web_directory)) ||			(cherokee_buffer_is_ending (&conn->web_directory, '/'))) {			len = conn->web_directory.len;		} else {			len = conn->web_directory.len + 1;		}		cherokee_buffer_clean (tmp);		cherokee_buffer_add   (tmp, conn->request.buf + len,									conn->request.len - len);		TRACE(ENTRIES, "ZeroMQ: incomming path '%s'/n", tmp->buf);		zmq_msg_init_size (&envelope, tmp->len);		memcpy (zmq_msg_data (&envelope), tmp->buf, tmp->len);		zmq_msg_init_size (&message, hdl->output.len);		memcpy (zmq_msg_data (&message), hdl->output.buf, hdl->output.len);		/* Atomic Section */		CHEROKEE_MUTEX_LOCK (&props->mutex);		zmq_msg_send (&envelope, props->socket, ZMQ_DONTWAIT | ZMQ_SNDMORE);		zmq_msg_send (&message, props->socket, ZMQ_DONTWAIT);		CHEROKEE_MUTEX_UNLOCK (&props->mutex);		zmq_msg_close (&envelope);		zmq_msg_close (&message);	}	return ret;}

void *thread_operator(void *attr){    int rc;    int my_id = (int) attr;        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);    void *socket = zmq_socket(server_pool.context, ZMQ_REP);    zmq_connect(socket, ZMQ_INPROC_ADDR);    pthread_cleanup_push((void (*)(void *)) zmq_close, socket);        pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);                zmq_msg_t reply_msgs[DCS_SERVER_REPLY_COUNT];                pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);        for(int i = 0; i < DCS_SERVER_REPLY_COUNT; i++){            zmq_msg_init_size(&reply_msgs[i], DCS_SERVER_REPLY_SIZE);            memcpy(zmq_msg_data(&reply_msgs[i]), &server_replys[i], DCS_SERVER_REPLY_SIZE);        }        pthread_cleanup_push(thread_operator_msg_clean, reply_msgs);            pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);                    if(!server_pool.no_barr){                rc = pthread_barrier_wait(&server_pool.proxy_barr);                if(rc != 0 && rc != PTHREAD_BARRIER_SERIAL_THREAD)                    syslog(LOG_ERR, "Thread #%d cannot wait on barrier.", my_id);            }                        while(1){                int        reply_id = DCS_SERVER_REPLY_OK;                zmq_msg_t  client_msg;                char      *message;                size_t     msg_size;                unsigned char digest[MSG_DIGEST_SIZE];                char      *domain;                char      *md5sum;                char      *sep;                HASH_ELEMENT *comp;                                pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);                zmq_msg_init(&client_msg);                pthread_cleanup_push((void (*)(void *)) zmq_msg_close, &client_msg);                    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);                                        zmq_msg_recv(&client_msg, socket, 0);                    message  = (char *) zmq_msg_data(&client_msg);                    msg_size = zmq_msg_size(&client_msg);                    DEBUGMSG(syslog(LOG_DEBUG, "msg_size = %d/n", msg_size));                    if(msg_size >= MSG_DIGEST_SIZE + 1 + 1 + 2){                        //~ проверка размера сообщения здесь!!!                        //~ decrypt and verify message here!                        //~ message = func(message)                        memset(digest, '/0', MSG_DIGEST_SIZE);                        rc = msg_digest(message + MSG_DIGEST_SIZE, MSG_SALT_PATH,                                        msg_size - MSG_DIGEST_SIZE, digest);                        if(rc){                            DEBUGMSG(syslog(LOG_DEBUG, "msg_digest failed!!!"));                        }                        if(memcmp(message, digest, MSG_DIGEST_SIZE) == 0){                            message += MSG_DIGEST_SIZE;                            DEBUGMSG(syslog(LOG_DEBUG, "Thread #%d catch message: '%s'/n", my_id, message + 1));                                                        switch(*message){                                case DCS_CLIENT_REQ_MD5:                                    message++;                                    sep = strchr(message, MSG_SEPARATOR);                                    if(sep){                                        *sep   = '/0';                                        domain = message;                                        md5sum = sep + 1;                                        /* Проверки на длину md5-сум!!! */                                        comp = Hash_find(&server_pool.hash, domain,                                            (size_t) (unsigned int) sep - (unsigned int) message);                                        if(comp){                                            if(memcmp(md5sum, comp->val, HASH_ELEMENT_VAL_SIZE) != 0){                                                /* Суммы различны, подать сюда полный список деталей! */                                                reply_id = DCS_SERVER_REPLY_FULL;                                                DEBUGMSG(syslog(LOG_DEBUG, "Суммы различны/n"));                                            }                                            else{                                                //~ Суммы совпали, всё хорошо.                                                reply_id = DCS_SERVER_REPLY_OK;                                            }                                        }                                        else{ /* Компьютера в хэше нет. */                                            reply_id = DCS_SERVER_REPLY_FULL;                                            DEBUGMSG(syslog(LOG_DEBUG, "Компьютера в хэше нет/n"));                                        }                                    }                                    break;                                case DCS_CLIENT_REQ_FULL:                                    message++;                                    sep = strchr(message, MSG_SEPARATOR);                                    if(sep){                                        *sep   = '/0';                                        domain = message;                                        size_t domain_size;                                        CL_Detail *details;                                        size_t     details_count;                                        unsigned char *hwdata = (unsigned char *) sep + 1;                                        msg_size -= (MSG_DIGEST_SIZE + 2 + (                                                (size_t) (unsigned int) sep - (unsigned int) message));//.........这里部分代码省略.........

intmdp_client_msg_send (mdp_client_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 MDP_CLIENT_MSG_CLIENT_REQUEST:            frame_size += 1 + strlen ("MDPC02");            frame_size += 1;            //  messageid            frame_size += 1 + strlen (self->service);            break;        case MDP_CLIENT_MSG_CLIENT_PARTIAL:            frame_size += 1 + strlen ("MDPC02");            frame_size += 1;            //  messageid            frame_size += 1 + strlen (self->service);            break;        case MDP_CLIENT_MSG_CLIENT_FINAL:            frame_size += 1 + strlen ("MDPC02");            frame_size += 1;            //  messageid            frame_size += 1 + strlen (self->service);            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 | 4);    PUT_NUMBER1 (self->id);    bool send_body = false;    size_t nbr_frames = 1;              //  Total number of frames to send        switch (self->id) {        case MDP_CLIENT_MSG_CLIENT_REQUEST:            PUT_STRING ("MDPC02");            PUT_NUMBER1 (1);            PUT_STRING (self->service);            nbr_frames += self->body? zmsg_size (self->body): 1;            send_body = true;            break;        case MDP_CLIENT_MSG_CLIENT_PARTIAL:            PUT_STRING ("MDPC02");            PUT_NUMBER1 (2);            PUT_STRING (self->service);            nbr_frames += self->body? zmsg_size (self->body): 1;            send_body = true;            break;        case MDP_CLIENT_MSG_CLIENT_FINAL:            PUT_STRING ("MDPC02");            PUT_NUMBER1 (3);            PUT_STRING (self->service);            nbr_frames += self->body? zmsg_size (self->body): 1;            send_body = true;            break;    }    //  Now send the data frame    zmq_msg_send (&frame, zsock_resolve (output), --nbr_frames? ZMQ_SNDMORE: 0);        //  Now send the body if necessary    if (send_body) {        if (self->body) {            zframe_t *frame = zmsg_first (self->body);            while (frame) {                zframe_send (&frame, output, ZFRAME_REUSE + (--nbr_frames? ZFRAME_MORE: 0));                frame = zmsg_next (self->body);            }        }        else            zmq_send (zsock_resolve (output), NULL, 0, 0);    }    return 0;}

//.........这里部分代码省略.........	return -1;retry:	// retry til EAGAIN;	wsgi_req->do_not_log = 1;	wsgi_req->socket->retry[wsgi_req->async_id] = 1;	return -1;}void uwsgi_proto_zeromq_close(struct wsgi_request *wsgi_req) {	zmq_msg_t reply;	// check for already freed wsgi_req->proto_parser_buf/wsgi_req->proto_parser_pos	if (!wsgi_req->proto_parser_pos)		return;	// no need to pass a free function (the buffer will be freed during cloe_request)	zmq_msg_init_data(&reply, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos, NULL, NULL);	if (uwsgi.threads > 1)		pthread_mutex_lock(&wsgi_req->socket->lock);#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)	if (zmq_sendmsg(wsgi_req->socket->pub, &reply, 0)) {		uwsgi_error("uwsgi_proto_zeromq_close()/zmq_sendmsg()");#else	if (zmq_send(wsgi_req->socket->pub, &reply, 0)) {		uwsgi_error("uwsgi_proto_zeromq_close()/zmq_send()");#endif	}	if (uwsgi.threads > 1)		pthread_mutex_unlock(&wsgi_req->socket->lock);	zmq_msg_close(&reply);}int uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_t len) {	zmq_msg_t reply;	if (zmq_msg_init_size(&reply, wsgi_req->proto_parser_pos + len)) {		uwsgi_error("uwsgi_proto_zeromq_write()/zmq_msg_init_size()");		return -1;	}	char *zmq_body = zmq_msg_data(&reply);	memcpy(zmq_body, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos);	memcpy(zmq_body + wsgi_req->proto_parser_pos, buf, len);	if (uwsgi.threads > 1)		pthread_mutex_lock(&wsgi_req->socket->lock);#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)	if (zmq_sendmsg(wsgi_req->socket->pub, &reply, 0)) {#else	if (zmq_send(wsgi_req->socket->pub, &reply, 0)) {#endif		if (uwsgi.threads > 1)			pthread_mutex_unlock(&wsgi_req->socket->lock);		zmq_msg_close(&reply);		return -1;	}	if (uwsgi.threads > 1)		pthread_mutex_unlock(&wsgi_req->socket->lock);	zmq_msg_close(&reply);	return UWSGI_OK;}/*	we have a problem... recent Mongrel2 releases introduced a ring buffer that limit the amount of messages we can send (or better, the amount of	messages mongrel2 is able to manage). If we send a big static file we can fill that buffer immediately. How to deal with this ? We know that the	message ring can contains a fixed amount of messages. We could try to split the file in chunks (upto the maximum number supported by a specific mongrel2 instance).	This is suboptimal, but there are no better solutions for now. Before you ask: do you really think that sending a single message with a 2GB file is a good approach ?????	By the way, for now, waiting for a better approach, we use a 2MB buffer. Should support flawlessly files up to 32MB without being rejected by mongrel2. For bigger files you	can tune it to higher values (or increase the mongrel2 ring buffer)*/#define UWSGI_MONGREL2_MAX_MSGSIZE 2*1024*1024int uwsgi_proto_zeromq_sendfile(struct wsgi_request *wsgi_req, int fd, size_t pos, size_t len) {	size_t chunk_size = UMIN( len - wsgi_req->write_pos, UWSGI_MONGREL2_MAX_MSGSIZE);	char *tmp_buf = uwsgi_malloc(chunk_size);	ssize_t rlen = read(fd, tmp_buf, chunk_size);	if (rlen <= 0) {		free(tmp_buf);		return -1;	}	wsgi_req->write_pos += rlen;	if (uwsgi_proto_zeromq_write(wsgi_req, tmp_buf, rlen) < 0) {		free(tmp_buf);		return -1;	}	free(tmp_buf);	if (wsgi_req->write_pos == len) {		return UWSGI_OK;	}	return UWSGI_AGAIN;}

ret_tcherokee_handler_tmi_read_post (cherokee_handler_tmi_t *hdl){    zmq_msg_t message;    int					  re;    ret_t					ret;    ret_t					ret_final;    cherokee_buffer_t	   *post = &HANDLER_THREAD(hdl)->tmp_buf1;    cherokee_buffer_t	   *encoded = &HANDLER_THREAD(hdl)->tmp_buf2;    cherokee_connection_t   *conn = HANDLER_CONN(hdl);    /* Check for the post info     */    if (! conn->post.has_info) {        conn->error_code = http_bad_request;        return ret_error;    }    cherokee_buffer_clean (post);    ret = cherokee_post_read (&conn->post, &conn->socket, post);    switch (ret) {    case ret_ok:        cherokee_connection_update_timeout (conn);        break;    case ret_eagain:        ret = cherokee_thread_deactive_to_polling (HANDLER_THREAD(hdl),                HANDLER_CONN(hdl),                conn->socket.socket,                FDPOLL_MODE_READ, false);        if (ret != ret_ok) {            return ret_error;        } else {            return ret_eagain;        }    default:        conn->error_code = http_bad_request;        return ret_error;    }    TRACE (ENTRIES, "Post contains: '%s'/n", post->buf);    re = cherokee_post_read_finished (&conn->post);    ret_final = re ? ret_ok : ret_eagain;    cherokee_buffer_clean(encoded);    if (hdl->encoder != NULL) {        if (ret == ret_ok) {            cherokee_encoder_flush(hdl->encoder, post, encoded);        } else {            cherokee_encoder_encode(hdl->encoder, post, encoded);        }    } else {        encoded = post;    }    cherokee_buffer_add_buffer(&hdl->output, post);    if (ret_final == ret_ok) {        cherokee_buffer_t	 *tmp  = &HANDLER_THREAD(hdl)->tmp_buf1;        cherokee_handler_tmi_props_t *props = HANDLER_TMI_PROPS(hdl);        zmq_msg_t envelope;        zmq_msg_t message;        cuint_t len;        if ((cherokee_buffer_is_empty (&conn->web_directory)) ||                (cherokee_buffer_is_ending (&conn->web_directory, '/')))        {            len = conn->web_directory.len;        } else {            len = conn->web_directory.len + 1;        }        cherokee_buffer_clean (tmp);        cherokee_buffer_add   (tmp, conn->request.buf + len,                               conn->request.len - len);        TRACE(ENTRIES, "ZeroMQ: incomming path '%s'/n", tmp->buf);        zmq_msg_init_size (&envelope, tmp->len);        memcpy (zmq_msg_data (&envelope), tmp->buf, tmp->len);        zmq_msg_init_size (&message, hdl->output.len);        memcpy (zmq_msg_data (&message), hdl->output.buf, hdl->output.len);        /* Atomic Section */        CHEROKEE_MUTEX_LOCK (&props->mutex);        zmq_msg_send (&envelope, props->socket, ZMQ_DONTWAIT | ZMQ_SNDMORE);        zmq_msg_send (&message, props->socket, ZMQ_DONTWAIT);        CHEROKEE_MUTEX_UNLOCK (&props->mutex);        zmq_msg_close (&envelope);        zmq_msg_close (&message);#ifdef LIBXML_PUSH_ENABLED        if (hdl->validate_xml) {            if (hdl->inflated) {                hdl->strm.avail_in = hdl->output.len;                hdl->strm.next_in = hdl->output.buf;                /* run inflate() on input until output buffer not full */                do  {//.........这里部分代码省略.........

BtcNetMsg* SampleEscrowServerZmq::SendData(BtcNetMsg* message){    // Prepare our context and socket    zmq_context_t* context = zmq_init(1);    zmq_socket_t* socket = zmq_socket(context, ZMQ_REQ);    // Configure socket to not wait at close time    int timeOut = 3000;#ifdef OT_USE_ZMQ4    zmq_setsockopt(socket, ZMQ_RCVTIMEO, &timeOut, sizeof(timeOut));    zmq_setsockopt(socket, ZMQ_SNDTIMEO, &timeOut, sizeof(timeOut));#endif    int linger = 0;    zmq_setsockopt(socket, ZMQ_LINGER, &linger, sizeof (linger));    if(zmq_connect(socket, this->connectString.c_str()) != 0)    {        int error = zmq_errno();        zmq_close(socket);        zmq_term(context);        return NULL;    }    size_t size = NetMessageSizes[static_cast<NetMessageType>(message->MessageType)];    zmq_msg_t* request = new zmq_msg_t();    zmq_msg_init_size(request, size);    zmq_msg_init_data(request, message->data, size, NULL, NULL);#ifdef OT_USE_ZMQ4    if(zmq_msg_send(request, socket, 0) == -1)#else#ifndef Q_OS_MAC    if(zmq_send(socket, request, 0) == -1)#endif#endif    {        zmq_close(socket);        zmq_term(context);        return NULL;    }    zmq_msg_t reply;    zmq_msg_init(&reply);    bool incomingData = false;    for(int i = 0; i < 3; i++)    {        // Wait for response from server        zmq_pollitem_t items[] = { { socket, 0, ZMQ_POLLIN, 0 } };        zmq_poll (&items[0], 1, 1000);        // Cancel if there is no response        if ((items[0].revents & ZMQ_POLLIN))        {            incomingData = true;            break;        }        if(master != NULL)        {            master->UpdateServer();        }    }    if(!incomingData)    {        zmq_close(socket);        zmq_term(context);        zmq_msg_close(&reply);        return NULL;    }#ifdef OT_USE_ZMQ4    zmq_msg_recv(&reply, socket, 0);#else    #ifndef Q_OS_MAC    zmq_recv(socket, &reply, 0);#endif#endif    if(zmq_msg_size(&reply) < NetMessageSizes[Unknown])    {        zmq_close(socket);        zmq_term(context);        zmq_msg_close(&reply);        return NULL;    }    NetMessageType messageType = static_cast<NetMessageType>(static_cast<BtcNetMsg*>(zmq_msg_data(&reply))->MessageType);    if(messageType == Unknown || zmq_msg_size(&reply) < NetMessageSizes[messageType])    {        zmq_close(socket);        zmq_term(context);        zmq_msg_close(&reply);        return NULL;    }    char* data = new char[NetMessageSizes[messageType]];//.........这里部分代码省略.........

intzgossip_msg_send (zgossip_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 ZGOSSIP_MSG_HELLO:            frame_size += 1;            //  version            break;        case ZGOSSIP_MSG_PUBLISH:            frame_size += 1;            //  version            frame_size += 1 + strlen (self->key);            frame_size += 4;            if (self->value)                frame_size += strlen (self->value);            frame_size += 4;            //  ttl            break;        case ZGOSSIP_MSG_PING:            frame_size += 1;            //  version            break;        case ZGOSSIP_MSG_PONG:            frame_size += 1;            //  version            break;        case ZGOSSIP_MSG_INVALID:            frame_size += 1;            //  version            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 ZGOSSIP_MSG_HELLO:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_PUBLISH:            PUT_NUMBER1 (1);            PUT_STRING (self->key);            if (self->value) {                PUT_LONGSTR (self->value);            }            else                PUT_NUMBER4 (0);    //  Empty string            PUT_NUMBER4 (self->ttl);            break;        case ZGOSSIP_MSG_PING:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_PONG:            PUT_NUMBER1 (1);            break;        case ZGOSSIP_MSG_INVALID:            PUT_NUMBER1 (1);            break;    }    //  Now send the data frame    zmq_msg_send (&frame, zsock_resolve (output), --nbr_frames? ZMQ_SNDMORE: 0);    return 0;}

//.........这里部分代码省略.........            if(pubKey.empty())                break;            BtcNetMsgPubKey* replyMsg = new BtcNetMsgPubKey();            memcpy(replyMsg->pubKey, pubKey.c_str(), std::min(pubKey.size(), sizeof(replyMsg->pubKey)));            replyPtr = (BtcNetMsg*)replyMsg;            break;        }        case GetBalance:        {            BtcNetMsgGetBalancePtr message = BtcNetMsgGetBalancePtr(new BtcNetMsgGetBalance());            memcpy(message->data, zmq_msg_data(&request), NetMessageSizes[messageType]);            int64_t balance = GetClientBalance(message);            BtcNetMsgBalance* replyMsg = new BtcNetMsgBalance();            replyMsg->balance = balance;            replyPtr = (BtcNetMsg*)replyMsg;            break;        }        case GetTxCount:        {            BtcNetMsgGetTxCountPtr message = BtcNetMsgGetTxCountPtr(new BtcNetMsgGetTxCount());            memcpy(message->data, zmq_msg_data(&request), NetMessageSizes[messageType]);            int32_t txCount = GetClientTransactionCount(message);            BtcNetMsgTxCount* replyMsg = new BtcNetMsgTxCount();            replyMsg->txCount = txCount;            replyPtr = (BtcNetMsg*)replyMsg;            break;        }        case GetTx:        {            BtcNetMsgGetTxPtr message = BtcNetMsgGetTxPtr(new BtcNetMsgGetTx());            memcpy(message->data, zmq_msg_data(&request), NetMessageSizes[messageType]);            SampleEscrowTransactionPtr tx = GetClientTransaction(message);            if(tx == NULL)                break;            BtcNetMsgTx* replyMsg = new BtcNetMsgTx();            memcpy(replyMsg->txId, tx->txId.c_str(), std::min(tx->txId.size(), sizeof(replyMsg->txId)));            memcpy(replyMsg->toAddress, tx->targetAddr.c_str(), std::min(tx->targetAddr.size(), sizeof(replyMsg->toAddress)));            replyMsg->amount = tx->amountToSend;            replyMsg->type = static_cast<int8_t>(tx->type);            replyMsg->status = static_cast<int8_t>(tx->status);            replyPtr = (BtcNetMsg*)replyMsg;            break;        }        case RequestRelease:        {                  BtcNetMsgReqWithdrawPtr message = BtcNetMsgReqWithdrawPtr(new BtcNetMsgReqWithdraw());            memcpy(message->data, zmq_msg_data(&request), NetMessageSizes[messageType]);            bool accepted = RequestEscrowWithdrawal(message);            BtcNetMsgWithdrawReply* replyMsg = new BtcNetMsgWithdrawReply();            replyMsg->accepted = static_cast<int8_t>(accepted);            replyPtr = (BtcNetMsg*)replyMsg;            break;        }        case ReqSignedTx:        {            BtcNetMsgReqSignedTxPtr message = BtcNetMsgReqSignedTxPtr(new BtcNetMsgReqSignedTx());            memcpy(message->data, zmq_msg_data(&request), NetMessageSizes[messageType]);            std::string partiallySignedTx = RequestSignedWithdrawal(message);            if(partiallySignedTx.empty())                break;            BtcNetMsgSignedTx* replyMsg = new BtcNetMsgSignedTx();            memcpy(replyMsg->rawTx, partiallySignedTx.c_str(), std::min(partiallySignedTx.size(), sizeof(replyMsg->rawTx)));            replyPtr = (BtcNetMsg*)replyMsg;            break;        }        default:            std::printf("received malformed message/n");            std::cout.flush();            break;        }        zmq_msg_close(&request);        // Send reply back to client        size_t size = NetMessageSizes[(NetMessageType)replyPtr->MessageType];        zmq_msg_t reply;        zmq_msg_init_size(&reply, size);        zmq_msg_init_data(&reply, replyPtr->data, size, &DeleteNetMsg, replyPtr);#ifdef OT_USE_ZMQ4        zmq_msg_send(&reply, this->serverSocket, 0);#else        #ifndef Q_OS_MAC        zmq_send(this->serverSocket, &reply, 0);#endif#endif        // note: replyPtr is not deleted on purpose, see DeleteNetMsg()    }}

 void message_t::reset(size_t size) {   assert(zmq_msg_close(ptr.get()) == 0);   if(zmq_msg_init_size(ptr.get(), size) != 0)     throw std::runtime_error(zmq_strerror(zmq_errno())); }

 void rebuild(size_t size) {     close();     auto rc = zmq_msg_init_size(&msg_, size);     if (rc)         throw boost::system::system_error(make_error_code()); }

voidciao_zmq_send(char *socket_atom, 	      int indicative_size, ciao_term byte_list, 	      ciao_term option_list) {  char *buff;  size_t size= collect_bytes(indicative_size, byte_list, &buff);  // .. Find the socket ............................................  ciao_zmq_socket_assoc *assoc= find_socket(socket_atom);  if(assoc == NULL) {    report_error(EINVAL, "socket_not_found", socket_atom);    return;  }   // .. Parse options ..............................................  int flags= 0;  if(!ciao_is_variable(option_list)) {    while(ciao_is_list(option_list)) {      ciao_term option_term= ciao_list_head(option_list);      option_list= ciao_list_tail(option_list);      if(ciao_is_atom(option_term)) {	char *atom= (char *)ciao_atom_name(option_term);	ciao_zmq_atom_option *atom_option= find_option(send_options, atom);	if(atom_option!=NULL) {	  flags|= atom_option->value;	} else {	  report_error(EINVAL, "unknown_send_option", socket_atom);	  if(buff != NULL) free(buff);	  return;	}      } else {	report_error(EINVAL, "invalid_send_option", socket_atom);	if(buff != NULL) free(buff);	return;      }    }    if(!ciao_is_empty_list(option_list)) {      report_error(EINVAL, "option_list_error", socket_atom);      if(buff != NULL) free(buff);      return;    }  }  // .. Allocate message ...........................................  zmq_msg_t msg;  if(zmq_msg_init_size(&msg, size)) {    report_error(errno, "message_init_error", socket_atom);    if(buff != NULL) free(buff);    return;  }  // .. Fill up the message data ...................................  if(size>0) {    memcpy((char *)zmq_msg_data(&msg), buff, size);    free(buff);  }  // .. Send message ...............................................  if(zmq_send(assoc->zmq_socket, &msg, flags)) {    report_error(errno, "send_error", socket_atom);  }}

int main(int, char**){    setup_test_environment();    void* context = zmq_ctx_new ();    void* sockets [2];    int rc = 0;    sockets [SERVER] = zmq_socket (context, ZMQ_STREAM);    rc = zmq_bind (sockets [SERVER], "tcp://0.0.0.0:6666");    assert (rc == 0);    sockets [CLIENT] = zmq_socket (context, ZMQ_STREAM);    rc = zmq_connect (sockets [CLIENT], "tcp://localhost:6666");    assert (rc == 0);    // wait for connect notification    // Server: Grab the 1st frame (peer identity).    zmq_msg_t peer_frame;    rc = zmq_msg_init (&peer_frame);    assert (rc == 0);    rc = zmq_msg_recv (&peer_frame, sockets [SERVER], 0);    assert (rc != -1);    assert(zmq_msg_size (&peer_frame) > 0);    assert (has_more (sockets [SERVER]));    // Server: Grab the 2nd frame (actual payload).    zmq_msg_t data_frame;    rc = zmq_msg_init (&data_frame);    assert (rc == 0);    rc = zmq_msg_recv (&data_frame, sockets [SERVER], 0);    assert (rc != -1);    assert(zmq_msg_size (&data_frame) == 0);    // Client: Grab the 1st frame (peer identity).    rc = zmq_msg_init (&peer_frame);    assert (rc == 0);    rc = zmq_msg_recv (&peer_frame, sockets [CLIENT], 0);    assert (rc != -1);    assert(zmq_msg_size (&peer_frame) > 0);    assert (has_more (sockets [CLIENT]));    // Client: Grab the 2nd frame (actual payload).    rc = zmq_msg_init (&data_frame);    assert (rc == 0);    rc = zmq_msg_recv (&data_frame, sockets [CLIENT], 0);    assert (rc != -1);    assert(zmq_msg_size (&data_frame) == 0);    // Send initial message.    char blob_data [256];    size_t blob_size = sizeof(blob_data);    rc = zmq_getsockopt (sockets [CLIENT], ZMQ_IDENTITY, blob_data, &blob_size);    assert (rc != -1);    assert(blob_size > 0);    zmq_msg_t msg;    rc = zmq_msg_init_size (&msg, blob_size);    assert (rc == 0);    memcpy (zmq_msg_data (&msg), blob_data, blob_size);    rc = zmq_msg_send (&msg, sockets [dialog [0].turn], ZMQ_SNDMORE);    assert (rc != -1);    rc = zmq_msg_close (&msg);    assert (rc == 0);    rc = zmq_msg_init_size (&msg, strlen(dialog [0].text));    assert (rc == 0);    memcpy (zmq_msg_data (&msg), dialog [0].text, strlen(dialog [0].text));    rc = zmq_msg_send (&msg, sockets [dialog [0].turn], ZMQ_SNDMORE);    assert (rc != -1);    rc = zmq_msg_close (&msg);    assert (rc == 0);    // TODO: make sure this loop doesn't loop forever if something is wrong    //       with the test (or the implementation).    int step = 0;    while (step < steps) {        // Wait until something happens.        zmq_pollitem_t items [] = {            { sockets [SERVER], 0, ZMQ_POLLIN, 0 },            { sockets [CLIENT], 0, ZMQ_POLLIN, 0 },        };        int rc = zmq_poll (items, 2, 100);        assert (rc >= 0);        // Check for data received by the server.        if (items [SERVER].revents & ZMQ_POLLIN) {            assert (dialog [step].turn == CLIENT);            // Grab the 1st frame (peer identity).            zmq_msg_t peer_frame;            rc = zmq_msg_init (&peer_frame);	    assert (rc == 0);            rc = zmq_msg_recv (&peer_frame, sockets [SERVER], 0);	    assert (rc != -1);	    assert(zmq_msg_size (&peer_frame) > 0);            assert (has_more (sockets [SERVER]));            // Grab the 2nd frame (actual payload).            zmq_msg_t data_frame;            rc = zmq_msg_init (&data_frame);//.........这里部分代码省略.........

int main (int argc, char *argv []){#if defined ZMQ_HAVE_WINDOWS    HANDLE local_thread;#else    pthread_t local_thread;#endif    void *ctx;    void *s;    int rc;    int i;    zmq_msg_t msg;    void *watch;    unsigned long elapsed;    double latency;    if (argc != 3) {        printf ("usage: inproc_lat <message-size> <roundtrip-count>/n");        return 1;    }    message_size = atoi (argv [1]);    roundtrip_count = atoi (argv [2]);    ctx = zmq_init (1);    if (!ctx) {        printf ("error in zmq_init: %s/n", zmq_strerror (errno));        return -1;    }    s = zmq_socket (ctx, ZMQ_REQ);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        return -1;    }    rc = zmq_bind (s, "inproc://lat_test");    if (rc != 0) {        printf ("error in zmq_bind: %s/n", zmq_strerror (errno));        return -1;    }#if defined ZMQ_HAVE_WINDOWS    local_thread = (HANDLE) _beginthreadex (NULL, 0,        worker, ctx, 0 , NULL);    if (local_thread == 0) {        printf ("error in _beginthreadex/n");        return -1;    }#else    rc = pthread_create (&local_thread, NULL, worker, ctx);    if (rc != 0) {        printf ("error in pthread_create: %s/n", zmq_strerror (rc));        return -1;    }#endif    rc = zmq_msg_init_size (&msg, message_size);    if (rc != 0) {        printf ("error in zmq_msg_init_size: %s/n", zmq_strerror (errno));        return -1;    }    memset (zmq_msg_data (&msg), 0, message_size);    printf ("message size: %d [B]/n", (int) message_size);    printf ("roundtrip count: %d/n", (int) roundtrip_count);    watch = zmq_stopwatch_start ();    for (i = 0; i != roundtrip_count; i++) {        rc = zmq_sendmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_sendmsg: %s/n", zmq_strerror (errno));            return -1;        }        rc = zmq_recvmsg (s, &msg, 0);        if (rc < 0) {            printf ("error in zmq_recvmsg: %s/n", zmq_strerror (errno));            return -1;        }        if (zmq_msg_size (&msg) != message_size) {            printf ("message of incorrect size received/n");            return -1;        }    }    elapsed = zmq_stopwatch_stop (watch);    rc = zmq_msg_close (&msg);    if (rc != 0) {        printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));        return -1;    }    latency = (double) elapsed / (roundtrip_count * 2);#if defined ZMQ_HAVE_WINDOWS    DWORD rc2 = WaitForSingleObject (local_thread, INFINITE);    if (rc2 == WAIT_FAILED) {        printf ("error in WaitForSingleObject/n");//.........这里部分代码省略.........

//.........这里部分代码省略.........			json_decref(root);#else			uwsgi_log("JSON support not enabled (recompile uWSGI with libjansson support, or re-configure mongrel2 with /"protocol='tnetstring'/". skip request/n");#endif		}		// pre-build the mongrel2 response_header		wsgi_req->proto_parser_buf_size = req_uuid_len + 1 + 11 + 1 + req_id_len + 1 + 1;		wsgi_req->proto_parser_buf = uwsgi_malloc(wsgi_req->proto_parser_buf_size);		memcpy(wsgi_req->proto_parser_buf, req_uuid, req_uuid_len);		((char *) wsgi_req->proto_parser_buf)[req_uuid_len] = ' ';		resp_id_len = uwsgi_num2str2(req_id_len, wsgi_req->proto_parser_buf + req_uuid_len + 1);		((char *) wsgi_req->proto_parser_buf)[req_uuid_len + 1 + resp_id_len] = ':';		memcpy((char *) wsgi_req->proto_parser_buf + req_uuid_len + 1 + resp_id_len + 1, req_id, req_id_len);		memcpy((char *) wsgi_req->proto_parser_buf + req_uuid_len + 1 + resp_id_len + 1 + req_id_len, ", ", 2);		wsgi_req->proto_parser_pos = (uint64_t) req_uuid_len + 1 + resp_id_len + 1 + req_id_len + 1 + 1;		// handle post data (in memory)		// reallocate wsgi_req->proto_parser_buf and change its size to be able to store request body		// the parser status holds the current position for read_body hook		if (wsgi_req->post_cl > 0 && !wsgi_req->post_file) {			if (uwsgi_netstring(post_data, message_size - (post_data - message_ptr), &message_ptr, &wsgi_req->post_cl)) {				char *tmp = realloc(wsgi_req->proto_parser_buf, wsgi_req->proto_parser_buf_size + wsgi_req->post_cl);				if (!tmp) {					uwsgi_error("realloc()");					exit(1);					}				wsgi_req->proto_parser_buf = tmp;				// status is an offset...				wsgi_req->proto_parser_status = 0;#ifdef UWSGI_DEBUG				uwsgi_log("post_size: %d/n", wsgi_req->post_cl);#endif				memcpy(wsgi_req->proto_parser_buf + wsgi_req->proto_parser_buf_size, message_ptr, wsgi_req->post_cl);			}		}		zmq_msg_close(&message);		// retry by default		wsgi_req->socket->retry[wsgi_req->async_id] = 1;		return 0;	}repoll:	// force polling of the socket	wsgi_req->socket->retry[wsgi_req->async_id] = 0;	return -1;retry:	// retry til EAGAIN;	wsgi_req->do_not_log = 1;	wsgi_req->socket->retry[wsgi_req->async_id] = 1;	return -1;}static int uwsgi_proto_zeromq_write_do(struct wsgi_request *wsgi_req, char *buf, size_t len) {	zmq_msg_t reply;	if (zmq_msg_init_size(&reply, wsgi_req->proto_parser_pos + len)) {		uwsgi_error("uwsgi_proto_zeromq_write()/zmq_msg_init_size()");		return -1;	}	char *zmq_body = zmq_msg_data(&reply);	memcpy(zmq_body, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos);	if (len > 0)		memcpy(zmq_body + wsgi_req->proto_parser_pos, buf, len);	if (uwsgi.threads > 1)		pthread_mutex_lock(&wsgi_req->socket->lock);#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)	if (zmq_sendmsg(wsgi_req->socket->pub, &reply, 0)) {#else	if (zmq_send(wsgi_req->socket->pub, &reply, 0)) {#endif		if (uwsgi.threads > 1)			pthread_mutex_unlock(&wsgi_req->socket->lock);		zmq_msg_close(&reply);		return -1;	}	if (uwsgi.threads > 1)		pthread_mutex_unlock(&wsgi_req->socket->lock);	zmq_msg_close(&reply);	return UWSGI_OK;}int uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_t len) {	int ret = uwsgi_proto_zeromq_write_do(wsgi_req, buf, len);	if (ret == UWSGI_OK) {		wsgi_req->write_pos += len;	}	return ret;}

int cmd_server(int argc, char *argv[]) {    void *context = zmq_init(1);    assert(context);    void *subscriber = zmq_socket(context, ZMQ_SUB);    assert(subscriber);    int rc;    rc = zmq_connect(subscriber, "tcp://localhost:50000");    assert_rc;    rc = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "", 0);    assert_rc;    void *publisher = zmq_socket(context, ZMQ_PUB);    assert(publisher);    rc = zmq_bind(publisher, "tcp://*:50001");    assert_rc;    timeval tv_start;    gettimeofday_exn(&tv_start);    long long num_msgs = 0;    catch_signals();    while (!interrupted) {        zmq_msg_t in_message;        rc = zmq_msg_init(&in_message);        assert_rc;        rc = zmq_recv(subscriber, &in_message, 0);        if (0 != rc)            break;        int size = zmq_msg_size(&in_message);        zmq_msg_t out_message;        rc = zmq_msg_init_size(&out_message, size);        assert_rc;        memcpy(zmq_msg_data(&out_message), zmq_msg_data(&in_message),               size);        rc = zmq_send(publisher, &out_message, 0);        assert_rc;        ignore(zmq_msg_close(&out_message));        ignore(zmq_msg_close(&in_message));        ++num_msgs;        timeval tv_interval_end;        gettimeofday_exn(&tv_interval_end);        if (num_msgs % 100000 == 0) {            printf("/rMsgs: %lld; Freq: %lld msgs/sec;  ", num_msgs,                   ((num_msgs * 1000000) / diff_us(&tv_interval_end,                                                   &tv_start)));            fflush(stdout);        }    }    printf("/nExited cleanly/n");    ignore(zmq_close(subscriber));    ignore(zmq_close(publisher));    ignore(zmq_term(context));    return 0;}

ZMQ_EXPORT int WINAPI mql4zmq_msg_init_size (zmq_msg_t *msg, size_t size){	return zmq_msg_init_size(msg, size);}

int main (void){    setup_test_environment ();    //  Create the infrastructure    void *ctx = zmq_ctx_new ();    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_ROUTER);    assert (sb);    int rc = zmq_bind (sb, "inproc://a");    assert (rc == 0);    void *sc = zmq_socket (ctx, ZMQ_DEALER);    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);    //  Routing id comes first.    zmq_msg_t msg;    rc = zmq_msg_init (&msg);    assert (rc == 0);    rc = zmq_msg_recv (&msg, sb, 0);    assert (rc >= 0);    int more = zmq_msg_more (&msg);    assert (more == 1);    //  Then the first part of the message body.    rc = zmq_msg_recv (&msg, sb, 0);    assert (rc == 1);    more = zmq_msg_more (&msg);    assert (more == 1);    //  And finally, the second part of the message body.    rc = zmq_msg_recv (&msg, sb, 0);    assert (rc == 1);    more = zmq_msg_more (&msg);    assert (more == 0);    // Test ZMQ_SHARED property (case 1, refcounted messages)    zmq_msg_t msg_a;    rc = zmq_msg_init_size (&msg_a, 1024); // large enough to be a type_lmsg    assert (rc == 0);    // Message is not shared    rc = zmq_msg_get (&msg_a, ZMQ_SHARED);    assert (rc == 0);    zmq_msg_t msg_b;    rc = zmq_msg_init (&msg_b);    assert (rc == 0);    rc = zmq_msg_copy (&msg_b, &msg_a);    assert (rc == 0);    // Message is now shared    rc = zmq_msg_get (&msg_b, ZMQ_SHARED);    assert (rc == 1);    // cleanup    rc = zmq_msg_close (&msg_a);    assert (rc == 0);    rc = zmq_msg_close (&msg_b);    assert (rc == 0);    // Test ZMQ_SHARED property (case 2, constant data messages)    rc = zmq_msg_init_data (&msg_a, (void *) "TEST", 5, 0, 0);    assert (rc == 0);    // Message reports as shared    rc = zmq_msg_get (&msg_a, ZMQ_SHARED);    assert (rc == 1);    // cleanup    rc = zmq_msg_close (&msg_a);    assert (rc == 0);    //  Deallocate the infrastructure.    rc = zmq_close (sc);    assert (rc == 0);    rc = zmq_close (sb);    assert (rc == 0);    rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0;}

int main (void){    zmq_msg_t request, reply;    char *query_string, *datos_jugador, **datos_juego, *datos;    int *jugador;    query_string  = NULL;    datos_jugador = NULL;    datos_juego   = NULL;    jugador       = NULL;    datos         = NULL;    query_string  = (char *) malloc(10 * sizeof(char));    datos_jugador = (char *) malloc(10 * sizeof(char));    datos_juego   = (char **) malloc(10 * sizeof(char*));    datos         = (char *) malloc(10 * sizeof(char));    jugador       = (int *) malloc(sizeof(int));    printf ("Estableciendo conexion con el servidor.../n");    void *context = zmq_ctx_new ();    void *requester = zmq_socket (context, ZMQ_REQ);    zmq_connect (requester, "tcp://localhost:5529");    int request_nbr;        while(1){               if(!strlen(query_string)){            datos_jugador = "%";        }else{                        if(strlen(query_string)){                                datos_juego[0] = strtok(query_string,"|");                datos_juego[1] = strtok (NULL,"|");                            datos_juego[2] = strtok (NULL,"|");                 *jugador     = atoi(datos_juego[0]);                strcpy(datos,"");                triqui(jugador, datos_juego[1]);                                                strcpy(query_string,datos_juego[0]);                strcpy(datos,datos_juego[1]);                                strcat(query_string,"|");                strcat(query_string,datos);                strcpy(datos,datos_juego[2]);                                                strcat(query_string,"|");                strcat(query_string,datos);                                datos_jugador = query_string;                            }                    }        //Datos enviados del cliente al servidor                zmq_msg_init_size(&request, 23);        memcpy(zmq_msg_data(&request),datos_jugador, 23);                zmq_msg_send(&request,requester,0);        zmq_msg_close(&request);                //Datos recividos por el cliente provenientes del servidor        zmq_msg_init(&reply);        zmq_msg_recv(&reply,requester,0);                query_string = (char *)zmq_msg_data(&reply);                        zmq_msg_close(&reply);    }    sleep(2);    zmq_close (requester);    zmq_ctx_destroy (context);    return 0;}

/** * Called by Java's Socket::send(byte [] msg, int flags). */JNIEXPORT jboolean JNICALL Java_org_zeromq_ZMQ_00024Socket_send (JNIEnv *env,                                                                 jobject obj,                                                                 jbyteArray msg,                                                                 jint flags){    void *s = get_socket (env, obj, 1);    jsize size = env->GetArrayLength (msg);     zmq_msg_t message;    int rc = zmq_msg_init_size (&message, size);    int err = zmq_errno();    if (rc != 0) {        raise_exception (env, err);        return JNI_FALSE;    }    jbyte *data = env->GetByteArrayElements (msg, 0);    if (! data) {        raise_exception (env, EINVAL);        return JNI_FALSE;    }    memcpy (zmq_msg_data (&message), data, size);    env->ReleaseByteArrayElements (msg, data, 0);#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)    rc = zmq_sendmsg (s, &message, flags);#else    rc = zmq_send (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 JNI_FALSE;        }        return JNI_FALSE;    }        if (rc < 0) {        raise_exception (env, err);        rc = zmq_msg_close (&message);        err = zmq_errno();        if (rc != 0) {            raise_exception (env, err);            return JNI_FALSE;        }        return JNI_FALSE;    }    rc = zmq_msg_close (&message);    err = zmq_errno();    if (rc != 0) {        raise_exception (env, err);        return JNI_FALSE;    }    return JNI_TRUE;}

 inline explicit message_t (size_t size_) {     int rc = zmq_msg_init_size (&msg, size_);     if (rc != 0)         throw error_t (); }

 explicit frame(std::size_t size) : raw_msg_() {     if (0 != zmq_msg_init_size(&raw_msg_, size)) throw exception(); }