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

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

int zmq::fq_t::recv (zmq_msg_t *msg_, int flags_){    //  Deallocate old content of the message.    zmq_msg_close (msg_);    //  Round-robin over the pipes to get next message.    for (int count = active; count != 0; count--) {        bool fetched = pipes [current]->read (msg_);        current++;        if (current >= active)            current = 0;        if (fetched)            return 0;    }    //  No message is available. Initialise the output parameter    //  to be a 0-byte message.    zmq_msg_init (msg_);    errno = EAGAIN;    return -1;}

static int pollitem_on_input(void *socket, void *ehub, void *data) {   zmq_msg_t msg;   int r, n;   char *message;   r = zmq_msg_init(&msg);   if (r==-1) return zmq_errno();   n = zmq_msg_recv(&msg, socket, 0);   if (n==-1) return zmq_errno();   message = malloc(n + 1);   if (!message) return ENOMEM;   memcpy(message, zmq_msg_data(&msg), n);   r = zmq_msg_close(&msg);   if (r==-1) return zmq_errno();   ehub_on_input(ehub, socket, message, data);   free(message);}

// Receive a multi-part message// // Receives up to *count_ parts of a multi-part message.// Sets *count_ to the actual number of parts read.// ZMQ_RCVMORE is set to indicate if a complete multi-part message was read.// Returns number of message parts read, or -1 on error.//// Note: even if -1 is returned, some parts of the message// may have been read. Therefore the client must consult// *count_ to retrieve message parts successfully read,// even if -1 is returned.//// The iov_base* buffers of each iovec *a_ filled in by this // function may be freed using free().//// Implementation note: We assume zmq::msg_t buffer allocated// by zmq::recvmsg can be freed by free().// We assume it is safe to steal these buffers by simply// not closing the zmq::msg_t.//int zmq_recviov (void *s_, iovec *a_, size_t *count_, int flags_){    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {        errno = ENOTSOCK;        return -1;    }    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;    size_t count = (int) *count_;    int nread = 0;    bool recvmore = true;    for (size_t i = 0; recvmore && i < count; ++i) {        // Cheat! We never close any msg        // because we want to steal the buffer.        zmq_msg_t msg;        int rc = zmq_msg_init (&msg);        errno_assert (rc == 0);        int nbytes = s_recvmsg (s, &msg, flags_);        if (unlikely (nbytes < 0)) {            int err = errno;            rc = zmq_msg_close (&msg);            errno_assert (rc == 0);            errno = err;            nread = -1;            break;        }        ++*count_;        ++nread;        // Cheat: acquire zmq_msg buffer.        a_[i].iov_base = static_cast<char *> (zmq_msg_data (&msg));        a_[i].iov_len = zmq_msg_size (&msg);        // Assume zmq_socket ZMQ_RVCMORE is properly set.        recvmore = ((zmq::msg_t*) (void *) &msg)->flags () & zmq::msg_t::more;    }    return nread;}

SEXP R_zmq_msg_recv(SEXP R_socket, SEXP R_flags){	SEXP R_rmsg = R_NilValue;	int C_rmsg_length;	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(&msg);		if(C_ret == -1){			C_errno = zmq_errno();			REprintf("R_zmq_msg_init errno: %d strerror: %s/n",				C_errno, zmq_strerror(C_errno));		}		C_ret = zmq_msg_recv(&msg, C_socket, C_flags);		if(C_ret == -1){			C_errno = zmq_errno();			REprintf("R_zmq_msg_recv errno: %d strerror: %s/n",				C_errno, zmq_strerror(C_errno));		}		C_rmsg_length = zmq_msg_size(&msg);		PROTECT(R_rmsg = allocVector(RAWSXP, C_rmsg_length));		memcpy(RAW(R_rmsg), zmq_msg_data(&msg), C_rmsg_length);		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));		}		UNPROTECT(1);		return(R_rmsg);	} else{		REprintf("R_zmq_send: C_socket is not available./n");	}	return(R_rmsg);} /* End of R_zmq_msg_recv(). */

void zmq::session_t::clean_pipes (){    //  Get rid of half-processed messages in the out pipe. Flush any    //  unflushed messages upstream.    if (out_pipe) {        out_pipe->rollback ();        out_pipe->flush ();    }    //  Remove any half-read message from the in pipe.    if (in_pipe) {        while (incomplete_in) {            zmq_msg_t msg;            zmq_msg_init (&msg);            if (!read (&msg)) {                zmq_assert (!incomplete_in);                break;            }            zmq_msg_close (&msg);        }    }}

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

/** * Simple responses are composed of: * - A header frame, with character 4 expected to be 0, else an error is assumed * - If the 4th header character is not 0, a second frame containing an error message shall be received * @param errorString Left unchanged if no error occurs, set to an error description string if any error occurs * @return 0 on success. -1 for communication errors, 1 with errorMessage set in case of error-indicating response */static inline int receiveSimpleResponse(void* socket, std::string& errorString) {    zmq_msg_t msg;    zmq_msg_init(&msg);    int rc = zmq_msg_recv(&msg, socket, 0);    if (rc == -1) {        return -1;    }    //Check if there is any error frame (there *should* be one, if the fourth byte is != 0)    if (((char*) zmq_msg_data(&msg))[3] != 0) {        if (!zmq_sockopt_get_rcvmore(socket)) {            errorString = "No error message received from server -- Exact error cause is unknown";            return -1;        }        //We have an error frame from the server. Return it.        if(receiveStringFrame(socket, errorString) == -1) {            return -1;        }        return 1;    }    zmq_msg_close(&msg);    return 0;}

bool zmq::zmq_encoder_t::message_ready (){    //  Destroy content of the old message.    zmq_msg_close(&in_progress);    //  Read new message from the dispatcher. If there is none, return false.    //  Note that new state is set only if write is successful. That way    //  unsuccessful write will cause retry on the next state machine    //  invocation.    if (!source || !source->read (&in_progress)) {        zmq_msg_init (&in_progress);        return false;    }    //  Get the message size. If the prefix is not to be sent, adjust the    //  size accordingly.    size_t size = zmq_msg_size (&in_progress);    if (trim) {        zmq_assert (size);        size_t prefix_size =            (*(unsigned char*) zmq_msg_data (&in_progress)) + 1;        zmq_assert (prefix_size <= size);        size -= prefix_size;    }    //  For messages less than 255 bytes long, write one byte of message size.    //  For longer messages write 0xff escape character followed by 8-byte    //  message size.    if (size < 255) {        tmpbuf [0] = (unsigned char) size;        next_step (tmpbuf, 1, &zmq_encoder_t::size_ready, true);    }    else {        tmpbuf [0] = 0xff;        put_uint64 (tmpbuf + 1, size);        next_step (tmpbuf, 9, &zmq_encoder_t::size_ready, true);    }    return true;}

static int socket_write_exit(void *socket, int code){	zmq_msg_t msg;	if (zmq_msg_init_size(&msg, 5) == -1) {		TRACE_ERRNO("zmq_msg_init_size() failed");		return -1;	}	char *data = zmq_msg_data(&msg);	data[0] = msg_type_exit;	*(uint32_t*)&data[1] = htonl((uint32_t)code);	if (zmq_msg_send(&msg, socket, 0) == -1) {		TRACE_ERRNO("zmq_msg_send() failed");		if (zmq_msg_close(&msg) == -1)			TRACE_ERRNO("zmq_msg_close() failed");		return -1;	}	TRACE("sent exit message, code=%i", code);	return 0;}

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

void LastValueCache::replay_cache(void *entry){  std::vector<zmq_msg_t *> *cache_record = &_cache[entry];  if (cache_record->empty())  {    LOG_DEBUG("No cached record");    return;  }  LOG_DEBUG("Replaying cache for entry %p (length: %d)", cache_record, entry, cache_record->size());  std::vector<zmq_msg_t *>::iterator it;  for (std::vector<zmq_msg_t *>::iterator it = cache_record->begin();       it != cache_record->end();       it++)  {    zmq_msg_t message;    zmq_msg_init(&message);    zmq_msg_copy(&message, *it);    zmq_sendmsg(_publisher, &message, (it + 1 != cache_record->end()) ? ZMQ_SNDMORE : 0);    zmq_msg_close(&message);  }}

int main (){    void *ctx = zmq_init (1);    assert (ctx);    void *dealer = zmq_socket (ctx, ZMQ_DEALER);    assert (dealer);    int rc = zmq_connect (dealer, "tcp://localhost:2211");    assert (rc == 0);    while (1) {        //  Send signal to parent        zmq_msg_t msg;        zmq_msg_init (&msg);        zmq_sendmsg (dealer, &msg, 0);        zmq_msg_close (&msg);        //  Do some work        sleep (1);    }    return 0;}

int zmq_msg_copy (zmq_msg_t *dest_, zmq_msg_t *src_){    zmq_msg_close (dest_);    //  VSMs and delimiters require no special handling.    if (src_->content != (zmq::msg_content_t*) ZMQ_DELIMITER &&          src_->content != (zmq::msg_content_t*) ZMQ_VSM) {        //  One reference is added to shared messages. Non-shared messages        //  are turned into shared messages and reference count is set to 2.        zmq::msg_content_t *content = (zmq::msg_content_t*) src_->content;        if (src_->shared)            content->refcnt.add (1);        else {            src_->shared = true;            content->refcnt.set (2);        }    }    *dest_ = *src_;    return 0;}

void do_check(void* sb, void* sc, unsigned int msgsz){    setup_test_environment();    int rc;    int sum =0;    for (int i = 0; i < 10; i++)    {        zmq_msg_t msg;        zmq_msg_init_size(&msg, msgsz);        void * data = zmq_msg_data(&msg);        memcpy(data,&i, sizeof(int));        rc = zmq_msg_send(&msg,sc,i==9 ? 0 :ZMQ_SNDMORE);        assert (rc == (int)msgsz);        zmq_msg_close(&msg);        sum += i;    }    struct iovec ibuffer[32] ;    memset(&ibuffer[0], 0, sizeof(ibuffer));        size_t count = 10;    rc = zmq_recviov(sb,&ibuffer[0],&count,0);    assert (rc == 10);    int rsum=0;    for(;count;--count)    {        int v;        memcpy(&v,ibuffer[count-1].iov_base,sizeof(int));        rsum += v;        assert(ibuffer[count-1].iov_len == msgsz);        // free up the memory        free(ibuffer[count-1].iov_base);    }        assert ( sum == rsum );}

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

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

/** * Called by Java's Socket::recv(byte[] buffer, int offset, int len, int flags). */JNIEXPORT jint JNICALL Java_org_zeromq_ZMQ_00024Socket_recv___3BIII (JNIEnv *env,                                                                      jobject obj,                                                                      jbyteArray buff,                                                                      jint offset,                                                                      jint len,                                                                      jint flags){    zmq_msg_t message;    if (!do_read(env,obj,&message,flags)) {        return -1;    }    // No errors are defined for these two functions. Should they?    int sz = zmq_msg_size (&message);    void* pd = zmq_msg_data (&message);        int stored = sz > len ? len : sz;    env->SetByteArrayRegion (buff, offset, stored, (jbyte*) pd);    int rc = zmq_msg_close(&message);    assert(rc == 0);    return stored;}

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

static int luazmq_skt_sendx_impl(lua_State *L, int last_flag) {  zsocket *skt = luazmq_getsocket(L);  size_t i, n = lua_gettop(L);  for(i = 2; i<=n; ++i){    zmq_msg_t msg;    size_t len; const char *data = luaL_checklstring(L, i, &len);    int ret = zmq_msg_init_size(&msg, len);    if(-1 == ret){      ret = luazmq_fail(L, skt);      lua_pushinteger(L, i);      return ret + 1;    }    memcpy(zmq_msg_data(&msg), data, len);    ret = zmq_msg_send(&msg, skt->skt, (i == n)?last_flag:ZMQ_SNDMORE);    zmq_msg_close(&msg);    if(-1 == ret){      ret = luazmq_fail(L, skt);      lua_pushinteger(L, i);      return ret + 1;    }  }  return luazmq_pass(L);}

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

void * name_worker(void *p){    void *sender_s = zmq_socket(context, ZMQ_PUSH);    zmq_connect(sender_s, cstr);    while(1) {        zmq_msg_t msg;        mbuffer_key_t read_key;        opname_t *operation = (opname_t *)shm_mbuffer_get_read(name_data_storage, &read_key);        size_t real_size = (uintptr_t)(*operation).name - (uintptr_t)operation;        real_size += strlen((*operation).name) + 1;        zmq_msg_init_data(&msg, operation, real_size, shm_mbuff_put_read_zmq, name_data_storage);        zmq_send(sender_s, &msg, ZMQ_NOBLOCK);        zmq_msg_close(&msg);    }    pthread_exit(NULL);}

void* emit(void *context) {  void *emitter = zmq_socket(context, ZMQ_PUSH);  int rc;  rc = zmq_connect(emitter, "inproc://example");  if (rc != 0) {    printf("zmq_connect error: %s/n", zmq_strerror(errno));    abort();  }  char text[] = "Hello world!";  const size_t len = strlen(text);  while (1) {    zmq_msg_t message;    zmq_msg_init_data(&message, text, len, NULL, NULL);    rc = zmq_send(emitter, &message, 0);    if (rc != 0) {      printf("zmq_send error: %s/n", zmq_strerror(errno));      abort();    }    zmq_msg_close(&message);  }} /* void emit */

int zmq_send_const (void *s_, const void *buf_, size_t len_, int flags_){    zmq::socket_base_t *s = as_socket_base_t (s_);    if (!s)        return -1;    zmq_msg_t msg;    int rc =      zmq_msg_init_data (&msg, const_cast<void *> (buf_), len_, NULL, NULL);    if (rc != 0)        return -1;    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;        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;}

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

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

// Sends string as 0MQ string, as multipart non-terminalZMQ_EXPORT int WINAPI mql4s_sendmore (void *socket, char *text) {	// Strict "C" spec has to be followed because we outputing the function as 'extern "C"' (see mql4zmq.h).	// Hence specifing our variables right away instead of inline.    int result;	// Initialize message.    zmq_msg_t message;	// Set the message to have a spcified length.    zmq_msg_init_size (&message, strlen (text));	// Place the specified value of variable 'string' inside of the message buffer.     memcpy (zmq_msg_data (&message), text, strlen (text));	// Stream the message to the specified socket.    result = zmq_send (socket, &message, ZMQ_SNDMORE);	// Deallocate the message.    zmq_msg_close (&message);	// Return the response of the zmq_send call. 0 is success, -1 is error.    return (result);}

int zmq_send_const (void *s_, const void *buf_, size_t len_, int flags_){    if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) {        errno = ENOTSOCK;        return -1;    }    zmq_msg_t msg;    int rc = zmq_msg_init_data (&msg, (void *)buf_, len_, NULL, NULL);    if (rc != 0)        return -1;    zmq::socket_base_t *s = (zmq::socket_base_t *) s_;    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;        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;}

		void teRDNetwork::Send(u32 type, u32 * data, f32 * floatData)		{			if(!zmqContext)				return;			zmq_msg_t reply;			zmq_msg_init_size(&reply, sizeof(tePkg));			tePkg * pkg = (tePkg*)zmq_msg_data(&reply);			const c8 * magic = "TAT2";			pkg->magic = *(u32*)magic;			pkg->type = type;			if(data)				memcpy(pkg->d, data, 4 * sizeof(u32));			if(floatData)				memcpy(pkg->f, floatData, 4 * sizeof(f32));			zmq_send(zmqCmd, &reply, 0);			zmq_msg_close(&reply);		}

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