ZMQ_RECONNECT_IVL_MAX socket option added

It allows for exponential back-off strategy when reconnecting.

Signed-off-by: Thijs Terlouw <thijsterlouw@gmail.com>

1 files changed, 38 insertions, 17 deletions

diff --git a/src/zmq_connecter.cpp b/src/zmq_connecter.cpp
index 8a6d89a..d2638ac 100644
--- a/src/zmq_connecter.cpp
+++ b/src/zmq_connecter.cpp
@@ -40,10 +40,11 @@ zmq::zmq_connecter_t::zmq_connecter_t (class io_thread_t *io_thread_,
     io_object_t (io_thread_),
     handle_valid (false),
     wait (wait_before_connect),
-    session (session_)
+    session (session_),
+    current_reconnect_ivl(options.reconnect_ivl)
 {
-     int rc = tcp_connecter.set_address (protocol_, address_);
-     zmq_assert (rc == 0);
+    int rc = tcp_connecter.set_address (protocol_, address_);
+    zmq_assert (rc == 0);
 }
 
 zmq::zmq_connecter_t::~zmq_connecter_t ()
@@ -54,21 +55,10 @@ zmq::zmq_connecter_t::~zmq_connecter_t ()
         rm_fd (handle);
 }
 
-int zmq::zmq_connecter_t::get_reconnect_ivl ()
-{
-#if defined ZMQ_HAVE_WINDOWS
-    return (options.reconnect_ivl + (((int) GetCurrentProcessId () * 13)
-        % options.reconnect_ivl));
-#else
-    return (options.reconnect_ivl + (((int) getpid () * 13)
-        % options.reconnect_ivl));
-#endif
-}
-
 void zmq::zmq_connecter_t::process_plug ()
 {
     if (wait)
-        add_timer (get_reconnect_ivl (), reconnect_timer_id);
+        add_reconnect_timer();
     else
         start_connecting ();
 }
@@ -91,7 +81,7 @@ void zmq::zmq_connecter_t::out_event ()
     if (fd == retired_fd) {
         tcp_connecter.close ();
         wait = true;
-        add_timer (get_reconnect_ivl (), reconnect_timer_id);
+        add_reconnect_timer();
         return;
     }
 
@@ -140,5 +130,36 @@ void zmq::zmq_connecter_t::start_connecting ()
 
     //  Handle any other error condition by eventual reconnect.
     wait = true;
-    add_timer (get_reconnect_ivl (), reconnect_timer_id);
+    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;
 }