From 9119b4fd7b292b1a14db916040f8e7cc4731d4b6 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Tue, 26 Jul 2011 00:43:57 +0200 Subject: TCP transport classes simplified zmq_engine and tcp_socket merged into tcp_engine zmq_connecter and tcp_connecter merged into tcp_connecter zmq_listener and tcp_listener merged into tcp_listener Signed-off-by: Martin Sustrik --- src/zmq_connecter.cpp | 161 -------------------------------------------------- 1 file changed, 161 deletions(-) delete mode 100644 src/zmq_connecter.cpp (limited to 'src/zmq_connecter.cpp') diff --git a/src/zmq_connecter.cpp b/src/zmq_connecter.cpp deleted file mode 100644 index 0512c3c..0000000 --- a/src/zmq_connecter.cpp +++ /dev/null @@ -1,161 +0,0 @@ -/* - Copyright (c) 2007-2011 iMatix Corporation - Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file - - This file is part of 0MQ. - - 0MQ is free software; you can redistribute it and/or modify it under - the terms of the GNU Lesser General Public License as published by - the Free Software Foundation; either version 3 of the License, or - (at your option) any later version. - - 0MQ is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU Lesser General Public License for more details. - - You should have received a copy of the GNU Lesser General Public License - along with this program. If not, see . -*/ - -#include - -#include "platform.hpp" -#if defined ZMQ_HAVE_WINDOWS -#include "windows.hpp" -#else -#include -#include -#endif - -#include "zmq_connecter.hpp" -#include "zmq_engine.hpp" -#include "io_thread.hpp" -#include "err.hpp" - -zmq::zmq_connecter_t::zmq_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_), - handle_valid (false), - wait (wait_), - session (session_), - current_reconnect_ivl(options.reconnect_ivl) -{ - int rc = tcp_connecter.set_address (protocol_, address_); - zmq_assert (rc == 0); //TODO: take care ENOMEM, EINVAL -} - -zmq::zmq_connecter_t::~zmq_connecter_t () -{ - if (wait) - cancel_timer (reconnect_timer_id); - if (handle_valid) - rm_fd (handle); -} - -void zmq::zmq_connecter_t::process_plug () -{ - if (wait) - add_reconnect_timer(); - else - start_connecting (); -} - -void zmq::zmq_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::zmq_connecter_t::out_event () -{ - fd_t fd = tcp_connecter.connect (); - rm_fd (handle); - handle_valid = false; - - // Handle the error condition by attempt to reconnect. - if (fd == retired_fd) { - tcp_connecter.close (); - wait = true; - add_reconnect_timer(); - return; - } - - // Create the engine object for this connection. - zmq_engine_t *engine = new (std::nothrow) zmq_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::zmq_connecter_t::timer_event (int id_) -{ - zmq_assert (id_ == reconnect_timer_id); - wait = false; - start_connecting (); -} - -void zmq::zmq_connecter_t::start_connecting () -{ - // Open the connecting socket. - int rc = tcp_connecter.open (); - - // Connect may succeed in synchronous manner. - if (rc == 0) { - handle = add_fd (tcp_connecter.get_fd ()); - handle_valid = true; - out_event (); - return; - } - - // Connection establishment may be dealyed. Poll for its completion. - else if (rc == -1 && errno == EAGAIN) { - handle = add_fd (tcp_connecter.get_fd ()); - handle_valid = true; - set_pollout (handle); - return; - } - - // Handle any other error condition by eventual reconnect. - wait = true; - add_reconnect_timer(); -} - -void zmq::zmq_connecter_t::add_reconnect_timer() -{ - add_timer (get_new_reconnect_ivl(), reconnect_timer_id); -} - -int zmq::zmq_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; -} -- cgit v1.2.3