int main (void)
{
    //  Prepare our context and publisher socket
    zctx_t *ctx = zctx_new ();
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_set_hwm (publisher, 0);
    zsocket_bind (publisher, "tcp://*:5556");
    zclock_sleep (200);

    zhash_t *kvmap = zhash_new ();
    int64_t sequence = 0;
    srandom ((unsigned) time (NULL));

    while (!zctx_interrupted) {
        //  Distribute as key-value message
        kvmsg_t *kvmsg = kvmsg_new (++sequence);
        kvmsg_fmt_key  (kvmsg, "%d", randof (10000));
        kvmsg_fmt_body (kvmsg, "%d", randof (1000000));
        kvmsg_send     (kvmsg, publisher);
        kvmsg_store   (&kvmsg, kvmap);
    }
    printf (" Interrupted\n%d messages out\n", (int) sequence);
    zhash_destroy (&kvmap);
    zctx_destroy (&ctx);
    return 0;
}

zactor_t *
zactor_new (zactor_fn *actor, void *args)
{
    zactor_t *self = (zactor_t *) zmalloc (sizeof (zactor_t));
    if (!self)
        return NULL;
    self->tag = ZACTOR_TAG;

    shim_t *shim = (shim_t *) zmalloc (sizeof (shim_t));
    if (!shim)
        return NULL;

    //  Create front-to-back pipe pair
    self->pipe = zsock_new (ZMQ_PAIR);
    assert (self->pipe);
    char endpoint [32];
    while (true) {
        sprintf (endpoint, "inproc://zactor-%04x-%04x\n",
                 randof (0x10000), randof (0x10000));
        if (zsock_bind (self->pipe, "%s", endpoint) == 0)
            break;
    }
    shim->pipe = zsock_new (ZMQ_PAIR);
    assert (shim->pipe);
    int rc = zsock_connect (shim->pipe, "%s", endpoint);
    assert (rc != -1);

    shim->handler = actor;
    shim->args = args;

#if defined (__UNIX__)
    pthread_t thread;
    pthread_create (&thread, NULL, s_thread_shim, shim);
    pthread_detach (thread);

#elif defined (__WINDOWS__)
    HANDLE handle = (HANDLE) _beginthreadex (
        NULL,                   //  Handle is private to this process
        0,                      //  Use a default stack size for new thread
        &s_thread_shim,         //  Start real thread function via this shim
        shim,                   //  Which gets arguments shim
        CREATE_SUSPENDED,       //  Set thread priority before starting it
        NULL);                  //  We don't use the thread ID
    assert (handle);

    //  Set child thread priority to same as current
    int priority = GetThreadPriority (GetCurrentThread ());
    SetThreadPriority (handle, priority);

    //  Start thread & release resources
    ResumeThread (handle);
    CloseHandle (handle);
#endif

    //  Mandatory handshake for new actor so that constructor returns only
    //  when actor has also initialized. This eliminates timing issues at
    //  application start up.
    zsock_wait (self->pipe);
    return self;
}

zyre_t *
zyre_new (const char *name)
{
    zyre_t *self = (zyre_t *) zmalloc (sizeof (zyre_t));
    assert (self);

    //  Create front-to-back pipe pair for data traffic
    self->inbox = zsock_new (ZMQ_PAIR);
    assert (self->inbox);
    char endpoint [32];
    while (true) {
        sprintf (endpoint, "inproc://zyre-%04x-%04x\n",
                 randof (0x10000), randof (0x10000));
        if (zsock_bind (self->inbox, "%s", endpoint) == 0)
            break;
    }
    //  Create other half of traffic pipe
    zsock_t *outbox = zsock_new_pair (endpoint);
    assert (outbox);
    
    //  Start node engine and wait for it to be ready
    self->actor = zactor_new (zyre_node_actor, outbox);
    assert (self->actor);

    //  Send name, if any, to node ending
    if (name)
        zstr_sendx (self->actor, "SET NAME", name, NULL);
    
    return self;
}

int main()
{
    zctx_t* ctx = zctx_new();
    void* server = zsocket_new(ctx, ZMQ_REP);
    zsocket_bind(server, SERVER_ENDPOINT);
    srandom((unsigned int)time(0));

    int cycles = 0;
    while (1) {
        char* request = zstr_recv(server);
        cycles++;

        // Simulate various problems, after a few cycles
        if (cycles > 3 && randof(3) == 0) {
            printf("I: simulating a crash\n");
            break;
        } else if (cycles > 3 && randof(3) == 0) {
            printf("I: simulating CPU overload\n");
            msleep(2000);
        }
        printf("I: normal request (%s)\n", request);
        msleep(1000);
        zstr_send(server, request);
        free(request);
    }

    zctx_destroy(&ctx);
    return 0;
}

static void
server_worker (void *args, zctx_t *ctx, void *pipe)
{
    void *worker = zsocket_new (ctx, ZMQ_DEALER);
    zsocket_connect (worker, "inproc://backend");

    while (true) {
        //  The DEALER socket gives us the reply envelope and message
        zmsg_t *msg = zmsg_recv (worker);
        zframe_t *identity = zmsg_pop (msg);
        zframe_t *content = zmsg_pop (msg);
        assert (content);
        zmsg_destroy (&msg);

        //  Send 0..4 replies back
        int reply, replies = randof (5);
        for (reply = 0; reply < replies; reply++) {
            //  Sleep for some fraction of a second
            zclock_sleep (randof (1000) + 1);
            zframe_send (&identity, worker, ZFRAME_REUSE + ZFRAME_MORE);
            zframe_send (&content, worker, ZFRAME_REUSE);
        }
        zframe_destroy (&identity);
        zframe_destroy (&content);
    }
}

static void s_set_id (void *socket)
{
	GET_OBJECT_VOID(ZmqMgr, iZmqMgr);
    char identity [10];
    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));
    iZmqMgr->Setsockopt(socket, LWDP_IDENTITY, identity, strlen (identity));
}

static zframe_t *
s_upstream_create_content (upstream_t *self)
{
    int msgsize = self->size + randof (self->variance) - randof (self->variance);
    zframe_t *content = zframe_new (NULL, msgsize);
    return content;
}

int main (void)
{
    //  Prepare our context and publisher
    void *context = zmq_ctx_new ();
    void *publisher = zmq_socket (context, ZMQ_PUSH);
    //int rc = zmq_bind (publisher, "tcp://192.168.0.254:5556");
    int rc = zmq_connect (publisher, "tcp://192.168.0.254:5559");
   assert (rc == 0);

    //  Initialize random number generator
    srandom ((unsigned) time (NULL));
    while (1) {
        //  Get values that will fool the boss
        int zipcode, temperature, relhumidity;
        zipcode     = randof (100000);
        temperature = randof (215) - 80;
        relhumidity = randof (50) + 10;

        //  Send message to all subscribers
        char update [20];
        //sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);
        sprintf (update, "hello %05d %d %d", zipcode, temperature, relhumidity);

		printf(update);
	    int size = zmq_send (publisher, update, strlen (update), 0);

        //s_send (publisher, update);
    }
    zmq_close (publisher);
    zmq_ctx_destroy (context);
    return 0;
}

int main () {
  // Prepare our context and publisher
  void* context = zmq_ctx_new();
  void* publisher = zmq_socket(context, ZMQ_PUB);
  int rc = zmq_bind(publisher, "tcp://*:5556");
  assert(rc == 0);
  rc = zmq_bind(publisher, "ipc://weather.ipc");
  assert(rc == 0);

  // Initialize rng
  srandom((unsigned) time(NULL));
  while (1) {
    // Get values that will fool the boss
    int zipcode, temperature, relhumidity;
    zipcode = randof(100000);
    temperature = randof(215) - 80;
    relhumidity = randof(50) + 10;

    // Send message to all subscribers
    char update[20];
    sprintf(update, "%05d %d %d", zipcode, temperature, relhumidity);
    s_send(publisher, update);
  }
  zmq_close(publisher);
  zmq_ctx_destroy(context);
}

int main (void)
{
    //  Prepare our context and sockets
    zctx_t *ctx = zctx_new ();
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (publisher, "tcp://*:5557");

    int64_t sequence = 0;
    srandom ((unsigned) time (NULL));

    //  Start state manager and wait for synchronization signal
    void *updates = zthread_fork (ctx, state_manager, NULL);
    free (zstr_recv (updates));

    while (!zctx_interrupted) {
        //  Distribute as key-value message
        kvmsg_t *kvmsg = kvmsg_new (++sequence);
        kvmsg_fmt_key  (kvmsg, "%d", randof (10000));
        kvmsg_fmt_body (kvmsg, "%d", randof (1000000));
        kvmsg_send     (kvmsg, publisher);
        kvmsg_send     (kvmsg, updates);
        kvmsg_destroy (&kvmsg);
    }
    printf (" Interrupted\n%d messages out\n", (int) sequence);
    zctx_destroy (&ctx);
    return 0;
}

int main (void)
{
  printf ("Connecting to math server…\n");
  void *context = zmq_ctx_new ();
  void *requester = zmq_socket (context, ZMQ_REQ);
  zmq_connect (requester, "tcp://localhost:5555");
  srand(time(NULL));

  int request_nbr;
  for (request_nbr = 0; request_nbr < 1000; request_nbr++) {
    char buffer [10];
    int a = randof(30);
    int b = randof(30);
    printf ("Sending (%d, %d)…\n", a, b);
    char to_send[100];
    sprintf(to_send, "%d %d", a, b);
    s_send(requester, to_send);
    char* ret = s_recv(requester);
    int sum = atoi(ret);
    printf("%d + %d = %d\n", a, b, sum);
    free(ret);
  }
  zmq_close (requester);
  zmq_ctx_destroy (context);
  return 0;
}

static void *
client_task (void *args)
{
    zctx_t *ctx = zctx_new ();
    void *client = zsocket_new (ctx, ZMQ_DEALER);

    //  Set random identity to make tracing easier
    char identity [10];
    sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));
    zsocket_set_identity (client, identity);
    zsocket_connect (client, "tcp://localhost:5570");

    zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
    int request_nbr = 0;
    while (true) {
        //  Tick once per second, pulling in arriving messages
        int centitick;
        for (centitick = 0; centitick < 100; centitick++) {
            zmq_poll (items, 1, 10 * ZMQ_POLL_MSEC);
            if (items [0].revents & ZMQ_POLLIN) {
                zmsg_t *msg = zmsg_recv (client);
                zframe_print (zmsg_last (msg), identity);
                zmsg_destroy (&msg);
            }
        }
        zstr_send (client, "request #%d");
    }
    zctx_destroy (&ctx);
    return NULL;
}

int main (int argc, char *argv [])
{
    //  Get number of nodes N to simulate
    //  We need 3 x N x N + 3N file handles
    int max_nodes = 10;
    int nbr_nodes = 0;
    if (argc > 1)
        max_nodes = atoi (argv [1]);
    assert (max_nodes);

    int max_iterations = -1;
    int nbr_iterations = 0;
    if (argc > 2)
        max_iterations = atoi (argv [2]);

    //  Our gossip network will use one fixed hub (not a Zyre node),
    //  to which all nodes will connect
    zactor_t *hub = zactor_new (zgossip, "hub");
    zstr_sendx (hub, "BIND", "inproc://zyre-hub", NULL);
        
    //  We address nodes as an array of actors
    zactor_t **actors = (zactor_t **) zmalloc (sizeof (zactor_t *) * max_nodes);

    //  We will randomly start and stop node threads
    uint index;
    while (!zsys_interrupted) {
        index = randof (max_nodes);
        //  Toggle node thread
        if (actors [index]) {
            zactor_destroy (&actors [index]);
            actors [index] = NULL;
            zsys_info ("stopped node (%d running)", --nbr_nodes);
        }
        else {
            char node_name [10];
            sprintf (node_name, "node-%d", index);
            actors [index] = zactor_new (node_actor, strdup (node_name));
            zsys_info ("started node (%d running)", ++nbr_nodes);
        }
        nbr_iterations++;
        if (max_iterations > 0 && nbr_iterations >= max_iterations)
            break;
        //  Sleep ~300 msecs randomly so we smooth out activity
        zclock_sleep (randof (100) + 100);
    }
    zsys_info ("stopped tester (%d iterations)", nbr_iterations);

    //  Stop all remaining actors
    for (index = 0; index < max_nodes; index++) {
        if (actors [index])
            zactor_destroy (&actors [index]);
    }
    free (actors);
    
    zactor_destroy (&hub);
    return 0;
}

static void
publisher_thread (void *args, zctx_t *ctx, void *pipe)
{
    void *publisher = zsocket_new (ctx, ZMQ_PUB);
    zsocket_bind (publisher, "tcp://*:6000");

    while (!zctx_interrupted) {
        char string [10];
        sprintf (string, "%c-%05d", randof (10) + 'A', randof (100000));
        if (zstr_send (publisher, string) == -1)
            break;              //  Interrupted
        zclock_sleep (100);     //  Wait for 1/10th second
    }
}

int main (int argc, char *argv [])
{
    //  Initialize context for talking to tasks
    zctx_t *ctx = zctx_new ();
    zctx_set_linger (ctx, 100);
    
    //  Get number of nodes N to simulate
    //  We need 3 x N x N + 3N file handles
    int max_nodes = 10;
    int nbr_nodes = 0;
    if (argc > 1)
        max_nodes = atoi (argv [1]);

    int max_iterations = -1;
    int nbr_iterations = 0;
    if (argc > 2)
        max_iterations = atoi (argv [2]);

    //  We address nodes as an array of pipes
    void **pipes = zmalloc (sizeof (void *) * max_nodes);

    //  We will randomly start and stop node threads
    while (!zctx_interrupted) {
        uint index = randof (max_nodes);
        //  Toggle node thread
        if (pipes [index]) {
            zstr_send (pipes [index], "STOP");
            zsocket_destroy (ctx, pipes [index]);
            pipes [index] = NULL;
            zclock_log ("I: Stopped node (%d running)", --nbr_nodes);
        }
        else {
            pipes [index] = zthread_fork (ctx, node_task, NULL);
            zclock_log ("I: Started node (%d running)", ++nbr_nodes);
        }
        nbr_iterations++;
        if (max_iterations > 0 && nbr_iterations >= max_iterations)
            break;
        //  Sleep ~750 msecs randomly so we smooth out activity
        zclock_sleep (randof (500) + 500);
    }
    zclock_log ("I: Stopped tester (%d iterations)", nbr_iterations);
    
    //  Does not actually terminate properly... :-/
    //  zctx_destroy (&ctx);
    free (pipes);
    return 0;
}

client_state* client_state_init(const char* server_url, const char* login)
{
    srand(time(0));
    client_state * state = new client_state();
    state->last_message_id = 0;
    state->context = zctx_new ();
    state->login = login;
    
    // Set random identity to make tracing easier
    sprintf (state->identity, "%04X-%04X", randof (0x10000), randof (0x10000));
    
    state->server_url = server_url;
    
    client_state_reset_heartbeat(state);
    return state;
}

static void *
worker_task(void *args) {
    void *context = zmq_init(1);
    void *worker = zmq_socket(context, ZMQ_REQ);

    // s_set_id()函数会根据套接字生成一个可打印的字符串，
    // 并以此作为该套接字的标识。
    s_set_id(worker);
    zmq_connect(worker, "ipc://routing.ipc");

    int total = 0;
    while (1) {
        // 告诉ROUTER我已经准备好了
        s_send(worker, "ready");

        // 从ROUTER中获取工作，直到收到结束的信息
        char *workload = s_recv(worker);
        int finished = (strcmp(workload, "END") == 0);
        free(workload);
        if (finished) {
            printf("Processed: %d tasks\n", total);
            break;
        }
        total++;

        // 随机等待一段时间
        s_sleep(randof(1000) + 1);
    }
    zmq_close(worker);
    zmq_term(context);
    return NULL;
}

int main (void)
{
    void *context = zmq_ctx_new ();
    void *worker = zmq_socket (context, ZMQ_DEALER);
    s_set_id (worker);          //  Set a printable identity
    zmq_connect (worker, "tcp://localhost:5671");

    int total = 0;
    while (1) {
        //  Tell the broker we're ready for work
        s_sendmore (worker, "");
        s_send (worker, "Hi Boss");

        //  Get workload from broker, until finished
        // free (s_recv (worker));     //  Envelope delimiter
        // char *workload = s_recv (worker);
        // int finished = (strcmp (workload, "Fired!") == 0);
        // free (workload);
        // if (finished) {
            // printf ("Completed: %d tasks\n", total);
            // break;
        // }
        total++;

        //  Do some random work
        s_sleep (randof (500) + 1);
    }
    zmq_close (worker);
    zmq_ctx_destroy (context);
    return 0;
}

int main (int argc, char *argv [])
{
    zctx_t *context = zctx_new ();
    void *publisher = zsocket_new (context, ZMQ_PUB);
    if (argc == 2)
        zsocket_connect (publisher, argv [1]);
    else
        zsocket_bind (publisher, "tcp://*:5556");

    //  Ensure subscriber connection has time to complete
    sleep (1);

    //  Send out all 1,000 topic messages
    int topic_nbr;
    for (topic_nbr = 0; topic_nbr < 1000; topic_nbr++) {
        zstr_sendm (publisher, "%03d", topic_nbr, ZMQ_SNDMORE);
        zstr_send (publisher, "Save Roger");
    }
    //  Send one random update per second
    srandom ((unsigned) time (NULL));
    while (!zctx_interrupted) {
        sleep (1);
        zstr_sendm (publisher, "%03d", randof (1000), ZMQ_SNDMORE);
        zstr_send (publisher, "Off with his head!");
    }
    zctx_destroy (&context);
    return 0;
}

static void
test_tcp_pub (void *args, zctx_t *ctx, void *pipe)
{
    vtx_t *vtx = vtx_new (ctx);
    int rc = vtx_tcp_load (vtx, FALSE);
    assert (rc == 0);
    char *port = zstr_recv (pipe);

    //  Create publisher socket and bind to all network interfaces
    void *publisher = vtx_socket (vtx, ZMQ_PUB);
    assert (publisher);
    rc = vtx_bind (vtx, publisher, "tcp://*:%s", port);
    assert (rc == 0);
    int sent = 0;

    while (!zctx_interrupted) {
        zstr_sendf (publisher, "NOM %04x", randof (0x10000));
        sent++;
        char *end = zstr_recv_nowait (pipe);
        if (end) {
            free (end);
            zstr_send (pipe, "OK");
            break;
        }
    }
    zclock_log ("I: PUB: sent=%d", sent);
    free (port);
    vtx_destroy (&vtx);
}