diff options
Diffstat (limited to 'src/tcp_connecter.cpp')
| -rw-r--r-- | src/tcp_connecter.cpp | 182 |
1 files changed, 143 insertions, 39 deletions
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp index 01d80cc..678d488 100644 --- a/src/tcp_connecter.cpp +++ b/src/tcp_connecter.cpp @@ -18,30 +18,168 @@ along with this program. If not, see <http://www.gnu.org/licenses/>. */ -#include <string.h> - +#include <new> #include <string> #include "tcp_connecter.hpp" +#include "tcp_engine.hpp" +#include "io_thread.hpp" #include "platform.hpp" #include "ip.hpp" #include "err.hpp" -#ifdef ZMQ_HAVE_WINDOWS +#if defined ZMQ_HAVE_WINDOWS +#include "windows.hpp" +#else +#include <unistd.h> +#include <sys/types.h> +#include <sys/socket.h> +#include <arpa/inet.h> +#include <netinet/tcp.h> +#include <netinet/in.h> +#include <netdb.h> +#include <fcntl.h> +#ifdef ZMQ_HAVE_OPENVMS +#include <ioctl.h> +#endif +#endif -zmq::tcp_connecter_t::tcp_connecter_t () : - s (retired_fd) +zmq::tcp_connecter_t::tcp_connecter_t (class io_thread_t *io_thread_, + class session_t *session_, const options_t &options_, + const char *protocol_, const char *address_, bool wait_) : + own_t (io_thread_, options_), + io_object_t (io_thread_), + s (retired_fd), + handle_valid (false), + wait (wait_), + session (session_), + current_reconnect_ivl(options.reconnect_ivl) { memset (&addr, 0, sizeof (addr)); addr_len = 0; + + int rc = set_address (protocol_, address_); + zmq_assert (rc == 0); //TODO: take care ENOMEM, EINVAL } zmq::tcp_connecter_t::~tcp_connecter_t () { + if (wait) + cancel_timer (reconnect_timer_id); + if (handle_valid) + rm_fd (handle); + if (s != retired_fd) close (); } +void zmq::tcp_connecter_t::process_plug () +{ + if (wait) + add_reconnect_timer(); + else + start_connecting (); +} + +void zmq::tcp_connecter_t::in_event () +{ + // We are not polling for incomming data, so we are actually called + // because of error here. However, we can get error on out event as well + // on some platforms, so we'll simply handle both events in the same way. + out_event (); +} + +void zmq::tcp_connecter_t::out_event () +{ + fd_t fd = connect (); + rm_fd (handle); + handle_valid = false; + + // Handle the error condition by attempt to reconnect. + if (fd == retired_fd) { + close (); + wait = true; + add_reconnect_timer(); + return; + } + + // Create the engine object for this connection. + tcp_engine_t *engine = new (std::nothrow) tcp_engine_t (fd, options); + alloc_assert (engine); + + // Attach the engine to the corresponding session object. + send_attach (session, engine); + + // Shut the connecter down. + terminate (); +} + +void zmq::tcp_connecter_t::timer_event (int id_) +{ + zmq_assert (id_ == reconnect_timer_id); + wait = false; + start_connecting (); +} + +void zmq::tcp_connecter_t::start_connecting () +{ + // Open the connecting socket. + int rc = open (); + + // Connect may succeed in synchronous manner. + if (rc == 0) { + handle = add_fd (s); + handle_valid = true; + out_event (); + return; + } + + // Connection establishment may be dealyed. Poll for its completion. + else if (rc == -1 && errno == EAGAIN) { + handle = add_fd (s); + handle_valid = true; + set_pollout (handle); + return; + } + + // Handle any other error condition by eventual reconnect. + wait = true; + add_reconnect_timer(); +} + +void zmq::tcp_connecter_t::add_reconnect_timer() +{ + add_timer (get_new_reconnect_ivl(), reconnect_timer_id); +} + +int zmq::tcp_connecter_t::get_new_reconnect_ivl () +{ +#if defined ZMQ_HAVE_WINDOWS + int pid = (int) GetCurrentProcessId (); +#else + int pid = (int) getpid (); +#endif + + // The new interval is the current interval + random value. + int this_interval = current_reconnect_ivl + + ((pid * 13) % options.reconnect_ivl); + + // Only change the current reconnect interval if the maximum reconnect + // interval was set and if it's larger than the reconnect interval. + if (options.reconnect_ivl_max > 0 && + options.reconnect_ivl_max > options.reconnect_ivl) { + + // Calculate the next interval + current_reconnect_ivl = current_reconnect_ivl * 2; + if(current_reconnect_ivl >= options.reconnect_ivl_max) { + current_reconnect_ivl = options.reconnect_ivl_max; + } + } + return this_interval; +} + +#ifdef ZMQ_HAVE_WINDOWS + int zmq::tcp_connecter_t::set_address (const char *protocol_, const char *addr_) { if (strcmp (protocol_, "tcp") == 0) @@ -100,11 +238,6 @@ int zmq::tcp_connecter_t::close () return 0; } -zmq::fd_t zmq::tcp_connecter_t::get_fd () -{ - return s; -} - zmq::fd_t zmq::tcp_connecter_t::connect () { // Nonblocking connect have finished. Check whether an error occured. @@ -132,30 +265,6 @@ zmq::fd_t zmq::tcp_connecter_t::connect () #else -#include <unistd.h> -#include <sys/socket.h> -#include <arpa/inet.h> -#include <netinet/tcp.h> -#include <netinet/in.h> -#include <netdb.h> -#include <fcntl.h> - -#ifdef ZMQ_HAVE_OPENVMS -#include <ioctl.h> -#endif - -zmq::tcp_connecter_t::tcp_connecter_t () : - s (retired_fd) -{ - memset (&addr, 0, sizeof (addr)); -} - -zmq::tcp_connecter_t::~tcp_connecter_t () -{ - if (s != retired_fd) - close (); -} - int zmq::tcp_connecter_t::set_address (const char *protocol_, const char *addr_) { if (strcmp (protocol_, "tcp") == 0) @@ -271,11 +380,6 @@ int zmq::tcp_connecter_t::close () return 0; } -zmq::fd_t zmq::tcp_connecter_t::get_fd () -{ - return s; -} - zmq::fd_t zmq::tcp_connecter_t::connect () { // Following code should handle both Berkeley-derived socket |