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

//  --------------------------------------------------------------------------//  Self test of this classvoidzclock_test (bool verbose){    printf (" * zclock: ");    //  @selftest    int64_t start = zclock_time ();    zclock_sleep (10);    assert ((zclock_time () - start) >= 10);    start = zclock_mono ();    int64_t usecs = zclock_usecs ();    zclock_sleep (10);    assert ((zclock_mono () - start) >= 10);    assert ((zclock_usecs () - usecs) >= 10000);    char *timestr = zclock_timestr ();    if (verbose)        puts (timestr);    freen (timestr);#if defined (__WINDOWS__)    zsys_shutdown();#endif    //  @end    printf ("OK/n");}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);    //  1. Send 'echo' request to Titanic    zmsg_t *request = zmsg_new ();    zmsg_addstr (request, "echo");    zmsg_addstr (request, "Hello world");    zmsg_t *reply = s_service_call (        session, "titanic.request", &request);    zframe_t *uuid = NULL;    if (reply) {        uuid = zmsg_pop (reply);        zmsg_destroy (&reply);        zframe_print (uuid, "I: request UUID ");    }    //  2. Wait until we get a reply    while (!zctx_interrupted) {        zclock_sleep (100);        request = zmsg_new ();        zmsg_add (request, zframe_dup (uuid));        zmsg_t *reply = s_service_call (            session, "titanic.reply", &request);        if (reply) {            char *reply_string = zframe_strdup (zmsg_last (reply));            printf ("Reply: %s/n", reply_string);            free (reply_string);            zmsg_destroy (&reply);            //  3. Close request            request = zmsg_new ();            zmsg_add (request, zframe_dup (uuid));            reply = s_service_call (session, "titanic.close", &request);            zmsg_destroy (&reply);            break;        }        else {            printf ("I: no reply yet, trying again.../n");            zclock_sleep (5000);     //  Try again in 5 seconds        }    }    zframe_destroy (&uuid);    mdcli_destroy (&session);    return 0;}

intzsock_connect (zsock_t *self, const char *format, ...){    assert (self);    assert (zsock_is (self));    //  Expand format to get full endpoint    va_list argptr;    va_start (argptr, format);    char *endpoint = zsys_vprintf (format, argptr);    va_end (argptr);    int rc = zmq_connect (self->handle, endpoint);    #if (ZMQ_VERSION < ZMQ_MAKE_VERSION (4,0,0))    int retries = 4;    while (rc == -1 && zmq_errno () == ECONNREFUSED && retries) {        //  This bruteforces a synchronization between connecting and        //  binding threads on ZeroMQ v3.2 and earlier, where the bind        //  MUST happen before the connect on inproc transports.        zclock_sleep (250);        rc = zmq_connect (self->handle, endpoint);        retries--;    }#endif    free (endpoint);    return rc;}

intzthread_test (Bool verbose){    printf (" * zthread: ");    //  @selftest    zctx_t *ctx = zctx_new ();    //  Create a detached thread, let it run    zthread_new (ctx, s_test_detached, NULL);    zclock_sleep (100);    //  Create an attached thread, check it's safely alive    void *pipe = zthread_fork (ctx, s_test_attached, NULL);    zstr_send (pipe, "ping");    char *pong = zstr_recv (pipe);    assert (streq (pong, "pong"));    free (pong);    //  Everything should be cleanly closed now    zctx_destroy (&ctx);    //  @end    printf ("OK/n");    return 0;}

//  Checks whether client can connect to serverstatic bools_can_connect (zctx_t *ctx, void **server, void **client){    int port_nbr = zsocket_bind (*server, "tcp://127.0.0.1:*");    assert (port_nbr > 0);    int rc = zsocket_connect (*client, "tcp://127.0.0.1:%d", port_nbr);    assert (rc == 0);    //  Give the connection time to fail if that's the plan    zclock_sleep (200);    //  By default PUSH sockets block if there's no peer    zsock_set_sndtimeo (*server, 200);    zstr_send (*server, "Hello, World");    zpoller_t *poller = zpoller_new (*client, NULL);    bool success = (zpoller_wait (poller, 400) == *client);    zpoller_destroy (&poller);    zsocket_destroy (ctx, *client);    zsocket_destroy (ctx, *server);    *server = zsocket_new (ctx, ZMQ_PUSH);    assert (*server);    *client = zsocket_new (ctx, ZMQ_PULL);    assert (*client);    return success;}

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 interfaces to simulate, default 100    int max_interface = 100;    int nbr_interface = 0;    if (argc > 1)        max_interface = atoi (argv [1]);    //  We address interfaces as an array of pipes    void **pipes = zmalloc (sizeof (void *) * max_interface);    for (nbr_interface = 0; nbr_interface < max_interface; nbr_interface++) {        pipes [nbr_interface] = zthread_fork (ctx, interface_task, NULL);        zclock_log ("I: Started interface (%d running)", nbr_interface + 1);    }    //  We will randomly start and stop interface threads    while (!zctx_interrupted) {        zclock_sleep (1000);    }    zclock_log ("I: Stopped perl_remote");    zctx_destroy (&ctx);    free (pipes);    return 0;}

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

// Basic request-reply client using REQ socket//static void *client_task(void *args){	zctx_t *ctx = zctx_new();	void *client = zsocket_new(ctx, ZMQ_REQ);#if (defined (WIN32))	zsocket_connect(client, "tcp://localhost:5672"); // frontend#else	zsocket_connect(client, "ipc://frontend.ipc");#endif	// Send request, get reply	while (1) {		zstr_send(client, "HELLO");		char *reply = zstr_recv(client);		if (!reply) {			break;		}		printf("Client: %s/n", reply);		free(reply);		zclock_sleep(1);	}	zctx_destroy(&ctx);	return NULL;}

static voidclient_task (void *args, zctx_t *ctx, void *pipe){    void *client = zsocket_new (ctx, ZMQ_DEALER);    zsocket_connect (client, "tcp://localhost:5555");    printf ("Setting up test.../n");    zclock_sleep (100);    int requests;    int64_t start;    printf ("Synchronous round-trip test.../n");    start = zclock_time ();    for (requests = 0; requests < 10000; requests++) {        zstr_send (client, "hello");        char *reply = zstr_recv (client);        free (reply);    }    printf (" %d calls/second/n",        (1000 * 10000) / (int) (zclock_time () - start));    printf ("Asynchronous round-trip test.../n");    start = zclock_time ();    for (requests = 0; requests < 100000; requests++)        zstr_send (client, "hello");    for (requests = 0; requests < 100000; requests++) {        char *reply = zstr_recv (client);        free (reply);    }    printf (" %d calls/second/n",        (1000 * 100000) / (int) (zclock_time () - start));    zstr_send (pipe, "done");}

void send_outgoing_messages(client_state* state, void * socket){    for(zchat_message_vector_t::iterator         it = state->out_messages.begin();        it != state->out_messages.end();        it++)    {        zchat_string_t serialised;        zchat_message * message = *it;                serialize_message_to_string(message, &serialised);        zframe_t* content = zframe_new (serialised.c_str(),                                        serialised.length());                zclock_sleep (randof (1000) + 1);                zframe_send (&content, socket, ZFRAME_REUSE);        if(message->type() == zchat_message_message_type_PING)        {            client_state_set_heartbeat_time(state);        }                zframe_destroy (&content);        zchat_message_destroy(message);    }        state->out_messages.clear();}

static intzyre_node_stop (zyre_node_t *self){    if (self->beacon) {        //  Stop broadcast/listen beacon        beacon_t beacon;        beacon.protocol [0] = 'Z';        beacon.protocol [1] = 'R';        beacon.protocol [2] = 'E';        beacon.version = BEACON_VERSION;        beacon.port = 0;            //  Zero means we're stopping        zuuid_export (self->uuid, beacon.uuid);        zsock_send (self->beacon, "sbi", "PUBLISH",            (byte *) &beacon, sizeof (beacon_t), self->interval);        zclock_sleep (1);           //  Allow 1 msec for beacon to go out        zpoller_remove (self->poller, self->beacon);        zactor_destroy (&self->beacon);    }    //  Stop polling on inbox    zpoller_remove (self->poller, self->inbox);    zstr_sendm (self->outbox, "STOP");    zstr_sendm (self->outbox, zuuid_str (self->uuid));    zstr_send (self->outbox, self->name);    return 0;}

static voidserver_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 bool s_heart_hid_printer(ticket_printer_t *self) {    if ((zclock_time() - self->hid_status_at) > 5000) {#ifdef __WIN32__        char *command = strdup("<S1>");        int ret = boca_hid_write(self->hid_printer, command, strlen(command));        free(command);        enum BOCA_STATUS status = BOCA_OFFLINE;        if (ret > 0) {            zclock_sleep(500);            status = boca_hid_status(self->hid_printer);            self->event = EVENT_BOCA_WRITE_OK;            s_state_machine(self);        } else {            self->event = EVENT_BOCA_WRITE_ERROR;            s_state_machine(self);            s_handle_boca_status(self, &status, true);        }#else        enum BOCA_FEATURE_REPORT_STATUS feature_status = boca_hid_status_feature(self->hid_printer);        enum BOCA_STATUS status = boca_status_from_feature(feature_status);#endif        s_handle_boca_status(self, &status, true);        return true;    }    return false;}

voiddrops_test (bool verbose){    printf (" * drops: ");    //  @selftest    //  Create temporary directory for test files#   define TESTDIR ".test_drops"    //  Create two drops instances and test some to and fro    zsys_dir_create (TESTDIR "/box1");    drops_t *drops1 = drops_new (TESTDIR "/box1");    zsys_dir_create (TESTDIR "/box2");    drops_t *drops2 = drops_new (TESTDIR "/box2");    //  Give time for nodes to discover each other and interconnect    zclock_sleep (100);    FILE *output = fopen (TESTDIR "/box1/bilbo", "w");    fprintf (output, "Hello, world");    fclose (output);    //  Directory monitoring is once per second at present, so this gives    //  time for box1 to see its new file, and send it to box2    zclock_sleep (1200);    char buffer [256];    FILE *input = fopen (TESTDIR "/box2/bilbo", "r");    assert (input);    char *result = fgets (buffer, 256, input);    assert (result == buffer);    assert (streq (buffer, "Hello, world"));    fclose (input);    drops_destroy (&drops2);    drops_destroy (&drops1);    //  Delete all test files    zdir_t *dir = zdir_new (TESTDIR, NULL);    zdir_remove (dir, true);    zdir_destroy (&dir);    //  @end    zclock_sleep (100);    printf ("OK/n");}

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

int main (void){    zthread_new (client_task, NULL);    zthread_new (client_task, NULL);    zthread_new (client_task, NULL);    zthread_new (server_task, NULL);    zclock_sleep (5 * 1000);    //  Run for 5 seconds then quit    return 0;}

//  atexit or manual termination for the processvoidzsys_shutdown (void){    if (!s_initialized) {        return;    }    s_initialized = false;    //  The atexit handler is called when the main function exits;    //  however we may have zactor threads shutting down and still    //  trying to close their sockets. So if we suspect there are    //  actors busy (s_open_sockets > 0), then we sleep for a few    //  hundred milliseconds to allow the actors, if any, to get in    //  and close their sockets.    ZMUTEX_LOCK (s_mutex);    size_t busy = s_open_sockets;    ZMUTEX_UNLOCK (s_mutex);    if (busy)        zclock_sleep (200);    //  No matter, we are now going to shut down    //  Print the source reference for any sockets the app did not    //  destroy properly.    ZMUTEX_LOCK (s_mutex);    s_sockref_t *sockref = (s_sockref_t *) zlist_pop (s_sockref_list);    while (sockref) {        assert (sockref->filename);        zsys_error ("dangling '%s' socket created at %s:%d",                    zsys_sockname (sockref->type),                    sockref->filename, (int) sockref->line_nbr);        zmq_close (sockref->handle);        free (sockref);        sockref = (s_sockref_t *) zlist_pop (s_sockref_list);    }    zlist_destroy (&s_sockref_list);    ZMUTEX_UNLOCK (s_mutex);    //  Close logsender socket if opened (don't do this in critical section)    if (s_logsender) {        zsys_close (s_logsender, NULL, 0);        s_logsender = NULL;    }    if (s_open_sockets == 0)        zmq_term (s_process_ctx);    else        zsys_error ("dangling sockets: cannot terminate ZMQ safely");    ZMUTEX_DESTROY (s_mutex);    //  Free dynamically allocated properties    free (s_interface);    free (s_logident);#if defined (__UNIX__)    closelog ();                //  Just to be pedantic#endif}

int main() {    zactor_t *actor = zactor_new (s_alerts, NULL);    //XXX: this is UGLY    while (!zsys_interrupted) {        zclock_sleep(100);    }    zactor_destroy (&actor);}

voidzgtask_worker_start (zgtask_worker_t *self){    assert (self);    zyre_t *zyre = zgtask_net_init_zyre_parent (self->net);    zyre_gossip_connect (zyre, self->url_parent);    zyre_start (zyre);    zyre_join (zyre, "GLOBAL");    zclock_sleep (250);}

int _tmain(int argc, _TCHAR* argv[]){	zctx_t* context = zctx_new();	titanic_publish* component = new titanic_publish(context,TADD_COMP,TADD_PUBSUB,100*1000,100*1000);	//Lets give the broker time to get bound up and set up all the pollers on its sockets	zclock_sleep(1000);	cout << "Starting Publisher";	component->Start();	delete component;	return NULL;}

intzfile_test (bool verbose){    printf (" * zfile: ");    //  @selftest    zfile_t *file = zfile_new (".", "bilbo");    assert (streq (zfile_filename (file, "."), "bilbo"));    assert (zfile_is_readable (file) == false);    zfile_destroy (&file);    //  Create a test file in some random subdirectory    file = zfile_new ("./this/is/a/test", "bilbo");    int rc = zfile_output (file);    assert (rc == 0);    zchunk_t *chunk = zchunk_new (NULL, 100);    zchunk_fill (chunk, 0, 100);    //  Write 100 bytes at position 1,000,000 in the file    rc = zfile_write (file, chunk, 1000000);    assert (rc == 0);    zfile_close (file);    assert (zfile_is_readable (file));    assert (zfile_cursize (file) == 1000100);    assert (!zfile_is_stable (file));    zchunk_destroy (&chunk);    zclock_sleep (1001);    assert (zfile_is_stable (file));    //  Check we can read from file    rc = zfile_input (file);    assert (rc == 0);    chunk = zfile_read (file, 1000100, 0);    assert (chunk);    assert (zchunk_size (chunk) == 1000100);    zchunk_destroy (&chunk);    //  Remove file and directory    zdir_t *dir = zdir_new ("./this", NULL);    assert (zdir_cursize (dir) == 1000100);    zdir_remove (dir, true);    assert (zdir_cursize (dir) == 0);    zdir_destroy (&dir);    //  Check we can no longer read from file    assert (!zfile_is_readable (file));    rc = zfile_input (file);    assert (rc == -1);    zfile_destroy (&file);    //  @end    printf ("OK/n");    return 0;}

//int server_process(char *server_id, char *servers_str, char *port, char *init_data, Server_Status *_status)int server_process(Server_Args *server_args, Server_Status *_status) {    status = _status;    zthread_new(server_task, (void *)server_args);    printf("Starting the thread id %s/n", server_args->server_id);    while(true) {        zclock_sleep(60*600*1000);    }    free(server_args);    return 0;}

int _tmain(int argc, _TCHAR* argv[]){	zclock_sleep(20000);	char* broker_loc = "tcp://localhost:5555";	zctx_t* ctx = zctx_new();	void* scket = zsocket_new(ctx,ZMQ_REQ);	zsocket_connect(scket,broker_loc);	zmsg_t* msg = zmsg_new();	zmsg_addstr(msg,TWRK_CLI_VER);	zmsg_addstr(msg,"Echo");	zmsg_addstr(msg,TMSG_TYPE_REQUEST);		int64_t sleep = 10*1000;	zclock_sleep(sleep);	zmsg_add(msg,zframe_new(&sleep,sizeof(sleep)));	zmsg_send(&msg,scket);	zmsg_t* reply =  zmsg_recv(scket);	zmsg_dump(reply);	return 0;}

int main (int argc, char *argv []){	lsd_handle_t* handle = lsd_init(NULL, info_callback, NULL);	lsd_join(handle, GROUP_0);	lsd_join(handle, GROUP_1);	zclock_sleep (500);	lsd_shout(handle, GROUP_0, (const uint8_t*)MSG_0,strlen(MSG_0));	zclock_sleep (500);	lsd_shout(handle, GROUP_1, (const uint8_t*)MSG_1,strlen(MSG_1));	zclock_sleep (500);	lsd_publish(handle, "core");	zclock_sleep (100000);	lsd_leave(handle, GROUP_0);	lsd_leave(handle, GROUP_1);	lsd_destroy(handle);	return 0;}

static voidzyre_node_stop (zyre_node_t *self){    //  Set broadcast/listen beacon    beacon_t beacon;    beacon.protocol [0] = 'Z';    beacon.protocol [1] = 'R';    beacon.protocol [2] = 'E';    beacon.version = BEACON_VERSION;    beacon.port = 0;            //  Zero means we're stopping    zuuid_export (self->uuid, beacon.uuid);    zbeacon_publish (self->beacon, (byte *) &beacon, sizeof (beacon_t));    zclock_sleep (1);           //  Allow 1 msec for beacon to go out}

int main(int argc, char **argv) {    zmon_gtop_t *self = zmon_gtop_new ("");    assert (self);    int n = 5;    for (int i = 0; i < n; i++) {        zmon_gtop_update (self);//        zmon_gtop_dump (self);        zclock_sleep (1000);    }    zmon_gtop_destroy (&self);    return 0;}

static voidpublisher_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    }}

staticzsock_t* parser_push_socket_new(){    zsock_t *socket = zsock_new(ZMQ_PUSH);    assert(socket);    int rc;    // connect socket, taking thread startup time into account    // TODO: this is a hack. better let controller coordinate this    for (int i=0; i<10; i++) {        rc = zsock_connect(socket, "inproc://graylog-forwarder-writer");        if (rc == 0) break;        zclock_sleep(100);    }    return socket;}

void test_integrate_components (){    printf ("Integration Test: ");    agent = zsync_agent_new ();    zsync_agent_set_pass_update (agent, pass_update);    zsync_agent_set_pass_chunk (agent, pass_chunk);    zsync_agent_set_get_update (agent, get_update);    zsync_agent_set_get_current_state (agent, get_current_state);    zsync_agent_set_get_chunk (agent, get_chunk);    zsync_agent_start (agent);    zlist_t *files;    while (zsync_agent_running (agent)) {        zclock_sleep (25000);        DIR *dir;        struct dirent *ent;        int count = -2;        if ((dir = opendir ("./syncfolder")) != NULL) {            while ((ent = readdir (dir)) != NULL) {                count++;             }        }        if (count == 2)            break;    }    zsync_agent_stop (agent);            zclock_sleep (500);    zsync_agent_destroy (&agent);        printf ("OK/n");}