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

static struct client_socket *client_socket_create(					      struct addrinfo *addrinfos,					      void (*fai)(struct addrinfo *)){	struct connector *c = xalloc(sizeof *c);	struct client_socket *s = xalloc(sizeof *s);	c->socket = s;	tasklet_init(&c->tasklet, &s->base.mutex, c);	wait_list_init(&c->connecting, 0);	c->fd = -1;	c->connected = 0;	c->next_addrinfo = c->addrinfos = addrinfos;	c->fai = fai;	error_init(&c->err);	simple_socket_init(&s->base, &client_socket_ops, -1);	s->connector = c;	/* Start connecting */	mutex_lock(&s->base.mutex);	start_connecting(c);	tasklet_later(&c->tasklet, finish_connecting);	mutex_unlock(&s->base.mutex);	return s;}

void zmq::ipc_connecter_t::process_plug (){    if (delayed_start)        add_reconnect_timer ();    else        start_connecting ();}

void zmq::tcp_connecter_t::process_plug (){    if (wait)        add_reconnect_timer();    else        start_connecting ();}

void zmq::session_base_t::detached (){    //  Transient session self-destructs after peer disconnects.    if (!connect) {        terminate ();        return;    }    //  For delayed connect situations, terminate the pipe    //  and reestablish later on    if (pipe && options.immediate == 1        && addr->protocol != "pgm" && addr->protocol != "epgm") {        pipe->hiccup ();        pipe->terminate (false);        terminating_pipes.insert (pipe);        pipe = NULL;    }    reset ();    //  Reconnect.    if (options.reconnect_ivl != -1)        start_connecting (true);    //  For subscriber sockets we hiccup the inbound pipe, which will cause    //  the socket object to resend all the subscriptions.    if (pipe && (options.type == ZMQ_SUB || options.type == ZMQ_XSUB))        pipe->hiccup ();}

void zmq::session_base_t::reconnect (){    //  For delayed connect situations, terminate the pipe    //  and reestablish later on    if (pipe && options.immediate == 1        && addr->protocol != "pgm" && addr->protocol != "epgm"        && addr->protocol != "norm" && addr->protocol != "udp") {        pipe->hiccup ();        pipe->terminate (false);        terminating_pipes.insert (pipe);        pipe = NULL;        if (has_linger_timer) {            cancel_timer (linger_timer_id);            has_linger_timer = false;        }    }    reset ();    //  Reconnect.    if (options.reconnect_ivl != -1)        start_connecting (true);    //  For subscriber sockets we hiccup the inbound pipe, which will cause    //  the socket object to resend all the subscriptions.    if (pipe && (options.type == ZMQ_SUB || options.type == ZMQ_XSUB || options.type == ZMQ_DISH))        pipe->hiccup ();}

static void finish_connecting(void *v_c){	struct connector *c = v_c;	struct client_socket *s = c->socket;	for (;;) {		if (!wait_list_down(&c->connecting, 1, &c->tasklet))			return;		if (c->connected) {			int fd = c->fd;			struct watched_fd *watched_fd = c->watched_fd;			c->fd = -1;			connector_destroy(c);			s->connector = NULL;			simple_socket_set_fd(&s->base, fd, watched_fd);			/* Access to the ops pointer is not locked.			 * But it is fine if some threads continue to			 * use the old client_socket_ops value. */			s->base.base.ops = &simple_socket_ops;			/* Tasklet got destroyed, so we are still			   holding the associated lock. */			mutex_unlock(&s->base.mutex);			return;		}		else {			/* Got POLLERR */			int e;			socklen_t len = sizeof e;			const char *syscall = "connect";			if (getsockopt(c->fd, SOL_SOCKET, SO_ERROR, &e, &len)) {				e = errno;				syscall = "getsockopt";			}			if (e) {				/* Stash the error and try another address */				error_errno_val(&c->err, e, "%s", syscall);				start_connecting(c);			}			/* Strange, no error.  Continue to poll. */			else if (!watched_fd_set_interest(c->watched_fd,						  WATCHED_FD_OUT, &c->err)) {				simple_socket_wake_all(&c->socket->base);			}		}	}}

void zmq::tcp_connecter_t::timer_event (int id_){    zmq_assert (id_ == reconnect_timer_id || id_ == connect_timer_id);    if (id_ == connect_timer_id) {        connect_timer_started = false;        rm_handle ();        close ();        add_reconnect_timer ();    } else if (id_ == reconnect_timer_id) {        reconnect_timer_started = false;        start_connecting ();    }}

void zmq::session_base_t::detached (){    //  Transient session self-destructs after peer disconnects.    if (!connect) {        terminate ();        return;    }    //  Reconnect.    start_connecting (true);    //  For subscriber sockets we hiccup the inbound pipe, which will cause    //  the socket object to resend all the subscriptions.    if (pipe && (options.type == ZMQ_SUB || options.type == ZMQ_XSUB))        pipe->hiccup ();  }

void zmq::ipc_connecter_t::timer_event (int id_){    zmq_assert (id_ == reconnect_timer_id);    timer_started = false;    start_connecting ();}

void zmq::connect_session_t::process_plug (){    //  Start connection process immediately.    start_connecting (false);}

void zmq::tcp_connecter_t::timer_event (int id_){    zmq_assert (id_ == reconnect_timer_id);    wait = false;    start_connecting ();}

void zmq::session_base_t::process_plug (){    if (active)        start_connecting (false);}

void zmq::zmq_connecter_t::timer_event (){    wait = false;    start_connecting ();}