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

void reply_message::emit(libfault::zmq_msg_vector& msg) {  assert(zmqbodyused == false);  // first block is the reply status  zmq_msg_t* z_status = msg.insert_back();  zmq_msg_init_size(z_status, sizeof(reply_status));  (*reinterpret_cast<reply_status*>(zmq_msg_data(z_status))) = status;  // second block is the property bag  graphlab::oarchive oarc;  oarc << properties;  zmq_msg_t* z_propertybag = msg.insert_back();  zmq_msg_init_data(z_propertybag, oarc.buf, oarc.off,                     libfault::zmq_msg_standard_free, NULL);  // third block is the serialization body  zmq_msg_t* z_body= msg.insert_back();  if (body != NULL) {    zmq_msg_init_data(z_body, body, bodylen, libfault::zmq_msg_standard_free, NULL);  }    // we are giving away the body pointer  body = NULL;  clear();}

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

ssize_t uwsgi_proto_zeromq_write(struct wsgi_request * wsgi_req, char *buf, size_t len) {	zmq_msg_t reply;	char *zmq_body;	if (len == 0)		return 0;	zmq_body = uwsgi_concat2n(wsgi_req->proto_parser_buf, (int) wsgi_req->proto_parser_pos, buf, (int) len);	//uwsgi_log("|%.*s|/n", (int)wsgi_req->proto_parser_pos+len, zmq_body);	zmq_msg_init_data(&reply, zmq_body, wsgi_req->proto_parser_pos + len, uwsgi_proto_zeromq_free, NULL);	if (uwsgi.threads > 1) pthread_mutex_lock(&uwsgi.zmq_lock);	if (zmq_send(uwsgi.zmq_pub, &reply, 0)) {		if (!uwsgi.ignore_write_errors) {			uwsgi_error("zmq_send()");		}		wsgi_req->write_errors++;		if (uwsgi.threads > 1) pthread_mutex_unlock(&uwsgi.zmq_lock);		zmq_msg_close(&reply);		return 0;	}	if (uwsgi.threads > 1) pthread_mutex_unlock(&uwsgi.zmq_lock);	zmq_msg_close(&reply);	return len;}

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

int main(){    void *ctx, *s;    zmq_msg_t query, result;    ctx = zmq_init(1);    s = zmq_socket(ctx, ZMQ_REP);    if (zmq_bind(s, "tcp://127.0.0.1:5554") < 0) {        printf("could not bind/n");        exit(-1);    }    while (1) {        printf("listening.../n");        zmq_msg_init(&query);        zmq_recv(s, &query, 0);        int msg_len = zmq_msg_size(&query) + 1;        char *out = (char *) malloc(msg_len);        snprintf(out, msg_len, "%s", (char *) zmq_msg_data(&query));        out[msg_len-1]='/0';        printf("%s/n", out);        zmq_msg_close(&query);        free(out);        zmq_msg_init_data(&result, reply, strlen(reply), NULL, NULL);        zmq_send(s, &result, 0);        zmq_msg_close(&result);    }}

intmain(int argc, char *argv){    int i = 0;    char msg[10] = "aaaaaaaaaa";    void *context = zmq_init(1);    void *sender = zmq_socket(context, ZMQ_PUSH);    zmq_msg_t m;    printf("%d bind/n", zmq_bind(sender, "tcp://*:3333"));    //printf("%d bind/n", zmq_bind(sender, "ipc:///tmp/zmqpy_bench"));    for (i = 0; i < 10000000; i++) {        if (i % 10000 == 0)            printf("%d/n", i);#if (ZMQ_VERSION_MAJOR == 3)        zmq_send(sender, msg, 10, 0);#else        zmq_msg_init_data(&m, msg, 10, NULL, 0);        zmq_send(sender, &m, 0);        zmq_msg_close(&m);#endif    }    zmq_close(sender);    zmq_term(context);}

// Move operators will take ownership of message parts without copyingvoid message::move(void* part, size_t& size, release_function const& release){	parts_type tmp(_parts.size() + 1);	for(size_t i = 0; i < _parts.size(); ++i)	{		zmq_msg_t& dest = tmp[i].msg;		if( 0 != zmq_msg_init(&dest) )		{			throw zmq_internal_exception();		}		zmq_msg_t& src = _parts[i].msg;		if( 0 != zmq_msg_move(&dest, &src) )		{			throw zmq_internal_exception();		}	}	std::swap(tmp, _parts);	callback_releaser* hint = new callback_releaser();	hint->func = release;	zmq_msg_t& msg = _parts.back().msg;	if (0 != zmq_msg_init_data(&msg, part, size, &message::release_callback, hint))	{		throw zmq_internal_exception();	}}

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

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

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

int Handler_deliver(void *handler_socket, char *buffer, size_t len){    int rc = 0;    zmq_msg_t *msg = calloc(sizeof(zmq_msg_t), 1);    bstring msg_buf = NULL;    rc = zmq_msg_init(msg);    check(rc == 0, "Failed to initialize 0mq message to send.");    msg_buf = blk2bstr(buffer, len);    check_mem(msg_buf);    rc = zmq_msg_init_data(msg, bdata(msg_buf), blength(msg_buf), bstring_free, msg_buf);    check(rc == 0, "Failed to init 0mq message data.");    rc = mqsend(handler_socket, msg, 0);    check(rc == 0, "Failed to deliver 0mq message to handler.");    if(msg) free(msg);    return 0;error:    // TODO: confirm what if this is the right shutdown    if(msg) free(msg);    return -1;}

int main(int argc, char *argv[]){  if (argc < 2) return EXIT_FAILURE;  int N = atoi(argv[1]);  printf("N: %d/n", N);  void *ctx = zmq_init(1);  void *pub = zmq_socket(ctx, ZMQ_PUB); // Output channel of 0  assert(pub);  void *sub = zmq_socket(ctx, ZMQ_SUB); // Input channel of 0  assert(sub);  int rc;  rc = zmq_bind(sub, "tcp://*:8887"); // Waits for publishers  assert(rc == 0);  rc = zmq_connect(pub, "tcp://localhost:8888"); // Actively connect to subscribers  assert(rc == 0);  zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "", 0);  int *val = (int *)calloc(N, sizeof(int));  zmq_msg_t msg;  long long start_time = sc_time();  // Send  int *buf = (int *)calloc(N, sizeof(int));  memcpy(buf, val, N * sizeof(int));  zmq_msg_init_data(&msg, buf, N * sizeof(int), _dealloc, NULL);  zmq_send(pub, &msg, 0);  zmq_msg_close(&msg);  // Receive  zmq_msg_init(&msg);  zmq_recv(sub, &msg, 0);  memcpy(val, (int *)zmq_msg_data(&msg), zmq_msg_size(&msg));  zmq_msg_close(&msg);  long long end_time = sc_time();  printf("%s: Time elapsed: %f sec/n", argv[0], sc_time_diff(start_time, end_time));#ifdef __DEBUG__  int i;  printf("%s [ ", argv[0]);  for (i=0; i<N; ++i) {    printf("%d ", val[i]);  }  printf("]/n");#endif  free(val);  zmq_close(sub);  zmq_close(pub);  zmq_term(ctx);  return EXIT_SUCCESS;}

 inline message_t (void *data_, size_t size_, free_fn *ffn_,     void *hint_ = NULL) {     int rc = zmq_msg_init_data (&msg, data_, size_, ffn_, hint_);     if (rc != 0)         throw error_t (); }

 inline void rebuild(void *data_, size_t size_, free_fn *ffn_, void *hint_ = NULL) {     int rc = zmq_msg_close(&msg);     if (rc != 0)         throw error_t();     rc = zmq_msg_init_data(&msg, data_, size_, ffn_, hint_);     if (rc != 0)         throw error_t(); }

 message(nocopy_t, boost::asio::mutable_buffer const& buffer, void* hint, zmq_free_fn* deleter) {     auto rc = zmq_msg_init_data(&msg_,                                 boost::asio::buffer_cast<void*>(buffer),                                 boost::asio::buffer_size(buffer),                                 deleter, hint);     if (rc)         throw boost::system::system_error(make_error_code()); }

void deliver_message(void* sock, int size, int flags){  zmq_msg_t msg;  int retval;  // Prep the message from preset value. ZMQ takes ownership.  void *data = calloc(size,1);  assert(data != NULL);  for(int i=0; i<size; i++){    ((char*) data)[i] = 'A';  }  retval = zmq_msg_init_data(&msg, data, size, free_msg, NULL);  assert(retval == 0);  // printf("sending %s/n",zmq_msg_data(&msg));  retval = zmq_send(sock,&msg,flags);  if(retval != 0){    switch(errno){      case EAGAIN:        puts("zmq_send EAGAIN");        break;      case ENOTSUP:        puts("zmq_send ENOTSUP");        break;      case EFSM:        puts("zmq_send EFSM");        break;      case ETERM:        puts("zmq_send ETERM");        break;      case ENOTSOCK:        puts("zmq_send ENOTSOCK");        break;      case EINTR:        puts("zmq_send EINTR");        break;      case EFAULT:        puts("zmq_send EFAULT");        break;      default:        puts("zmq_send: errno default");    }    exit(EXIT_FAILURE);  } else {    #ifdef LOGGING      puts("sent the message");    #endif  }  // Cleanup  #ifdef LOGGING    printf("cleaning up msg...");  #endif  zmq_msg_close(&msg);}

static voidzmqdrv_send(zmq_drv_t *drv, ErlIOVec *ev){    ErlDrvBinary*  bin   = ev->binv[1];    char*          bytes = bin->orig_bytes;    uint32_t       idx   = ntohl(*(uint32_t*)(bytes+1));    zmq_sock_info* si    = drv->get_socket_info(idx);    uint32_t       flags = ntohl(*(uint32_t*)(bytes+5));    void*          data  = (void *)(bytes + 9);    size_t         size  = bin->orig_size - 9;    if (idx > drv->zmq_socket_count || !si) {        zmqdrv_error_code(drv, ENODEV);        return;    }#ifdef ZMQDRV_DEBUG    uint32_t events;    size_t events_size = sizeof(events);    zmq_getsockopt(si->socket, ZMQ_EVENTS, &events, &events_size);    zmqdrv_fprintf("sending %p [idx=%d] %lu bytes (events=%d)/r/n", si->socket, idx, size, events);#endif    if (si->out_caller != 0) {        // There's still an unwritten message pending        zmqdrv_error_code(drv, EBUSY);        return;    }    // Increment the reference count on binary so that zmq can    // take ownership of it.    driver_binary_inc_refc(bin);    if (zmq_msg_init_data(&si->out_msg, data, size, &zmq_free_binary, bin)) {        zmqdrv_error_code(drv, zmq_errno());        driver_binary_dec_refc(bin);        return;    }    if (zmq_send(si->socket, &si->out_msg, flags | ZMQ_NOBLOCK) == 0) {        zmqdrv_ok(drv);        zmqdrv_ready_input((ErlDrvData)drv, (ErlDrvEvent)si->fd);    } else {        int e = zmq_errno();        if (e == EAGAIN) {            // No msg returned to caller - make him wait until async            // send succeeds            si->out_caller = driver_caller(drv->port);            return;        }        zmqdrv_error_code(drv, e);    }    zmq_msg_close(&si->out_msg);}

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

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

        bool SendMessage(Msgpack* msgpack, bool isRelease = true) const            {                try {                    zmq_msg_t msg;                    zmq_msg_init(&msg);                    if (isRelease) {                        zmq_msg_init_data(&msg, msgpack->GetSbuf().data(), msgpack->GetSbuf().size(), Tool::Network::Release, msgpack);                    } else {                        zmq_msg_init_data(&msg, msgpack->GetSbuf().data(), msgpack->GetSbuf().size(), 0, 0);                    }                    zmq_msg_send(&msg, m_socket, 0);                    return true;                } catch (...) {                    std::cerr << "Network Send Failed." << std::endl;                }                return false;            }

int main(int argc, char **argv) {  logging_init("cli.log");  void *ctx = zmq_init(1);  void *socket = zmq_socket(ctx, ZMQ_DEALER);  int rc = zmq_connect(socket, "tcp://127.0.0.1:9999");  // use below for when this is a router socket  // int rc = zmq_bind(socket, "tcp://127.0.0.1:9999");  assert(rc == 0);  zmq_msg_t msg;  size_t msg_size = sizeof(int);  int x = 99;  boost::posix_time::ptime start_ptime(      boost::posix_time::microsec_clock::local_time());  sleep(3);#ifdef ZC  // with zerocopy  rc = zmq_msg_init_data(&msg, (void *)&x, msg_size, NULL, NULL);  assert(rc == 0);#else  // without zerocopy  pan::log_DEBUG("NOT using zerocopy");  rc = zmq_msg_init_size(&msg, msg_size);  assert(rc == 0);  memcpy(zmq_msg_data(&msg), (void *)&x, msg_size);#endif  // use below line for router socket  // must be the same identity as the receiving socket has set  // s_sendmore(socket, "A");  rc = zmq_send(socket, &msg, 0);  assert(rc == 0);  zmq_msg_close(&msg);  pan::log_DEBUG("Receiving message");  zmq_msg_init(&msg);  rc = zmq_recv(socket, &msg, 0);  pan::log_DEBUG("Received msg: ", pan::integer(*(int *)zmq_msg_data(&msg)));  zmq_close(&msg);  boost::posix_time::ptime stop_ptime(      boost::posix_time::microsec_clock::local_time());  boost::posix_time::time_duration ptime_duration(stop_ptime - start_ptime);  std::cerr << ptime_duration << "/n";  zmq_close(socket);  zmq_term(ctx);}

voidsend(char *buf, size_t size){    zmq_msg_t msg;    void *sock, *context;    context = zmq_init(1);    sock = zmq_socket(context, ZMQ_PUSH);    zmq_connect(sock, "tcp://localhost:5555");    zmq_msg_init_data(&msg, buf, size, NULL, NULL);    zmq_send(sock, &msg, 0);    zmq_close(sock);    zmq_term(context);}

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

int main (void) {    void *context = zmq_init (1);    //  Socket to send messages on    void *sender = zmq_socket (context, ZMQ_PUSH);    uint64_t hwm = 0;    zmq_setsockopt (sender, ZMQ_HWM, &hwm, sizeof (hwm));    int linger = 0;    zmq_setsockopt (sender, ZMQ_LINGER, &linger, sizeof (linger));    zmq_bind (sender, "tcp://*:5557");    pthread_t worker;    pthread_create (&worker, NULL, worker_task_a, NULL);    pthread_create (&worker, NULL, worker_task_a, NULL);    //  Wait for threads to connect, since otherwise the messages    //  we send will be dropped    sleep (1);    printf ("Sending tasks to workers.../n");    int rc = 0;    int i;    int successful = 0;    int eagain = 0;        for(i = 0; i < 1000000; i++) {      zmq_msg_t task;      zmq_msg_init_data (&task, "Hello world", 11, NULL, NULL);      rc = zmq_send (sender, &task, 0/*ZMQ_NOBLOCK */);            if (rc == 0) {        successful++;      }      else {        eagain++;      }            zmq_msg_close (&task);    }        printf("Successfully sent [%d] messages/n", successful);    printf("EAGAIN received for [%d] messages/n", eagain);    zmq_close (sender);    zmq_term (context);    return 0;}

static gint lua_zmq_send(lua_State* lua) {  gpointer socket = *(gpointer*)luaL_checkudata(lua, 1, SOCKET_TYPE);  gsize len;  const gchar* str = luaL_checklstring(lua, 2, &len);  gint flags = luaL_optint(lua, 3, 0);  zmq_msg_t msg;  zmq_msg_init_data(&msg,                    memcpy(g_new(gchar, len), str, len),                    len,                    zmq_g_free,                    NULL);  if (zmq_send(socket, &msg, flags) == -1) {    zmq_msg_close(&msg);    return lua_zmq_error(lua, "zmq_send");  }  return 0;}

int main(){	void *context = zmq_ctx_new();	void *pusher = zmq_socket(context, ZMQ_PUSH);	zmq_bind(pusher, "tcp://*:5555");		while(1){		zmq_msg_t send;		zmq_msg_init_data(&send, "hello", 5, NULL, NULL);		printf("Sending hello/n");		zmq_msg_send(&send, pusher, 0);		zmq_msg_close(&send);		sleep(1);	}	zmq_close(pusher);	zmq_ctx_destroy(context);	return 0;}

void publish_output(process_t *process, struct proc_data *procdata,                    const char *ioname, const char *data, ssize_t length) {  /* TODO(sissel): move this to a separate file for 'event' streaming */  zmq_msg_t event;  int rc;  size_t msgsize;  program_t *program = pn_proc_program(process);  procnanny_t *pn = program->data;  fprintf(stdout, "%s[%d]: (%d bytes) %.*s/n", pn_proc_program(process)->name,          pn_proc_instance(process), length, length, data);  /* Fields:   *  - data (the string read)   *  - program name   *  - process instance   *  - stdout or stderr   */  msgpack_sbuffer *buffer = msgpack_sbuffer_new();  msgpack_packer *output_msg = msgpack_packer_new(buffer, msgpack_sbuffer_write);  msgpack_pack_map(output_msg, 5);  /* "event" => "data" */  msgpack_pack_string(output_msg, "event", -1);  msgpack_pack_string(output_msg, "data", -1);  msgpack_pack_string(output_msg, "program", -1);  msgpack_pack_string(output_msg, program->name, program->name_len);  msgpack_pack_string(output_msg, "instance", -1);  msgpack_pack_int(output_msg, process->instance);  msgpack_pack_string(output_msg, "source", -1);  msgpack_pack_string(output_msg, ioname, -1);  msgpack_pack_string(output_msg, "data", -1);  msgpack_pack_string(output_msg, data, length);  zmq_msg_init_data(&event, buffer->data, buffer->size, free_msgpack_buffer, buffer);   zmq_send(pn->eventsocket, &event, 0);  zmq_msg_close(&event);  msgpack_packer_free(output_msg);} /* publish_output */