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

//  Send a monitor eventvoid zmq::socket_base_t::monitor_event (int event_, int value_, const std::string &addr_){    if (monitor_socket) {        //  Send event in first frame        zmq_msg_t msg;        zmq_msg_init_size (&msg, 6);#ifdef ZMQ_HAVE_HPUX        // avoid SIGBUS        union {          uint8_t data[6];          struct {            uint16_t event;            uint32_t value;          } v;        } u;        u.v.event = event_;        u.v.value = value_;        memcpy(zmq_msg_data (&msg), u.data, 6);#else        uint8_t *data = (uint8_t *) zmq_msg_data (&msg);        *(uint16_t *) (data + 0) = (uint16_t) event_;        *(uint32_t *) (data + 2) = (uint32_t) value_;#endif        zmq_sendmsg (monitor_socket, &msg, ZMQ_SNDMORE);        //  Send address in second frame        zmq_msg_init_size (&msg, addr_.size());        memcpy (zmq_msg_data (&msg), addr_.c_str (), addr_.size ());        zmq_sendmsg (monitor_socket, &msg, 0);    }}

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

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

JNIEXPORT jboolean JNICALLJava_org_zeromq_ZMQ_00024Socket_sendZeroCopy (JNIEnv *env,                                              jobject obj,                                              jobject buffer,                                              jint length,                                              jint flags){#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)    int rc = 0;    void *sock = get_socket (env, obj);    // init the message    zmq_msg_t message;    jboolean retval = s_zerocopy_init (env, &message, buffer, length);    if (retval == JNI_FALSE)        return JNI_FALSE;    rc = zmq_sendmsg (sock, &message, flags);    if (rc == -1) {        int err = zmq_errno();        zmq_msg_close (&message);        raise_exception (env, err);        return JNI_FALSE;    }    return JNI_TRUE;#else    return JNI_FALSE;#endif}

//  Send a monitor eventvoid zmq::socket_base_t::monitor_event (int event_, int value_, const std::string &addr_){    if (monitor_socket) {        //  Send event in first frame        zmq_msg_t msg;        zmq_msg_init_size (&msg, 6);        uint8_t *data = (uint8_t *) zmq_msg_data (&msg);        *(uint16_t *) (data + 0) = (uint16_t) event_;        *(uint32_t *) (data + 2) = (uint32_t) value_;        zmq_sendmsg (monitor_socket, &msg, ZMQ_SNDMORE);        //  Send address in second frame        zmq_msg_init_size (&msg, addr_.size());        memcpy (zmq_msg_data (&msg), addr_.c_str (), addr_.size ());        zmq_sendmsg (monitor_socket, &msg, 0);    }}

intzsocket_signal (void *zocket){    zmq_msg_t msg;    zmq_msg_init_size (&msg, 0);    int rc = zmq_sendmsg (zocket, &msg, 0);    return rc == -1? -1: 0;}

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

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

int main(int argc, char *argv[]) {  void *sp=NULL;  void *set=NULL;  int opt,rc=-1;  char *config_file;  set = kv_set_new();  utarray_new(output_keys, &ut_str_icd);  utarray_new(output_defaults, &ut_str_icd);  utarray_new(output_types,&ut_int_icd);  utstring_new(tmp);  while ( (opt = getopt(argc, argv, "v+d:b:s")) != -1) {    switch (opt) {      case 'v': verbose++; break;      case 's': push_mode++; break;      case 'd': spool=strdup(optarg); break;      case 'b': config_file=strdup(optarg); break;      default: usage(argv[0]); break;    }  }  if (optind < argc) pub_transport = argv[optind++];  if (!pub_transport) usage(argv[0]);  if (spool == NULL) usage(argv[0]);  if (parse_config(config_file) < 0) goto done;  if ( !(pub_context = zmq_init(1))) goto done;  if ( !(pub_socket = zmq_socket(pub_context, push_mode?ZMQ_PUSH:ZMQ_PUB))) goto done;  if (zmq_setsockopt(pub_socket, ZMQ_SNDHWM, &hwm, sizeof(hwm))) goto done;  if (zmq_bind(pub_socket, pub_transport) == -1) goto done;  sp = kv_spoolreader_new(spool);  if (!sp) goto done;  while (kv_spool_read(sp,set,1) > 0) { /* read til interrupted by signal */    zmq_msg_t part;    if (set_to_binary(set,&part) < 0) goto done;    rc = zmq_sendmsg(pub_socket, &part, 0);    zmq_msg_close(&part);    if (rc == -1) goto done;  }  rc = 0; done:  if (rc) fprintf(stderr,"zmq: %s %s/n", pub_transport, zmq_strerror(errno));  if (pub_socket) zmq_close(pub_socket);  if (pub_context) zmq_term(pub_context);  if (sp) kv_spoolreader_free(sp);  kv_set_free(set);  utarray_free(output_keys);  utarray_free(output_defaults);  utarray_free(output_types);  utstring_free(tmp);  return 0;}

int main(){	zmq::context_t context(1);	zmq::socket_t client (context, ZMQ_PUSH);	client.setsockopt( ZMQ_IDENTITY, "B", 1);	client.connect("tcp://127.0.0.1:5560");	std::cout << "connected!" << std::endl;	sleep(1);	void *watch;    unsigned long elapsed;    unsigned long throughput;    int message_count = 1;	watch = zmq_stopwatch_start ();	/*for( int i = 0; i < message_count; i++){		//s_sendmore (client, "A");		//s_sendmore(client, "");				s_send (client, "This is the workload");		//s_dump(client);		//std::string string = s_recv (client);		//zmq::message_t message;        //client.recv(&message);	}*/	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;        }    }	elapsed = zmq_stopwatch_stop (watch);	throughput = (unsigned long)        ((double) message_count / (double) elapsed * 1000000);    printf ("mean throughput: %d [msg/s]/n", (int) throughput);	sleep(1);	return 0;}

void send_message(void *z_sock, int flags, Arg &&arg) {  msgpack::sbuffer buf;  msgpack::pack(buf, arg);  zmq_msg_t msg;  zmq_msg_init_size(&msg, buf.size());  memcpy(zmq_msg_data(&msg), buf.data(), buf.size());  zmq_sendmsg(z_sock, &msg, flags);  zmq_msg_close(&msg);}

static void *worker (void *ctx_)#endif{    void *s;    int rc;    int i;    zmq_msg_t msg;    s = zmq_socket (ctx_, ZMQ_PUSH);    if (!s) {        printf ("error in zmq_socket: %s/n", zmq_strerror (errno));        exit (1);    }    rc = zmq_connect (s, "inproc://thr_test");    if (rc != 0) {        printf ("error in zmq_connect: %s/n", zmq_strerror (errno));        exit (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));            exit (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));            exit (1);        }        rc = zmq_msg_close (&msg);        if (rc != 0) {            printf ("error in zmq_msg_close: %s/n", zmq_strerror (errno));            exit (1);        }    }    rc = zmq_close (s);    if (rc != 0) {        printf ("error in zmq_close: %s/n", zmq_strerror (errno));        exit (1);    }#if defined ZMQ_HAVE_WINDOWS    return 0;#else    return NULL;#endif}

//  Send a monitor eventvoid zmq::socket_base_t::monitor_event (int event_, int value_, const std::string &addr_){    if (monitor_socket) {        //  Send event in first frame        zmq_msg_t msg;        zmq_msg_init_size (&msg, 6);        uint8_t *data = (uint8_t *) zmq_msg_data (&msg);        //  Avoid dereferencing uint32_t on unaligned address        uint16_t event = (uint16_t) event_;        uint32_t value = (uint32_t) value_;        memcpy (data + 0, &event, sizeof(event));        memcpy (data + 2, &value, sizeof(value));        zmq_sendmsg (monitor_socket, &msg, ZMQ_SNDMORE);        //  Send address in second frame        zmq_msg_init_size (&msg, addr_.size());        memcpy (zmq_msg_data (&msg), addr_.c_str (), addr_.size ());        zmq_sendmsg (monitor_socket, &msg, 0);    }}

void zmq::socket_base_t::monitor_event (zmq_event_t event_, const std::string& addr_){    if (monitor_socket) {        const uint16_t eid = (uint16_t)event_.event;        const uint32_t value = (uint32_t)event_.value;        // prepare and send first message frame        // containing event id and value        zmq_msg_t msg;        zmq_msg_init_size (&msg, sizeof(eid) + sizeof(value));        char* data1 = (char*)zmq_msg_data(&msg);        memcpy (data1, &eid, sizeof(eid));        memcpy (data1+sizeof(eid), &value, sizeof(value));        zmq_sendmsg (monitor_socket, &msg, ZMQ_SNDMORE);        // prepare and send second message frame        // containing the address (endpoint)        zmq_msg_init_size (&msg, addr_.size());        memcpy(zmq_msg_data(&msg), addr_.c_str(), addr_.size());        zmq_sendmsg (monitor_socket, &msg, 0);    }}

void zmq::socket_base_t::monitor_event (zmq_event_t event_){    if (monitor_socket) {        zmq_msg_t msg;        void *event_data = malloc (sizeof (event_));        alloc_assert (event_data);        memcpy (event_data, &event_, sizeof (event_));        zmq_msg_init_data (&msg, event_data, sizeof (event_), zmq_free_event, NULL);        zmq_sendmsg (monitor_socket, &msg, 0);        zmq_msg_close (&msg);    }}

intzsocket_signal (void *zocket){    zmq_msg_t msg;    zmq_msg_init_size (&msg, 0);    if (zmq_sendmsg (zocket, &msg, 0) == -1) {        zmq_msg_close (&msg);        return -1;    }    else        return 0;}

static ssize_tngx_zeromq_send_part(void *zmq, ngx_event_t *wev, u_char *buf, size_t size,    int flags){    zmq_msg_t  zmq_msg;    if (zmq_msg_init_size(&zmq_msg, size) == -1) {        ngx_log_error(NGX_LOG_ALERT, wev->log, 0,                      "zmq_msg_init_size(%uz) failed (%d: %s)",                      size, ngx_errno, zmq_strerror(ngx_errno));        goto failed_zmq;    }    ngx_memcpy(zmq_msg_data(&zmq_msg), buf, size);    for (;;) {        if (zmq_sendmsg(zmq, &zmq_msg, ZMQ_DONTWAIT|flags) == -1) {            if (ngx_errno == NGX_EINTR) {                ngx_log_debug0(NGX_LOG_DEBUG_EVENT, wev->log, 0,                               "zmq_send: interrupted");                wev->ready = 0;                continue;            }            ngx_zeromq_log_error(wev->log, "zmq_sendmsg()");            goto failed;        }        break;    }    ngx_log_debug2(NGX_LOG_DEBUG_EVENT, wev->log, 0,                   "zmq_send: %uz eom:%d", size, flags != ZMQ_SNDMORE);    if (zmq_msg_close(&zmq_msg) == -1) {        ngx_zeromq_log_error(wev->log, "zmq_msg_close()");        goto failed_zmq;    }    return size;failed:    if (zmq_msg_close(&zmq_msg) == -1) {        ngx_zeromq_log_error(wev->log, "zmq_msg_close()");    }failed_zmq:    wev->error = 1;    return NGX_ERROR;}

static ints_send_string (void *zocket, bool more, char *string){    assert (zocket);    int len = strlen (string);    zmq_msg_t message;    zmq_msg_init_size (&message, len);    memcpy (zmq_msg_data (&message), string, len);    int rc = zmq_sendmsg (zocket, &message, more? ZMQ_SNDMORE: 0);    return rc == -1? -1: 0;}

//  Send a monitor eventvoid zmq::socket_base_t::monitor_event (int event_, intptr_t value_, const std::string &addr_){    // this is a private method which is only called from    // contexts where the mutex has been locked before    if (monitor_socket) {        //  Send event in first frame        zmq_msg_t msg;        zmq_msg_init_size (&msg, 6);        uint8_t *data = (uint8_t *) zmq_msg_data (&msg);        //  Avoid dereferencing uint32_t on unaligned address        uint16_t event = (uint16_t) event_;        uint32_t value = (uint32_t) value_;        memcpy (data + 0, &event, sizeof(event));        memcpy (data + 2, &value, sizeof(value));        zmq_sendmsg (monitor_socket, &msg, ZMQ_SNDMORE);        //  Send address in second frame        zmq_msg_init_size (&msg, addr_.size());        memcpy (zmq_msg_data (&msg), addr_.c_str (), addr_.size ());        zmq_sendmsg (monitor_socket, &msg, 0);    }}

static void router_forward_from_proc(void *dest_sock, zmq_msg_t *msg, int pid) {    zmq_msg_t back_route;    struct router_route *route_data;    zmq_msg_init_size(&back_route, sizeof(struct router_route) + zmq_msg_size(msg));    route_data = zmq_msg_data(&back_route);    route_data->pid = pid;    // TODO: universal router_route size, arch. independent    memcpy(zmq_msg_data(&back_route) + sizeof(struct router_route), zmq_msg_data(msg), zmq_msg_size(msg));    zmq_sendmsg(dest_sock, &back_route, 0);    // Following send is no longer used    //zmq_sendmsg(dest_sock, msg, 0);    zmq_msg_close(&back_route);}

intzsocket_sendmem_zero_copy (void *zocket, void *data, size_t size,                           zsocket_free_fn *free_fn, void *hint, int flags){    assert (zocket);    assert (size == 0 || data);        int snd_flags = (flags & ZFRAME_MORE)? ZMQ_SNDMORE : 0;    snd_flags |= (flags & ZFRAME_DONTWAIT)? ZMQ_DONTWAIT : 0;        zmq_msg_t msg;    zmq_msg_init_data (&msg, data, size, free_fn, hint);    int rc = zmq_sendmsg (zocket, &msg, snd_flags);    return rc == -1? -1: 0;}

intzsocket_sendmem (void *socket, const void *data, size_t size, int flags){    assert (socket);    assert (size == 0 || data);    int snd_flags = (flags & ZFRAME_MORE)? ZMQ_SNDMORE : 0;    snd_flags |= (flags & ZFRAME_DONTWAIT)? ZMQ_DONTWAIT : 0;    zmq_msg_t msg;    zmq_msg_init_size (&msg, size);    memcpy (zmq_msg_data (&msg), data, size);    int rc = zmq_sendmsg (socket, &msg, snd_flags);    return rc == -1? -1: 0;}

static ints_send_string (void *socket, bool more, char *string){    assert (socket);    int len = strlen (string);    zmq_msg_t message;    zmq_msg_init_size (&message, len);    memcpy (zmq_msg_data (&message), string, len);    if (zmq_sendmsg (socket, &message, more? ZMQ_SNDMORE: 0) == -1) {        zmq_msg_close (&message);        return -1;    }    else        return 0;}

static ints_send_string (void *dest, bool more, char *string){    assert (dest);    void *handle = zsock_resolve (dest);    size_t len = strlen (string);    zmq_msg_t message;    zmq_msg_init_size (&message, len);    memcpy (zmq_msg_data (&message), string, len);    if (zmq_sendmsg (handle, &message, more ? ZMQ_SNDMORE : 0) == -1) {        zmq_msg_close (&message);        return -1;    }    else        return 0;}

staticint s_capture_message (void *socket, zmq_msg_t *msg, int more){	int rc;	zmq_msg_t msg_cp;	rc = zmq_msg_init (&msg_cp);	if (rc == -1)		return -1;	rc = zmq_msg_copy (&msg_cp, msg);	if (rc == -1) {		zmq_msg_close (&msg_cp);		return -1;	}	return		zmq_sendmsg (socket, &msg_cp, more ? ZMQ_SNDMORE : 0);}

static int send_msg(void *endpoint, void *msg, int size, int flag){    zmq_msg_t zmsg;    int ret;    if (zmq_msg_init(&zmsg) != 0) {            return -1;    }    zmq_msg_init_size(&zmsg, size);    memcpy((void *)zmq_msg_data(&zmsg), msg, zmq_msg_size(&zmsg));    ret = zmq_sendmsg(endpoint, &zmsg, flag);    zmq_msg_close(&zmsg);    return ret;}

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;	zmq_msg_init_data(&reply, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos, uwsgi_proto_zeromq_free, 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)) {#else	if (zmq_send(wsgi_req->socket->pub, &reply, 0)) {#endif		uwsgi_error("zmq_send()");	}	if (uwsgi.threads > 1) pthread_mutex_unlock(&wsgi_req->socket->lock);	zmq_msg_close(&reply);	if (wsgi_req->async_post && wsgi_req->body_as_file) {		fclose(wsgi_req->async_post);	}}ssize_t uwsgi_proto_zeromq_writev_header(struct wsgi_request *wsgi_req, struct iovec *iovec, size_t iov_len) {	int i;	ssize_t len;	ssize_t ret = 0;	for (i = 0; i < (int) iov_len; i++) {		len = uwsgi_proto_zeromq_write(wsgi_req, iovec[i].iov_base, iovec[i].iov_len);		if (len <= 0) {			wsgi_req->write_errors++;			return 0;		}		ret += len;	}	return ret;}

intzsocket_sendmem (void *self, const void *data, size_t size, int flags){    assert (self);    assert (size == 0 || data);    int snd_flags = (flags & ZFRAME_MORE)? ZMQ_SNDMORE: 0;    snd_flags |= (flags & ZFRAME_DONTWAIT)? ZMQ_DONTWAIT: 0;    zmq_msg_t msg;    zmq_msg_init_size (&msg, size);    memcpy (zmq_msg_data (&msg), data, size);    if (zmq_sendmsg (self, &msg, snd_flags) == -1) {        zmq_msg_close (&msg);        return -1;    }    else        return 0;}

bool ZMQSocket::send(const String& message_param, int64_t flags) {  zmq_msg_t message;  if (zmq_msg_init_size(&message, message_param.size()) != 0) {    throwExceptionClassZMQErr(s_ZMQSocketExceptionClass, "Failed to initialize message structure: {}", errno);  }  memcpy(zmq_msg_data(&message), message_param.data(), message_param.size());  int rc = zmq_sendmsg(socket->z_socket, &message, flags);  int err = errno;  zmq_msg_close(&message);  if (rc == -1) {    if (err == EAGAIN) {      return false;    }    throwExceptionClassZMQErr(s_ZMQSocketExceptionClass, "Failed to send message: {}", err);  }  return true;}

int zmq_send (void *s_, const void *buf_, size_t len_, int flags_){    zmq_msg_t msg;    int rc = zmq_msg_init_size (&msg, len_);    if (rc != 0)        return -1;    memcpy (zmq_msg_data (&msg), buf_, len_);    rc = zmq_sendmsg (s_, &msg, flags_);    if (unlikely (rc < 0)) {        int err = errno;        int rc2 = zmq_msg_close (&msg);        errno_assert (rc2 == 0);        errno = err;        return -1;    }    //  Note the optimisation here. We don't close the msg object as it is    //  empty anyway. This may change when implementation of zmq_msg_t changes.    return rc;}