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

triggerconfig_t * add_trigger(rulepackage_t * rpkg, zmsg_t * request,zmsg_t * reply) {    char * rule_id  = zmsg_popstr(request);    zclock_log("new trigger!");        if (remove_rule(rpkg->context, rpkg->triggers, rule_id)) {        // already have a rule with that id!       zclock_log("Received duplicate rule %s - killing old trigger", rule_id);    }           triggerconfig_t * tconf = malloc(sizeof(triggerconfig_t));    tconf->base_config = rpkg->base_config;    create_triggerconfig(tconf, request, rpkg->channel, rule_id);          // when we create this trigger, what guarantee do we have that    // the line listener is active? FIX    // basically we have none. crap.    char * created = create_trigger(rpkg->triggers, rule_id, rpkg->context, tconf);    if(NULL == created) {      // happy path, so add to db      zmsg_pushstr(reply, "ok");    } else {      zclock_log("create_trigger failed: %s", created);      free(tconf);      tconf=NULL;      zmsg_pushstr(reply, created);    }    free(created);    return tconf;        }

int port_changed(char * channel, channel_memory_t * m) {  // zclock_log("line/nchannel: %s/nchan_memory:%s/nnext_chan:%s",  //            channel, m->current_channel, m->next_channel);  if (strcmp(channel, m->current_channel)==0) {    return 0;  }  if(m->next_channel && strcmp(channel, m->next_channel)==0) {    if(m->confirmed_rounds++ > LIMIT) {      // time to change      zclock_log("channel change! %s to %s", m->current_channel, channel);            free(m->next_channel);      free(m->current_channel);      m->next_channel = NULL;      m->confirmed_rounds=0;      m->current_channel = strdup(channel);      return 1;    } else {      zclock_log("tick %s:%s", m->current_channel, channel);      return 0;    }  }  // new channel that we weren't expecting.  free(m->next_channel);  m->next_channel = strdup(channel);  return 0;}

static voids_send_wire (peering_t *self){    vocket_t *vocket = self->vocket;    driver_t *driver = self->driver;    while (TRUE) {        byte *data;        size_t size = vtx_codec_bin_get (self->output, &data);        if (size == 0) {            peering_poller (self, ZMQ_POLLIN);            break;      //  Buffer is empty, stop polling out        }        if (driver->verbose)            zclock_log ("I: (tcp) send %zd bytes to %s",                size, self->address);        int bytes_sent = send (self->handle, data, size, 0);        if (driver->verbose)            zclock_log ("I: (tcp) actually sent %d bytes", bytes_sent);        if (bytes_sent > 0) {            vtx_codec_bin_tick (self->output, bytes_sent);            if (bytes_sent < size)                break;      //  Wait until network can accept more        }        else        if (bytes_sent == 0 || s_handle_io_error ("send") == -1) {            self->exception = TRUE;            break;          //  Signal error and give up        }    }}

int main (int argc, char *argv []){    int verbose = (argc > 1 && streq (argv [1], "-v"));    broker_t *self = s_broker_new (verbose);    s_broker_bind (self, "tcp://*:5555");    //  Get and process messages forever or until interrupted    while (TRUE) {        zmq_pollitem_t items [] = {            { self->socket,  0, ZMQ_POLLIN, 0 } };        int rc = zmq_poll (items, 1, HEARTBEAT_INTERVAL * ZMQ_POLL_MSEC);        if (rc == -1)            break;              //  Interrupted        //  Process next input message, if any        if (items [0].revents & ZMQ_POLLIN) {            zmsg_t *msg = zmsg_recv (self->socket);            if (!msg)                break;          //  Interrupted            if (self->verbose) {                zclock_log ("I: received message:");                zmsg_dump (msg);            }            zframe_t *sender = zmsg_pop (msg);            zframe_t *empty  = zmsg_pop (msg);            zframe_t *header = zmsg_pop (msg);            if (zframe_streq (header, MDPC_CLIENT))                s_client_process (self, sender, msg);            else            if (zframe_streq (header, MDPW_WORKER))                s_worker_process (self, sender, msg);            else {                zclock_log ("E: invalid message:");                zmsg_dump (msg);                zmsg_destroy (&msg);            }            zframe_destroy (&sender);            zframe_destroy (&empty);            zframe_destroy (&header);        }        //  Disconnect and delete any expired workers        //  Send heartbeats to idle workers if needed        if (zclock_time () > self->heartbeat_at) {            s_broker_purge_workers (self);            worker_t *worker = (worker_t *) zlist_first (self->waiting);            while (worker) {                s_worker_send (self, worker, MDPW_HEARTBEAT, NULL, NULL);                worker = (worker_t *) zlist_next (self->waiting);            }            self->heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;        }    }    if (zctx_interrupted)        printf ("W: interrupt received, shutting down.../n");    s_broker_destroy (&self);    return 0;}

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

static ints_peering_activity (zloop_t *loop, zmq_pollitem_t *item, void *arg){    peering_t *peering = (peering_t *) arg;    vocket_t *vocket = peering->vocket;    driver_t *driver = peering->driver;    if (peering->alive) {        if (item->revents & ZMQ_POLLERR) {            if (driver->verbose)                zclock_log ("I: (tcp) peering alive/error %s",                    peering->address);            peering->exception = TRUE;        }        else        if (item->revents & ZMQ_POLLIN) {            if (driver->verbose)                zclock_log ("I: (tcp) peering alive/input %s",                    peering->address);            s_recv_wire (peering);        }        else        if (item->revents & ZMQ_POLLOUT) {            if (driver->verbose)                zclock_log ("I: (tcp) peering alive/output %s",                    peering->address);            s_send_wire (peering);        }    }    else    if (peering->outgoing) {        if (item->revents & ZMQ_POLLERR) {            if (driver->verbose)                zclock_log ("I: (tcp) peering dead/error %s", peering->address);            peering->exception = TRUE;        }        else        if (item->revents & ZMQ_POLLIN        ||  item->revents & ZMQ_POLLOUT) {            if (driver->verbose)                zclock_log ("I: (tcp) peering dead/inout %s", peering->address);            peering_poller (peering, ZMQ_POLLIN);            peering_raise (peering);        }    }    //  Handle exception peering by switching to monitoring, or killing it    if (peering->exception) {        peering_lower (peering);        if (peering->outgoing) {            close (peering->handle);            peering_poller (peering, 0);            peering->handle = 0;            zloop_timer (loop, peering->interval, 1, s_peering_monitor, peering);        }        else            peering_destroy (&peering);    }    return 0;}

int main (int argc, char *argv[]){	int verbose = (argc > 1 && streq(argv[1], "-v"));	broker_t *self = s_broker_new(verbose);	s_broker_bind(self, "tcp://127.0.0.1:5555");	while(true){		zmq_pollitem_t items [] = {			{self->socket, 0, ZMQ_POLLIN, 0}		};		int rc = zmq_poll(items, 1, HEARTBEAT_INTERVAL * ZMQ_POLL_MSEC);		if (rc == -1)			break;		if (items[0].revents & ZMQ_POLLIN){			zmsg_t *msg = zmsg_recv(self->socket);			if (!msg)				break;			if (self->verbose){				zclock_log("I: received message");				zmsg_dump(msg);			}			zframe_t *sender = zmsg_pop(msg);			zframe_t *empty = zmsg_pop(msg);			zframe_t *header = zmsg_pop(msg);			if (zframe_streq(header, MDPC_CLIENT))				s_broker_client_msg(self, sender, msg);			else			if (zframe_streq(header, MDPW_WORKER))				s_broker_worker_msg(self, sender, msg);			else{				zclock_log("E: invalid message:");				zmsg_dump(msg);				zmsg_destroy(&msg);			}			zframe_destroy(&sender);			zframe_destroy(&empty);			zframe_destroy(&header);		}		if (zclock_time() > self->heartbeat_at){			s_broker_purge(self);			worker_t *worker = (worker_t *)zlist_first(self->waiting);			while (worker){				s_worker_send(worker, MDPW_HEARTBEAT, NULL, NULL);				worker = (worker_t *)zlist_next(self->waiting);			}			self->heartbeat_at = zclock_time() + HEARTBEAT_INTERVAL;		}	}	if (zctx_interrupted)		printf("W: interrupt receivde, shutting down.../n");	s_broker_destroy(&self);	return 0;}

static ssize_ts_recv_wire (peering_t *self){    vocket_t *vocket = self->vocket;    driver_t *driver = self->driver;    //  Read into buffer and dump what we got    //  TODO: only read as much as we have space in input queue    //  implement exception strategy here    //  - drop oldest, drop newest, pushback    byte buffer [VTX_TCP_BUFSIZE];    ssize_t size = recv (self->handle, buffer, VTX_TCP_BUFSIZE, MSG_DONTWAIT);    if (size == 0)        //  Other side closed TCP socket, so our peering is down        self->exception = TRUE;    else    if (size == -1) {        if (s_handle_io_error ("recv") == -1)            //  Hard error on socket, so peering is down            self->exception = TRUE;    }    else {        if (driver->verbose)            zclock_log ("I: (tcp) recv %zd bytes from %s",                size, self->address);        int rc = vtx_codec_bin_put (self->input, buffer, size);        assert (rc == 0);        //  store binary data into codec        //  if routing = request        //      retrieve message, if available        //      this is a reply        //      check reply is allowed in state        //      then send message through to msgpipe        //      else discard message        //      reset state machine on peering        //  if routing = reply        //      this is a request        //      if state allows incoming request        //          retrieve message, if available        //          vocket->reply_to = peering        //          send message through to msgpipe        //  if routing = router        //      retrieve message, if available        //      send schemed identity to msgpipe        //      send message through to msgpipe        //  any other routing, nomnom allowed        //      retrieve message, if available        //      send message through to msgpipe#if 0        else            zclock_log ("W: unexpected message from %s - dropping", address);        char *colon = strchr (address, ':');        assert (colon);        *colon = 0;        strcpy (vocket->sender, address);#endif    }

void start_echo(char *arg){	struct timeval tv;	gettimeofday(&tv,NULL);	start = (uint64_t)tv.tv_sec;	//while(!interrupt) 	//{		char *actionid = getActionid();		zmsg_t *msg = create_call(actionid, "worker", "sleep", arg);		if (msg) 		{			zmsg_send(&msg, dealer);			counter ++;			zmsg_destroy(&msg);			zmsg_t *msg = zmsg_recv(dealer);			zframe_t *payload = zmsg_pop(msg);			zmsg_destroy(&msg);			msgpack_unpacked object;			msgpack_unpacked_init(&object);					if (msgpack_unpack_next(&object, (char*)zframe_data(payload), zframe_size(payload) , NULL))			{				//zclock_log("message");				//msgpack_object_print(stdout, object.data);				char *command = (char*)m_lookup(object.data, "command");				if (command) {					//zclock_log("command: %s", command);					if (streq(command, "exception")) {						failed++;						zclock_log("exception");					}					if (streq(command, "result")) {						success++;						zclock_log("result ok");					}					free(command);				}			}			msgpack_unpacked_destroy(&object);			zframe_destroy(&payload);		}				gettimeofday(&tv,NULL);		uint64_t end = (uint64_t)tv.tv_sec;		if ((end - start) == 1) 		{			float speed = counter/(end-start);			zclock_log("speed %f m/s, failed %ld, success %ld", speed, failed, success);			counter = 0;			failed = 0;			success = 0;			start = end;		} 	//} //end while}

static ints_subscriber (zloop_t *loop, zmq_pollitem_t *poller, void *args){    clonesrv_t *self = (clonesrv_t *) args;    //  Get state snapshot if necessary    if (self->kvmap == NULL) {        self->kvmap = zhash_new ();        void *snapshot = zsocket_new (self->ctx, ZMQ_DEALER);        zsocket_connect (snapshot, "tcp://localhost:%d", self->peer);        zclock_log ("I: asking for snapshot from: tcp://localhost:%d",                    self->peer);        zstr_sendm (snapshot, "ICANHAZ?");        zstr_send (snapshot, ""); // blank subtree to get all        while (true) {            kvmsg_t *kvmsg = kvmsg_recv (snapshot);            if (!kvmsg)                break;          //  Interrupted            if (streq (kvmsg_key (kvmsg), "KTHXBAI")) {                self->sequence = kvmsg_sequence (kvmsg);                kvmsg_destroy (&kvmsg);                break;          //  Done            }            kvmsg_store (&kvmsg, self->kvmap);        }        zclock_log ("I: received snapshot=%d", (int) self->sequence);        zsocket_destroy (self->ctx, snapshot);    }    //  Find and remove update off pending list    kvmsg_t *kvmsg = kvmsg_recv (poller->socket);    if (!kvmsg)        return 0;    if (strneq (kvmsg_key (kvmsg), "HUGZ")) {        if (!s_was_pending (self, kvmsg)) {            //  If active update came before client update, flip it            //  around, store active update (with sequence) on pending            //  list and use to clear client update when it comes later            zlist_append (self->pending, kvmsg_dup (kvmsg));        }        //  If update is more recent than our kvmap, apply it        if (kvmsg_sequence (kvmsg) > self->sequence) {            self->sequence = kvmsg_sequence (kvmsg);            kvmsg_store (&kvmsg, self->kvmap);            zclock_log ("I: received update=%d", (int) self->sequence);        }        else            kvmsg_destroy (&kvmsg);    }    else        kvmsg_destroy (&kvmsg);    return 0;}

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

intzloop_poller (zloop_t *self, zmq_pollitem_t *item, zloop_fn handler, void *arg){    assert (self);    if (!item->socket && !item->fd)        return -1;    if (item->socket)        if (streq (zsocket_type_str (item->socket), "UNKNOWN"))            return -1;    s_poller_t *poller = s_poller_new (item, handler, arg);    if (poller) {        if (zlist_append (self->pollers, poller))            return -1;        self->dirty = true;        if (self->verbose)            zclock_log ("I: zloop: register %s poller (%p, %d)",                item->socket? zsocket_type_str (item->socket): "FD",                item->socket, item->fd);        return 0;    }    else        return -1;}

static voidpeering_raise (peering_t *self){    vocket_t *vocket = self->vocket;    driver_t *driver = self->driver;    if (driver->verbose)        zclock_log ("I: (tcp) bring up peering to %s", self->address);    if (!self->alive) {        self->alive = TRUE;        zlist_append (vocket->live_peerings, self);        //  Send ZMTP handshake, which is an empty message        zmq_msg_t msg;        zmq_msg_init_size (&msg, 0);        s_queue_output (self, &msg, FALSE);        //  If we can now route to peerings, start reading from msgpipe        if (zlist_size (vocket->live_peerings) == vocket->min_peerings) {            //  Ask reactor to start monitoring vocket's msgpipe pipe            zmq_pollitem_t item = { vocket->msgpipe, 0, ZMQ_POLLIN, 0 };            zloop_poller (driver->loop, &item, s_vocket_input, vocket);        }    }}

static ints_binding_input (zloop_t *loop, zmq_pollitem_t *item, void *arg){    vocket_t *vocket = (vocket_t *) arg;    driver_t *driver = vocket->driver;    struct sockaddr_in addr;        //  Peer address    socklen_t addr_len = sizeof (addr);    int handle = accept (item->fd, (struct sockaddr *) &addr, &addr_len);    if (handle >= 0) {        s_set_nonblock (handle);        if (vocket->peerings < vocket->max_peerings) {            char *address = s_sin_addr_to_str (&addr);            peering_t *peering = peering_require (vocket, address, FALSE);            peering->handle = handle;            peering_raise (peering);            peering_poller (peering, ZMQ_POLLIN + ZMQ_POLLOUT);        }        else {            zclock_log ("W: Max peerings reached for socket");            close (handle);        }    }    else        s_handle_io_error ("accept");    return 0;}

static voidtest_tcp_pair_cli (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);    void *pair = vtx_socket (vtx, ZMQ_PAIR);    assert (pair);    rc = vtx_connect (vtx, pair, "tcp://localhost:%s", port);    assert (rc == 0);    int sent = 0;    int recd = 0;    while (!zctx_interrupted) {        zstr_send (pair, "ICANHAZ?");        sent++;        char *reply = zstr_recv_nowait (pair);        if (reply) {            recd++;            free (reply);        }        char *end = zstr_recv_nowait (pipe);        if (end) {            free (end);            zstr_send (pipe, "OK");            break;        }    }    zclock_log ("I: PAIR CLI: sent=%d recd=%d", sent, recd);    free (port);    vtx_destroy (&vtx);}

static ints_snapshots (zloop_t *loop, zmq_pollitem_t *poller, void *args){    clonesrv_t *self = (clonesrv_t *) args;    zframe_t *identity = zframe_recv (poller->socket);    if (identity) {        //  Request is in second frame of message        char *request = zstr_recv (poller->socket);        char *subtree = NULL;        if (streq (request, "ICANHAZ?")) {            free (request);            subtree = zstr_recv (poller->socket);        }        else            printf ("E: bad request, aborting/n");        if (subtree) {            //  Send state socket to client            kvroute_t routing = { poller->socket, identity, subtree };            zhash_foreach (self->kvmap, s_send_single, &routing);            //  Now send END message with sequence number            zclock_log ("I: sending shapshot=%d", (int) self->sequence);            zframe_send (&identity, poller->socket, ZFRAME_MORE);            kvmsg_t *kvmsg = kvmsg_new (self->sequence);            kvmsg_set_key  (kvmsg, "KTHXBAI");            kvmsg_set_body (kvmsg, (byte *) subtree, 0);            kvmsg_send     (kvmsg, poller->socket);            kvmsg_destroy (&kvmsg);            free (subtree);        }    }    return 0;}

static ints_collector (zloop_t *loop, zmq_pollitem_t *poller, void *args){    clonesrv_t *self = (clonesrv_t *) args;    kvmsg_t *kvmsg = kvmsg_recv (poller->socket);    if (kvmsg) {        if (self->active) {            kvmsg_set_sequence (kvmsg, ++self->sequence);            kvmsg_send (kvmsg, self->publisher);            int ttl = atoi (kvmsg_get_prop (kvmsg, "ttl"));            if (ttl)                kvmsg_set_prop (kvmsg, "ttl",                    "%" PRId64, zclock_time () + ttl * 1000);            kvmsg_store (&kvmsg, self->kvmap);            zclock_log ("I: publishing update=%d", (int) self->sequence);        }        else {            //  If we already got message from active, drop it, else            //  hold on pending list            if (s_was_pending (self, kvmsg))                kvmsg_destroy (&kvmsg);            else                zlist_append (self->pending, kvmsg);        }    }    return 0;}

static service_t *s_service_require (broker_t *self, zframe_t *service_frame){    assert (service_frame);    char *name = zframe_strdup (service_frame);    service_t *service =        (service_t *) zhash_lookup (self->services, name);    if (service == NULL) {        service = (service_t *) zmalloc (sizeof (service_t));        service->broker = self;        service->name = name;        service->requests = zlist_new ();        service->waiting = zlist_new ();        service->blacklist = zlist_new ();        zhash_insert (self->services, name, service);        zhash_freefn (self->services, name, s_service_destroy);        if (self->verbose)            zclock_log ("I: added service: %s", name);    }    else        free (name);    return service;}

voidztask_job_request_clean (ztask_job_request_t *self, ztask_node_manager_t *node_mgr){    assert (self);//    zclock_log ("Cleaning job request from %s ...", zyre_event_sender (self->request));    zlist_t *keys = zhash_keys (self->processes);    char *key = (char *) zlist_first (keys);    ztask_job_proc_t *p;    int pid;    while (key) {        p = (ztask_job_proc_t *) zhash_lookup (self->processes, key);        zhash_delete (self->processes, key);        pid = ztask_job_proc_pid(p);//        assert (pid);        if (pid) {            zclock_log("Killing pid=%d ...", ztask_job_proc_pid(p));            kill (ztask_job_proc_pid(p), SIGKILL);            zmsg_t *msg_report = zmsg_new ();            zmsg_addstr (msg_report, "REPORT");            zmsg_addstr (msg_report, ztask_job_proc_jobid(p));            zmsg_addstr (msg_report, "-100");            zyre_whisper (ztask_node_manager_zyre_node(node_mgr), zyre_event_sender(self->request), &msg_report);            zmsg_destroy (&msg_report);            zhash_delete (ztask_node_manager_list_running_processes (node_mgr), ztask_job_proc_jobid(p));            zlist_append (ztask_node_manager_list_available_processes (node_mgr), p);            ztask_job_proc_reset(p);        }        key = (char *) zlist_next (keys);    }    zlist_destroy (&keys);}

static void s_broker_worker_msg(broker_t *self, zframe_t *sender, zmsg_t *msg){    assert (zmsg_size(msg) >= 1);     //  At least, command    zframe_t *command = zmsg_pop(msg);    char *id_string = zframe_strhex(sender);    int worker_ready = (zhash_lookup(self->workers, id_string) != NULL);    free (id_string);	    worker_t *worker = s_worker_require(self, sender);    if (zframe_streq(command, MDPW_READY)) {        if (worker_ready) {               //  Not first command in session            s_worker_delete(worker, 1);			// Додумать, по идеи синоним сердцебиения		} else {			if (zframe_size(sender) >= 4 &&  memcmp(zframe_data (sender), "mmi.", 4) == 0) {				s_worker_delete(worker, 1);				// Додумать, по идеи синоним сердцебиения			} else {				//  Attach worker to service and mark as idle				zframe_t *service_frame = zmsg_pop(msg);				worker->service = s_service_require(self, service_frame);				worker->service->workers++;				s_worker_waiting(worker);				zframe_destroy(&service_frame);			}		}    } else if (zframe_streq(command, MDPW_REPLY)) {        if (worker_ready) {            //  Remove and save client return envelope and insert the            //  protocol header and service name, then rewrap envelope.            zframe_t *client = zmsg_unwrap(msg);            zmsg_pushstr(msg, worker->service->name);            zmsg_pushstr(msg, MDPC_CLIENT);            zmsg_wrap(msg, client);            zmsg_send(&msg, self->socket);            s_worker_waiting(worker);        } else {			// Просто обрыв связи между воркером и брокером			// синоним сердцебиения            s_worker_delete(worker, 1);		}    } else if (zframe_streq(command, MDPW_HEARTBEAT)) {        if (worker_ready) {            worker->expiry = zclock_time () + HEARTBEAT_EXPIRY;		} else {			// Просто обрыв связи между воркером и брокером			// синоним сердцебиения            s_worker_delete(worker, 1);		}    } else if (zframe_streq (command, MDPW_DISCONNECT)) {        s_worker_delete(worker, 0);	} else {        zclock_log ("E: invalid input message");        zmsg_dump (msg);    }    free (command);    zmsg_destroy (&msg);}

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

static voids_engine_handle_request (engine_t *self, zmsg_t *request, zframe_t *reply_to){    assert (zmsg_size (request) >= 3);    zframe_t *operation = zmsg_pop (request);    zframe_t *price     = zmsg_pop (request);    zframe_t *volume    = zmsg_pop (request);    if (zframe_streq (operation, "SELL"))        s_engine_handle_sell_request (self, price, volume, reply_to);    else    if (zframe_streq (operation, "BUY"))        s_engine_handle_buy_request (self, price, volume, reply_to);    else {        zclock_log ("E: invalid message: ");        zmsg_dump (request);    }    zframe_destroy (&operation);    zframe_destroy (&price);    zframe_destroy (&volume);    zmsg_destroy (&request);}

//  Apply configuration tree://   * apply server configuration//   * print any echo items in top-level sections//   * apply sections that match methodsstatic voidserver_apply_config (server_t *self){    //  Apply echo commands and class methods    zconfig_t *section = zconfig_child (self->config);    while (section) {        zconfig_t *entry = zconfig_child (section);        while (entry) {            if (streq (zconfig_name (entry), "echo"))                zclock_log (zconfig_value (entry));            entry = zconfig_next (entry);        }        if (streq (zconfig_name (section), "bind")) {            char *endpoint = zconfig_resolve (section, "endpoint", "?");            self->port = zsocket_bind (self->router, endpoint);        }        else        if (streq (zconfig_name (section), "publish")) {            char *location = zconfig_resolve (section, "location", "?");            char *alias = zconfig_resolve (section, "alias", "?");            mount_t *mount = mount_new (location, alias);            zlist_append (self->mounts, mount);                  }        else        if (streq (zconfig_name (section), "set_anonymous")) {            long enabled = atoi (zconfig_resolve (section, "enabled", ""));            //  Enable anonymous access without a config file                               zconfig_put (self->config, "security/anonymous", enabled? "1" :"0");        }        section = zconfig_next (section);    }    server_config_self (self);}

static voids_worker_process (broker_t *self, zframe_t *sender, zmsg_t *msg){    assert (zmsg_size (msg) >= 1);     //  At least, command    zframe_t *command = zmsg_pop (msg);    char *identity = zframe_strhex (sender);    int worker_ready = (zhash_lookup (self->workers, identity) != NULL);    free (identity);    worker_t *worker = s_worker_require (self, sender);    if (zframe_streq (command, MDPW_READY)) {        if (worker_ready)               //  Not first command in session            s_worker_delete (self, worker, 1);        else        if (zframe_size (sender) >= 4  //  Reserved service name        &&  memcmp (zframe_data (sender), "mmi.", 4) == 0)            s_worker_delete (self, worker, 1);        else {            //  Attach worker to service and mark as idle            zframe_t *service_frame = zmsg_pop (msg);            worker->service = s_service_require (self, service_frame);            worker->service->workers++;            s_worker_waiting (self, worker);            zframe_destroy (&service_frame);        }    }    else    if (zframe_streq (command, MDPW_REPLY)) {        if (worker_ready) {            //  Remove & save client return envelope and insert the            //  protocol header and service name, then rewrap envelope.            zframe_t *client = zmsg_unwrap (msg);            zmsg_pushstr (msg, worker->service->name);            zmsg_pushstr (msg, MDPC_CLIENT);            zmsg_wrap (msg, client);            zmsg_send (&msg, self->socket);            s_worker_waiting (self, worker);        }        else            s_worker_delete (self, worker, 1);    }    else    if (zframe_streq (command, MDPW_HEARTBEAT)) {        if (worker_ready)            worker->expiry = zclock_time () + HEARTBEAT_EXPIRY;        else            s_worker_delete (self, worker, 1);    }    else    if (zframe_streq (command, MDPW_DISCONNECT))        s_worker_delete (self, worker, 0);    else {        zclock_log ("E: invalid input message");        zmsg_dump (msg);    }    free (command);    zmsg_destroy (&msg);}

int main (int argc, char *argv []){    char *sock = (argc > 1)? argv [1]: "";    if (streq(sock,""))    {        zclock_log("cannot start broker for %s", sock);        return -1;    }    char *job = (argc > 2)? argv [2]: "";    if (streq(job,""))     {		zclock_log("cannot start broker job %s", job);		return -1;	}        if (s_connect(sock)<0) {        zclock_log("cannot connect to %s", sock);        return -1;    }    else {        zclock_log("broker connected to %s", sock);    }            struct sigaction sa;    sigemptyset(&sa.sa_mask);    sa.sa_flags=0;    sa.sa_handler = s_signal_handler;    sigaction(SIGHUP, &sa, 0);    sigaction(SIGINT, &sa, 0);    sigaction(SIGQUIT, &sa, 0);    sigaction(SIGABRT, &sa, 0);    sigaction(SIGTERM, &sa, 0);	if (streq(job,"loop")) 	{		char *arg = (argc > 3)? argv [3]: "1";		int count = atoi(arg);		start_loop(count);	}	if (streq(job,"sleep"))	{		char *arg = (argc > 3)? argv [3]: "1000";		start_echo(arg);	}    exit(0);}

void s_mdp_client_connect_to_broker (mdp_client_t *self){    if (self->client)        zsocket_destroy (self->ctx, self->client);    self->client = zsocket_new (self->ctx, ZMQ_DEALER);    zmq_connect (self->client, self->broker);    if (self->verbose)        zclock_log ("I: connecting to broker at %s...", self->broker);}

static ints_peering_monitor (zloop_t *loop, zmq_pollitem_t *item, void *arg){    peering_t *peering = (peering_t *) arg;    vocket_t *vocket = peering->vocket;    driver_t *driver = peering->driver;    //  The peering monitor handles just outgoing peering reconnect    //  attempts. It'll keep trying until successful.    assert (peering->outgoing);    peering->exception = FALSE;    if (peering->alive)        return 0;           //  Stop monitor if peering came alive    if (driver->verbose)        zclock_log ("I: (tcp) connecting to '%s'...", peering->address);    peering->handle = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);    if (peering->handle == -1) {        zclock_log ("E: connect failed: no sockets - %s", strerror (errno));        goto error;    }    s_set_nonblock (peering->handle);    if (s_str_to_sin_addr (&peering->addr, peering->address)) {        zclock_log ("E: connect failed: bad address '%s'", peering->address);        goto error;    }    int rc = connect (peering->handle,        (const struct sockaddr *) &peering->addr, IN_ADDR_SIZE);    if (rc == -1 && errno != EINPROGRESS) {        zclock_log ("E: connect failed: '%s'", strerror (errno));        goto error;    }    peering_poller (peering, ZMQ_POLLIN + ZMQ_POLLOUT);    return 0;error:    if (peering->handle > 0) {        close (peering->handle);        peering->handle = 0;    }    //  Try again later    zloop_timer (loop, peering->interval, 1, s_peering_monitor, peering);    return 0;}

zlist_t * fetch_rules(void * sock, char * channel) {  zlist_t * rules = zlist_new();  /*  if(!endpoint) {    zclock_log("W: no reup endpoint defined, won't try te refresh rules");    return rules;    }*/  // zsocket_connect(sock, endpoint);  zstr_sendm(sock, "");  zstr_send(sock, channel);  zmsg_t * tmp;   //   while(tmp=zmsg_recv(sock)) {      //     zmsg_dump(tmp);   //}   // exit(1);   zmsg_t * msg = zmsg_recv(sock);  kill_envelope(msg);  char * status = zmsg_popstr(msg);  if(strcmp("200", status) != 0) {    zclock_log("W: reloading rules for %s failed: got |%s|", channel, status);    return rules;  }  free(status);  zmsg_destroy(&msg);    while((msg=zmsg_recv(sock))) {    zclock_log("once");    kill_envelope(msg);    zmsg_dump(msg);    zframe_t * header = zmsg_pop(msg);    if(zframe_streq(header, "")) {      // we're done      zclock_log("got a null header, we're out");      zframe_destroy(&header);      zmsg_destroy(&msg);      break;    }    zmsg_push(msg, header);    zlist_push(rules, msg);  }  return rules;}

static voids_broker_worker_msg(broker_t *self, zframe_t *sender, zmsg_t *msg){	assert(zmsg_size(msg) >= 1);	zframe_t *command = zmsg_pop(msg);	char *id_string = zframe_strhex(sender);	int worker_ready = (zhash_lookup(self->workers, id_string) != NULL);	free(id_string);	worker_t *worker = s_worker_require(self, sender);	if (zframe_streq(command, MDPW_READY)){		if (worker_ready)			s_worker_delete(worker, 1);		else 		if (zframe_size(sender) >= 4 && memcmp(zframe_data(sender), "mmi.", 4) == 0)			s_worker_delete(worker, 1);		else {			zframe_t *service_frame = zmsg_pop(msg);			worker->service = s_service_require(self, service_frame);			worker->service->workers++;			s_worker_waiting(worker);			zframe_destroy(&service_frame);		}	}	else	if (zframe_streq(command, MDPW_REPLY)){		if (worker_ready){			zframe_t *client = zmsg_unwrap(msg);			zmsg_pushstr(msg, worker->service->name);			zmsg_pushstr(msg, MDPC_CLIENT);			zmsg_wrap(msg, client);			zmsg_send(&msg, self->socket);			s_worker_waiting(worker);		}		else 			s_worker_delete(worker, 1);	}	else	if (zframe_streq(command, MDPW_HEARTBEAT)){		if (worker_ready)			worker->expiry = zclock_time() + HEARTBEAT_EXPIRY;		else			s_worker_delete(worker, 1);	}	else	if (zframe_streq(command, MDPW_DISCONNECT))		s_worker_delete(worker, 0);	else {		zclock_log("E: invalid input message");		zmsg_dump(msg);	}	free(command);	zmsg_destroy(&msg);}

static int s_connect(char *arg){		ctx = zctx_new();		if (!ctx) {			zclock_log("error creating zmq context - %s", zmq_strerror (errno));			return -1;		}		dealer = zsocket_new(ctx, ZMQ_DEALER);		if (!dealer) {			zclock_log("error creating zmq socket - %s", zmq_strerror (errno));			return -1;		}				if (zsocket_connect(dealer, "%s", arg) != 0) {				zclock_log("error connecting socket %s - %s",arg , zmq_strerror (errno));			return -1;		}		return 0;}