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

int Log_init(bstring access_log, bstring log_spec){    int rc = 0;    LogConfig *config = NULL;    if(LOG_SOCKET == NULL)     {        check(ZMQ_CTX, "No ZMQ context, cannot start access log.");        if(Setting_get_int("disable.access_logging", 0))        {            log_info("Access log is disabled according to disable.access_logging.");        }         else         {            config = LogConfig_create(access_log, log_spec);            check(config, "Failed to configure access logging.");            LOG_SOCKET = zmq_socket(ZMQ_CTX, ZMQ_PUB);            check(LOG_SOCKET != NULL, "Failed to create access log socket");#ifdef ZMQ_LINGER            int opt = 0;            rc = zmq_setsockopt(LOG_SOCKET, ZMQ_LINGER, &opt, sizeof(opt));            check(rc == 0, "Could not set the linger option.");#endif            rc = zmq_bind(LOG_SOCKET, bdata(log_spec));            check(rc == 0, "Failed to bind access_log zeromq socket.");            pthread_create(&LOG_THREAD, NULL, Log_internal_thread, config);        }    }    return 0;error:    LogConfig_destroy(config);    return -1;}

int zmq::socket_base_t::monitor (const char *addr_, int events_){    if (unlikely (ctx_terminated)) {        errno = ETERM;        return -1;    }    //  Support deregistering monitoring endpoints as well    if (addr_ == NULL) {        stop_monitor ();        return 0;    }    //  Parse addr_ string.    std::string protocol;    std::string address;    if (parse_uri (addr_, protocol, address) || check_protocol (protocol))        return -1;    //  Event notification only supported over inproc://    if (protocol != "inproc") {        errno = EPROTONOSUPPORT;        return -1;    }    //  Register events to monitor    monitor_events = events_;    monitor_socket = zmq_socket (get_ctx (), ZMQ_PAIR);    if (monitor_socket == NULL)        return -1;    //  Never block context termination on pending event messages    int linger = 0;    int rc = zmq_setsockopt (monitor_socket, ZMQ_LINGER, &linger, sizeof (linger));    if (rc == -1)        stop_monitor ();    //  Spawn the monitor socket endpoint    rc = zmq_bind (monitor_socket, addr_);    if (rc == -1)         stop_monitor ();    return rc;}

void throttleObject::sendLoginRegisterToThrottle(const string& ip,unsigned short manager_port){    if(!m_connectorLoginToThro)    {                managerProPackage::send(m_sendLHRToThrottleHandler, managerProtocol_messageTrans_CONNECTOR            , managerProtocol_messageType_LOGIN_REQ, ip, manager_port);        g_manager_logger->info("[login req][connector -> throttle]:{0},{1:d}", m_throttle_ip, m_throttle_managerPort);		        return;    }    for(auto it = m_throSubKey_list.begin(); it != m_throSubKey_list.end(); it++)    {        throSubKeyObject* subKeyObject = *it;        if(subKeyObject == NULL) continue;        string& key = subKeyObject->get_throSubKey();        if(key.empty()) continue;        if(subKeyObject->get_throSubscribed() == false)        {            //if don't setsockopt with ZMQ_SUBSCRIBE option,this zmq_socket will recv nothing !!!            for(auto it=m_recvAdReqVector.begin(); it != m_recvAdReqVector.end(); it++)            {                recvAdReq_t *obj = *it;                int rc = zmq_setsockopt(obj->m_recvAdReqFromThrottleHandler,ZMQ_SUBSCRIBE,key.c_str(),key.size());                if(rc == 0)                {                    subKeyObject->set_throSubscribed(true);                    g_manager_logger->info("[add ZMQ_SUBSCRIBE]: {0}",key);                    cout << "[add ZMQ_SUBSCRIBE]: " << key << endl;                }                }                                   }        if(subKeyObject->get_throRegisted() == false)        {            managerProPackage::send(m_sendLHRToThrottleHandler, managerProtocol_messageTrans_CONNECTOR               , managerProtocol_messageType_REGISTER_REQ, key, ip, manager_port);            g_manager_logger->info("[register req][connector -> throttle]:{0},{1:d}", m_throttle_ip, m_throttle_managerPort);               cout << "[register req][connector -> throttle]!" << endl;        }    }}

int main(void){    // connect to task ventilator    void *context = zmq_ctx_new();    void *receiver = zmq_socket(context, ZMQ_PULL);    zmq_connect(receiver, "tcp://localhost:5557");    // connect to weather server    void *subscriber = zmq_socket(context, ZMQ_SUB);    zmq_connect(subscriber, "tcp://localhost:5556");    zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "", 0);    // Process messages from both sockets    while (1) {        char msg[256];        zmq_pollitem_t items[] = {                { receiver, 0, ZMQ_POLLIN, 0 },                { subscriber, 0, ZMQ_POLLIN, 0 },        };        zmq_poll(items, 2, -1);        if (items[0].revents & ZMQ_POLLIN) {            int size = zmq_recv(receiver, msg, 255, 0);            if (size != -1) {                break;            }        }        if (items[1].revents & ZMQ_POLLIN) {            //int size = zmq_recv(subscriber, msg, 255, 0);            char *msg = s_recv(subscriber);            if (msg != NULL) {                // process weather update                puts(msg);                free(msg);            }        }    }    zmq_close(subscriber);    zmq_ctx_destroy(context);    return 0;}

int main(void){        void *context = zmq_ctx_new();        void *receiver = zmq_socket(context, ZMQ_PULL);        zmq_connect(receiver, "tcp://localhost:5557");        void *subscriber = zmq_socket(context, ZMQ_SUB);        zmq_connect(subscriber, "tcp://localhost:5556");        zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "10001 ", 6);        zmq_pollitem_t items[] = {                {receiver, 0, ZMQ_POLLIN, 0},                {subscriber, 0, ZMQ_POLLIN, 0},        };        while(1) {                zmq_msg_t message;                zmq_poll(items, 2, -1);                if (items[0].revents & ZMQ_POLLIN) {                        zmq_msg_init(&message);                        zmq_msg_recv(&message, receiver, 0);                        printf("receiver: %s/n", (char *)zmq_msg_data(&message));                        zmq_msg_close(&message);                }                if (items[1].revents & ZMQ_POLLIN) {                        zmq_msg_init(&message);                        zmq_msg_recv(&message, subscriber, 0);                        printf("subscribe: %s/n", (char *)zmq_msg_data(&message));                        zmq_msg_close(&message);                }        }        zmq_close(receiver);        zmq_close(subscriber);        zmq_ctx_destroy(context);        return 0;}

int main(int argc, char *argv []) {    void *context, *socket;    char *topic, *string;    int64_t more;        if (argc < 2) {        fprintf(stderr, "USAGE: zmq_echo <address> [topic]/n");        return 1;    }    context = zmq_init(1);    printf("connecting to %s.../n", argv[1]);    socket = zmq_socket(context, ZMQ_SUB);    zmq_connect(socket, argv[1]);        if (argc > 2) {        topic = argv[2];        printf("subscribing to /"%s/" topic.../n", topic);    } else {        topic = "";        printf("subscribing to all topics.../n");    }    zmq_setsockopt(socket, ZMQ_SUBSCRIBE, topic, strlen(topic));        printf(">> ");    fflush(stdout);        while (1) {        string = zmq_recv_string(socket, &more);        printf(more ? "%s" : "%s/n>> ", string);        fflush(stdout);        free(string);    }    zmq_close(socket);    zmq_term(context);    return 0;}

/** * Called by Java's Socket::setBytesSockopt(int option, byte[] value). */JNIEXPORT void JNICALL Java_org_zeromq_ZMQ_00024Socket_setBytesSockopt (JNIEnv *env,                                                                        jobject obj,                                                                        jint option,                                                                        jbyteArray value){    switch (option) {#if (ZMQ_VERSION_MAJOR <= 3)    case ZMQ_IDENTITY:#endif    case ZMQ_SUBSCRIBE:    case ZMQ_UNSUBSCRIBE:        {            if (value == NULL) {                raise_exception (env, EINVAL);                return;            }            void *s = get_socket (env, obj, 1);            jbyte *optval = env->GetByteArrayElements (value, NULL);            if (! optval) {                raise_exception (env, EINVAL);                return;            }            size_t optvallen = env->GetArrayLength (value);            int rc = zmq_setsockopt (s, option, optval, optvallen);            int err = zmq_errno();            env->ReleaseByteArrayElements (value, optval, 0);            if (rc != 0) {                raise_exception (env, err);            }            return;        }    default:        raise_exception (env, EINVAL);        return;    }}

int main(void){    struct heartbeat_socket_info * receiver;    struct heartbeat_socket_info * subscriber;    receiver = heartbeat_connect_socket(ZMQ_PULL, "tcp://localhost:5557");    subscriber = heartbeat_connect_socket(ZMQ_SUB, "tcp://localhost:5556");    zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "10001", 6);    char *string1;    char *string2;    while (1) {        char msg[256];        zmq_pollitem_t items[] = {            { receiver, 0, ZMQ_POLLIN, 0},            { subscriber, 0, ZMQ_POLLIN, 0}        };        zmq_poll(items, 2, -1);        if (items[0].revents & ZMQ_POLLIN) {            string1 = heartbeat_recv_msg(receiver, 0);            if (strlen(string1) != -1)            {            }            free(string1);        }        if (items[1].revents & ZMQ_POLLIN) {            string2 = heartbeat_recv_msg(subscriber, 0);            if (strlen(string2) != -1)            {            }            free(string2);        }    }    heartbeat_disconnect_socket(subscriber);    heartbeat_disconnect_socket(receiver);    return 0;}

void Chat::init(const std::string& my_id){	std::string bind_ep;	listening_port_ = create_port(my_id);	bind_ep = std::string("tcp://*:") + boost::lexical_cast<std::string>(listening_port_);	zmq_ctx_ = g_zmq_ctx.get_zmq_ctx();	publisher_ = zmq_socket(zmq_ctx_, ZMQ_PUB);	zmq_bind(publisher_, bind_ep.c_str());	subscriber_ = zmq_socket(zmq_ctx_, ZMQ_SUB);	zmq_setsockopt(subscriber_, ZMQ_SUBSCRIBE, "", 0);	gate_ = zmq_socket(zmq_ctx_, ZMQ_PAIR);	zmq_bind(gate_, MAGIC_GATE);	std::string test_msg("hello from ");	test_msg += my_id;	zmq_send(publisher_, test_msg.c_str(), test_msg.length(), 0);}

//********************************************************************************// Function: redirect_count//// Description: Recieve Control Plane redirect count.//********************************************************************************static int32_t redirect_count(void){    void *context = zmq_ctx_new ();    void *requester = zmq_socket(context, ZMQ_REQ);    int32_t timeo{};    zmq_setsockopt(requester, ZMQ_LINGER, (void*) &timeo, sizeof(timeo));    char buffer[64];    sprintf(buffer, "tcp://localhost:%d", TCPLANE_SERVICE);    zmq_connect(requester, buffer);    sprintf(buffer, "%d", redirectCount);    zmq_send(requester, buffer, strlen(buffer), 0);    size_t size = timed_read(requester, buffer, sizeof(buffer), READ_TIMEOUT);    if(size)        size = atoi(buffer);        zmq_close(requester);    zmq_ctx_destroy(context);    return size;}

int main(){  void *context=zmq_ctx_new();  void *sink=zmq_socket(context, ZMQ_ROUTER);  zmq_bind(sink, "inproc://example");  void *anonymous=zmq_socket(context, ZMQ_REQ);  zmq_connect(anonymous, "inproc://example");  s_send(anonymous, "ROUTER uses a generated UUID", 0, NULL);  s_dump(sink);  void *identified=zmq_socket(context, ZMQ_REQ);  zmq_setsockopt(identified, ZMQ_IDENTITY, "PEER2", 5);  zmq_connect(identified, "inproc://example");  s_send(identified, "ROUTER socker uses REQ's socker identity", 0, NULL);  s_dump(sink);  zmq_close(identified);  zmq_close(anonymous);  zmq_close(sink);  zmq_ctx_destroy(context);  return 0;}

void pre_allocate_sock (void *zmq_socket, const char *path){    struct sockaddr_un addr;    addr.sun_family = AF_UNIX;    strcpy (addr.sun_path, path);    unlink (path);    int s_pre = socket (AF_UNIX, SOCK_STREAM, 0);    assert (s_pre != -1);    int rc = bind (s_pre, (struct sockaddr *) &addr,            sizeof (struct sockaddr_un));    assert (rc == 0);    rc = listen (s_pre, SOMAXCONN);    assert (rc == 0);    rc = zmq_setsockopt (zmq_socket, ZMQ_USE_FD, &s_pre,            sizeof (s_pre));    assert(rc == 0);}

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

static int luazmq_skt_set_str_arr (lua_State *L, int option_name) {  zsocket *skt;  size_t len, tlen, i;  const char *option_value;  int ret;  if(!lua_istable(L, 2)) return luazmq_skt_set_str(L, option_name);  skt = luazmq_getsocket(L);  tlen = lua_objlen(L,2);  for (i = 1; i <= tlen; i++){    lua_rawgeti(L, 2, i);    option_value = luaL_checklstring(L, -1, &len);    ret = zmq_setsockopt(skt->skt, option_name, option_value, len);    if (ret == -1){      int n = luazmq_fail(L, skt);      lua_pushnumber(L, i);      return n + 1;    }  }  return luazmq_pass(L);}

        bool SubscriberZMQ::_onInit(TUInt port)        {            m_context = zmq_ctx_new();            if (m_context == 0)            {                zmqlog("Error occurred during zmq_ctx_new()");                return false;            }                         m_socket = zmq_socket(m_context, ZMQ_SUB);            if (m_socket == 0)            {                zmqlog("Error occurred during zmq_socket()");                clear();                return false;            }                         TChar address[100] = {'/0'};            sprintf(address, "tcp://localhost:%u", port);            int result = zmq_connect(m_socket, address);            if (result != 0)            {                zmqlog("Error occurred during zmq_init()");                clear();                return false;            }                        const TString& subscribeId = getSubscribeId();            result = zmq_setsockopt (m_socket, ZMQ_SUBSCRIBE, subscribeId.c_str(), subscribeId.length() - 1);            if (result != 0)            {                zmqlog("Error occurred during zmq_setsockopt");                clear();                return false;            }                        return true;        }

int main (void){    setup_test_environment();    void *ctx = zmq_ctx_new ();    assert (ctx);        //  Spawn ZAP handler    void *zap_thread = zmq_threadstart (&zap_handler, ctx);    //  Server socket will accept connections    void *server = zmq_socket (ctx, ZMQ_DEALER);    assert (server);    int rc = zmq_setsockopt (server, ZMQ_IDENTITY, "IDENT", 6);    assert (rc == 0);    rc = zmq_bind (server, "tcp://*:9999");    assert (rc == 0);            //  Client socket that will try to connect to server    void *client = zmq_socket (ctx, ZMQ_DEALER);    assert (client);    rc = zmq_connect (client, "tcp://localhost:9999");    assert (rc == 0);        bounce (server, client);        rc = zmq_close (client);    assert (rc == 0);    rc = zmq_close (server);    assert (rc == 0);        //  Shutdown    rc = zmq_ctx_term (ctx);    assert (rc == 0);        //  Wait until ZAP handler terminates.    zmq_threadclose (zap_thread);    return 0;}

//  We will do this all in one thread to emphasize the sequence//  of events...int main (void) {    void *context = zmq_init (1);    void *client = zmq_socket (context, ZMQ_ROUTER);    zmq_bind (client, "ipc://routing.ipc");    void *worker = zmq_socket (context, ZMQ_REP);    zmq_setsockopt (worker, ZMQ_IDENTITY, "A", 1);    zmq_connect (worker, "ipc://routing.ipc");    //  Wait for the worker to connect so that when we send a message    //  with routing envelope, it will actually match the worker...    sleep (1);    //  Send papa address, address stack, empty part, and request    s_sendmore (client, "A");    s_sendmore (client, "address 3");    s_sendmore (client, "address 2");    s_sendmore (client, "address 1");    s_sendmore (client, "");    s_send     (client, "This is the workload");    //  Worker should get just the workload	printf("begine dump worker/n");    s_dump (worker);    //  We don't play with envelopes in the worker    s_send (worker, "This is the reply");    //  Now dump what we got off the ROUTER socket...	printf("begine dump client/n");    s_dump (client);    zmq_close (client);    zmq_close (worker);    zmq_term (context);    return 0;}

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

int main (int argc, char *argv []){    fprintf (stderr, "test_router_behavior running.../n");    void *ctx = zmq_init (1);    assert (ctx);    // Creating the first socket.    void *sa = zmq_socket (ctx, ZMQ_ROUTER);    assert (sa);        int rc = zmq_bind (sa, "tcp://127.0.0.1:15560");    assert (rc == 0);    // Sending a message to an unknown peer with the default behavior.    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE);    assert (rc == 7);    rc = zmq_send (sa, "DATA", 4, 0);    assert (rc == 4);    int behavior = 1;    // Setting the socket behavior to a new mode.    rc = zmq_setsockopt (sa, ZMQ_ROUTER_BEHAVIOR, &behavior, sizeof (behavior));    assert (rc == 0);    // Sending a message to an unknown peer with verbose behavior.    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE | ZMQ_DONTWAIT);    assert (rc == -1 && errno == EAGAIN);    rc = zmq_close (sa);    assert (rc == 0);    rc = zmq_term (ctx);    assert (rc == 0);    return 0 ;}

/** * Called by Java's Socket::setStringSockopt(int option, String optval). */JNIEXPORT void JNICALL Java_org_zeromq_ZMQ_00024Socket_setStringSockopt (JNIEnv *env,                                                                         jobject obj,                                                                         jint option,                                                                         jstring optval){    switch (option) {    case ZMQ_IDENTITY:    case ZMQ_SUBSCRIBE:    case ZMQ_UNSUBSCRIBE:        {            if (optval == NULL) {                raise_exception (env, EINVAL);                return;            }	    void *s = get_socket (env, obj, 1);            const char *value = env->GetStringUTFChars (optval, NULL);            if (! value) {                raise_exception (env, EINVAL);                return;            }            int rc = zmq_setsockopt (s, option, value, strlen (value));            int err = errno;            env->ReleaseStringUTFChars (optval, value);            if (rc != 0) {                raise_exception (env, err);                return;            }            return;        }    default:        raise_exception (env, EINVAL);        return;    }}

int _tmain(int argc, _TCHAR* argv[]){	setlocale(LC_ALL,"Chinese");	setlocale(LC_ALL,"chs");	void *m_context;	void *m_subscriber;	char m_subAddr[64];	m_context = zmq_init(1);	m_subscriber = zmq_socket(m_context,ZMQ_SUB);		char *puberIp = "127.0.0.1";	WORD port = 8585;	memset(m_subAddr,0,sizeof(m_subAddr));	sprintf_s(m_subAddr,"tcp://%s:%d",puberIp,port);	zmq_connect(m_subscriber,m_subAddr);	char *option = "642";	int ret = zmq_setsockopt(m_subscriber,ZMQ_SUBSCRIBE,"642",strlen(option));	while (1)	{		BYTE buffer[1024] = {0};		DWORD bufLen = sizeof(buffer);		DWORD gotLen = zmq_recv(m_subscriber,buffer,bufLen,0);		printf("收到发布信息：%s/n",buffer);	}	zmq_close(m_subscriber);	zmq_term(m_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");}

int main (void){    fprintf (stderr, "test_router_mandatory_tipc running.../n");    void *ctx = zmq_init (1);    assert (ctx);    // Creating the first socket.    void *sa = zmq_socket (ctx, ZMQ_ROUTER);    assert (sa);    int rc = zmq_bind (sa, "tipc://{15560,0,0}");    assert (rc == 0);    // Sending a message to an unknown peer with the default setting    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE);    assert (rc == 7);    rc = zmq_send (sa, "DATA", 4, 0);    assert (rc == 4);    int mandatory = 1;    // Set mandatory routing on socket    rc = zmq_setsockopt (sa, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory));    assert (rc == 0);    // Send a message and check that it fails    rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE | ZMQ_DONTWAIT);    assert (rc == -1 && errno == EHOSTUNREACH);    rc = zmq_close (sa);    assert (rc == 0);    rc = zmq_term (ctx);    assert (rc == 0);    return 0 ;}

int main(void){    // Prepare our context and subscriber    void *context = zmq_ctx_new();    void *subscriber = zmq_socket(context, ZMQ_SUB);    zmq_connect(subscriber, "tcp://localhost:5563");    zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, "B", 1);    while (1) {        // Read envelope with address        char *address = s_recv(subscriber);        // Read messages contents        char *contents = s_recv(subscriber);        printf("[%s] %s/n", address, contents);        free(address);        free(contents);    }    // We never get here, but clean up anyhow    zmq_close(subscriber);    zmq_ctx_destroy(context);    return 0;}

int main (int argc, char *argv []){  // Socket to talk to server  printf ("Collecting updates from chat server…/n");  void *context = zmq_ctx_new ();  void *subscriber = zmq_socket (context, ZMQ_SUB);  int rc = zmq_connect (subscriber, "tcp://localhost:5556");  assert (rc == 0);  char *filter = (argc > 1)? argv [1]: ">";  rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,		       filter, strlen (filter));  assert (rc == 0);  for (;;) {    char *string = s_recv (subscriber);    printf("Received: %s/n", string);    free(string);  }  zmq_close (subscriber);  zmq_ctx_destroy (context);  return 0;}

int main (void){    //  Create the infrastructure    void *ctx = zmq_ctx_new ();    assert (ctx);    void *sb = zmq_socket (ctx, ZMQ_ROUTER);    assert (sb);    int val = 0;    int rc = zmq_setsockopt (sb, ZMQ_LINGER, &val, sizeof (val));    assert (rc == 0);    do_bind_and_verify (sb, "tcp://127.0.0.1:5560");    do_bind_and_verify (sb, "tcp://127.0.0.1:5561");    rc = zmq_close (sb);    assert (rc == 0);        rc = zmq_ctx_term (ctx);    assert (rc == 0);    return 0 ;}

int main(){    void* context = zmq_ctx_new();    void* sink = zmq_socket(context, ZMQ_ROUTER);    zmq_bind(sink, "inproc://identity-example");    // First allow 0MQ to set the identity    void* anonymous = zmq_socket(context, ZMQ_REQ);    zmq_connect(anonymous, "inproc://identity-example");    s_send(anonymous, "ROUTER uses a generated UUID");    s_dump(sink);    // Then set the identity ourselves    void* identified = zmq_socket(context, ZMQ_REQ);    zmq_setsockopt(identified, ZMQ_IDENTITY, "PEER2", 5);    zmq_connect(identified, "inproc://identity-example");    s_send(identified, "ROUTER socket uses REQ's socket identity");    s_dump(sink);    zmq_close(sink);    zmq_close(anonymous);    zmq_close(identified);    zmq_ctx_destroy(context);    return 0;}

int main(void) {  void *context = zmq_ctx_new();  // set up sender to send messages to the sb  void *sender = zmq_socket(context, ZMQ_PUSH);  zmq_connect(sender, "ipc:///tmp/zero_sb_in");  printf("PO: Connected to SB to send messages/n");  // set up subscriber to get new messages from subscribed MIDs  void *subscriber = zmq_socket(context, ZMQ_SUB);  int rc = zmq_connect(subscriber, "ipc:///tmp/zero_sb_pub");  assert(rc == 0);  printf("PO: Connected to SB to receive messages/n");  char *filter = "10";  rc = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter));  assert(rc == 0);  printf("PO: /tAnd the filter is in place./n");  // Wait for a ping  char buffer[256];  rc = zmq_recv(subscriber, buffer, 256, 0);  if (rc > 255) {    printf("PO: I can't handle this/n");  } else {    buffer[rc] = '/0';    printf("PO: Received: %s/n", buffer);  }  // Respond with a pong  zmq_send(sender, "11 pong", 7, 0);    zmq_close(sender);  zmq_close(subscriber);  zmq_ctx_destroy(context);  return 0;}

int main(int argc, char *argv[]){    // Socket to talk to server    printf("Collecting updates from weather server.../n");    void *context = zmq_ctx_new();    void *subscriber = zmq_socket(context, ZMQ_SUB);    int rc = zmq_connect(subscriber, "tcp://localhost:5556");    assert(rc == 0);    // Subscriber to zipcode, default is NYC, 10001    char *filter = (argc > 1)? argv[1] : "10001 ";    rc = zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter));    assert(rc == 0);    // Process 100 updates    int update_nbr;    long total_temp = 0;    for(update_nbr=0; update_nbr < 100;update_nbr++)    {        char *string = s_recv(subscriber);        int zipcode, temperature, relhumidity;        sscanf(string, "%d %d %d", &zipcode, &temperature, &relhumidity);        printf("%03d   ", update_nbr);        printf(string);        printf(" %d/n");        total_temp += temperature;        free(string);    }    printf("Average temperature for zipcode '%s' was %dF/n", filter, (int)(total_temp/update_nbr));    zmq_close(subscriber);    zmq_ctx_destroy(context);    return 0;}

//  We have two workers, here we copy the code, normally these would//  run on different boxes...//static void *worker_task_a (void *args){    void *context = zmq_init (1);    void *worker = zmq_socket (context, ZMQ_DEALER);    zmq_setsockopt (worker, ZMQ_IDENTITY, "A", 1);    zmq_connect (worker, "ipc://routing.ipc");    int total = 0;    while (1) {        //  We receive one part, with the workload        char *request = s_recv (worker);        int finished = (strcmp (request, "END") == 0);        free (request);        if (finished) {            printf ("A received: %d/n", total);            break;        }        total++;    }    zmq_close (worker);    zmq_term (context);    return NULL;}