reconnection mechanism added to tcp transport

14 files changed, 130 insertions, 37 deletions

diff --git a/src/i_inout.hpp b/src/i_inout.hpp
index 6be6169..b82a476 100644
--- a/src/i_inout.hpp
+++ b/src/i_inout.hpp
@@ -36,8 +36,19 @@ namespace zmq
         //  Flush all the previously written messages downstream.
         virtual void flush () = 0;
     
-        //  Drop all the references to the engine.
-        virtual void detach () = 0;
+        //  Drop all the references to the engine. The parameter is the object
+        //  to use to reconnect. If reconnection is not required, the argument
+        //  is set to NULL.
+        virtual void detach (class owned_t *reconnecter_) = 0;
+
+        //  Returns least loaded I/O thread.
+        virtual class io_thread_t *get_io_thread () = 0;
+
+        //  Return pointer to the owning socket.
+        virtual class socket_base_t *get_owner () = 0;
+
+        //  Returns the name of associated session.
+        virtual const char *get_session_name () = 0;
     };
 
 }
diff --git a/src/io_object.hpp b/src/io_object.hpp
index 2ed5e24..7b66687 100644
--- a/src/io_object.hpp
+++ b/src/io_object.hpp
@@ -22,6 +22,7 @@
 
 #include <stddef.h>
 
+#include "stdint.hpp"
 #include "poller.hpp"
 #include "i_poll_events.hpp"
 
diff --git a/src/io_thread.hpp b/src/io_thread.hpp
index 457cdbf..c0dd77b 100644
--- a/src/io_thread.hpp
+++ b/src/io_thread.hpp
@@ -22,6 +22,7 @@
 
 #include <vector>
 
+#include "stdint.hpp"
 #include "object.hpp"
 #include "poller.hpp"
 #include "i_poll_events.hpp"
diff --git a/src/object.hpp b/src/object.hpp
index 84a57cd..496fd49 100644
--- a/src/object.hpp
+++ b/src/object.hpp
@@ -55,9 +55,10 @@ namespace zmq
         void unregister_endpoints (class socket_base_t *socket_);
         class socket_base_t *find_endpoint (const char *addr_);
 
-        //  Derived object can use following functions to interact with
-        //  global repositories. See dispatcher.hpp for function details.
+        //  Returns number of thead slots in the dispatcher.
         int thread_slot_count ();
+
+        //  Chooses least loaded I/O thread.
         class io_thread_t *choose_io_thread (uint64_t taskset_);
 
         //  Derived object can use these functions to send commands
diff --git a/src/session.cpp b/src/session.cpp
index 912cfa8..cbcc883 100644
--- a/src/session.cpp
+++ b/src/session.cpp
@@ -65,9 +65,13 @@ void zmq::session_t::flush ()
         out_pipe->flush ();
 }
 
-void zmq::session_t::detach ()
+void zmq::session_t::detach (owned_t *reconnecter_)
 {
-    //  TODO:  Start reconnection process here.
+    //  Plug in the reconnecter object if any.
+    if (reconnecter_) {
+        send_plug (reconnecter_);
+        send_own (owner, reconnecter_);
+    }
 
     //  Engine is terminating itself. No need to deallocate it from here.
     engine = NULL;
@@ -77,6 +81,21 @@ void zmq::session_t::detach ()
         term ();
 }
 
+zmq::io_thread_t *zmq::session_t::get_io_thread ()
+{
+    return choose_io_thread (options.affinity);
+}
+
+class zmq::socket_base_t *zmq::session_t::get_owner ()
+{
+    return owner;
+}
+
+const char *zmq::session_t::get_session_name ()
+{
+    return name.c_str ();
+}
+
 void zmq::session_t::attach_pipes (class reader_t *inpipe_,
     class writer_t *outpipe_)
 {
diff --git a/src/session.hpp b/src/session.hpp
index 64900f2..72e1d59 100644
--- a/src/session.hpp
+++ b/src/session.hpp
@@ -41,7 +41,10 @@ namespace zmq
         bool read (::zmq_msg_t *msg_);
         bool write (::zmq_msg_t *msg_);
         void flush ();
-        void detach ();
+        void detach (owned_t *reconnecter_);
+        class io_thread_t *get_io_thread ();
+        class socket_base_t *get_owner ();
+        const char *get_session_name ();
 
         //  i_endpoint interface implementation.
         void attach_pipes (class reader_t *inpipe_, class writer_t *outpipe_);
diff --git a/src/zmq_connecter.cpp b/src/zmq_connecter.cpp
index 8d5c59f..cd7d1b5 100644
--- a/src/zmq_connecter.cpp
+++ b/src/zmq_connecter.cpp
@@ -38,9 +38,13 @@ zmq::zmq_connecter_t::~zmq_connecter_t ()
 {
 }
 
-int zmq::zmq_connecter_t::set_address (const char *addr_)
+int zmq::zmq_connecter_t::set_address (const char *address_)
 {
-     return tcp_connecter.set_address (addr_);
+     int rc = tcp_connecter.set_address (address_);
+     if (rc != 0)
+         return rc;
+     address = address_;
+     return 0;
 }
 
 void zmq::zmq_connecter_t::process_plug ()
@@ -84,7 +88,7 @@ void zmq::zmq_connecter_t::out_event ()
     //  Create an init object. 
     io_thread_t *io_thread = choose_io_thread (options.affinity);
     zmq_connecter_init_t *init = new zmq_connecter_init_t (io_thread, owner,
-        fd, options, session_name.c_str ());
+        fd, options, session_name.c_str (), address.c_str ());
     zmq_assert (init);
     send_plug (init);
     send_own (owner, init);
diff --git a/src/zmq_connecter.hpp b/src/zmq_connecter.hpp
index c443ce8..acd3352 100644
--- a/src/zmq_connecter.hpp
+++ b/src/zmq_connecter.hpp
@@ -40,7 +40,7 @@ namespace zmq
         ~zmq_connecter_t ();
 
         //  Set IP address to connect to.
-        int set_address (const char *addr_);
+        int set_address (const char *address_);
 
     private:
 
@@ -75,6 +75,9 @@ namespace zmq
         //  Name of the session associated with the connecter.
         std::string session_name;
 
+        //  Address to connect to.
+        std::string address;
+
         zmq_connecter_t (const zmq_connecter_t&);
         void operator = (const zmq_connecter_t&);
     };
diff --git a/src/zmq_connecter_init.cpp b/src/zmq_connecter_init.cpp
index ea6a8c0..1418429 100644
--- a/src/zmq_connecter_init.cpp
+++ b/src/zmq_connecter_init.cpp
@@ -25,13 +25,13 @@
 
 zmq::zmq_connecter_init_t::zmq_connecter_init_t (io_thread_t *parent_,
       socket_base_t *owner_, fd_t fd_, const options_t &options_,
-      const char *session_name_) :
+      const char *session_name_, const char *address_) :
     owned_t (parent_, owner_),
     options (options_),
     session_name (session_name_)
 {
     //  Create associated engine object.
-    engine = new zmq_engine_t (parent_, fd_, options);
+    engine = new zmq_engine_t (parent_, fd_, options, true, address_);
     zmq_assert (engine);
 }
 
@@ -87,27 +87,33 @@ void zmq::zmq_connecter_init_t::flush ()
     //  We are not expecting any messages. No point in flushing.
 }
 
-void zmq::zmq_connecter_init_t::detach ()
+void zmq::zmq_connecter_init_t::detach (owned_t *reconnecter_)
 {
-    //  TODO: Start reconnection process here.
-/*
-    //  Create a connecter object to attempt reconnect. Ask it to wait for a
-    //  while before reconnecting.
-    io_thread_t *io_thread = choose_io_thread (options.affinity);
-    zmq_connecter_t *connecter = new zmq_connecter_t (io_thread, owner,
-        options, session_name.c_str (), true);
-    connecter->set_address (...);
-    zmq_assert (connecter);
-    send_plug (connecter);
-    send_own (owner, connecter);
-*/
+    //  Plug in the reconnecter object.
+    zmq_assert (reconnecter_);
+    send_plug (reconnecter_);
+    send_own (owner, reconnecter_);
 
     //  This function is called by engine when disconnection occurs.
     //  The engine will destroy itself, so we just drop the pointer here and
     //  start termination of the init object.
     engine = NULL;
     term ();
+}
+
+zmq::io_thread_t *zmq::zmq_connecter_init_t::get_io_thread ()
+{
+    return choose_io_thread (options.affinity);
+}
 
+class zmq::socket_base_t *zmq::zmq_connecter_init_t::get_owner ()
+{
+    return owner;
+}
+
+const char *zmq::zmq_connecter_init_t::get_session_name ()
+{
+    return session_name.c_str ();
 }
 
 void zmq::zmq_connecter_init_t::process_plug ()
diff --git a/src/zmq_connecter_init.hpp b/src/zmq_connecter_init.hpp
index 3f42fc6..03ccd24 100644
--- a/src/zmq_connecter_init.hpp
+++ b/src/zmq_connecter_init.hpp
@@ -40,7 +40,8 @@ namespace zmq
     public:
 
         zmq_connecter_init_t (class io_thread_t *parent_, socket_base_t *owner_,
-            fd_t fd_, const options_t &options, const char *session_name_);
+            fd_t fd_, const options_t &options, const char *session_name_,
+            const char *address_);
         ~zmq_connecter_init_t ();
 
     private:
@@ -49,7 +50,10 @@ namespace zmq
         bool read (::zmq_msg_t *msg_);
         bool write (::zmq_msg_t *msg_);
         void flush ();
-        void detach ();
+        void detach (owned_t *reconnecter_);
+        class io_thread_t *get_io_thread ();
+        class socket_base_t *get_owner ();
+        const char *get_session_name ();
 
         //  Handlers for incoming commands.
         void process_plug ();
@@ -63,8 +67,7 @@ namespace zmq
         //  Associated socket options.
         options_t options;
 
-        //  Name of the session to bind new connection to. Makes sense only
-        //  when 'connected' is true.
+        //  Name of the session to bind new connection to.
         std::string session_name;
 
         zmq_connecter_init_t (const zmq_connecter_init_t&);
diff --git a/src/zmq_engine.cpp b/src/zmq_engine.cpp
index 18fc616..6b439f5 100644
--- a/src/zmq_engine.cpp
+++ b/src/zmq_engine.cpp
@@ -25,7 +25,7 @@
 #include "err.hpp"
 
 zmq::zmq_engine_t::zmq_engine_t (io_thread_t *parent_, fd_t fd_,
-      const options_t &options_) :
+      const options_t &options_, bool reconnect_, const char *address_) :
     io_object_t (parent_),
     inpos (NULL),
     insize (0),
@@ -34,8 +34,12 @@ zmq::zmq_engine_t::zmq_engine_t (io_thread_t *parent_, fd_t fd_,
     outsize (0),
     encoder (out_batch_size, false),
     inout (NULL),
-    options (options_)
+    options (options_),
+    reconnect (reconnect_)
 {
+    if (reconnect)
+        address = address_;
+
     //  Initialise the underlying socket.
     int rc = tcp_socket.open (fd_, options.sndbuf, options.rcvbuf);
     zmq_assert (rc == 0);
@@ -145,7 +149,20 @@ void zmq::zmq_engine_t::revive ()
 void zmq::zmq_engine_t::error ()
 {
     zmq_assert (inout);
-    inout->detach ();
+
+    zmq_connecter_t *reconnecter = NULL;
+    if (reconnect) {
+
+        //  Create a connecter object to attempt reconnect.
+        //  Ask it to wait for a while before reconnecting.
+        reconnecter = new zmq_connecter_t (
+            inout->get_io_thread (), inout->get_owner (),
+            options, inout->get_session_name (), true);
+        zmq_assert (reconnecter);
+        reconnecter->set_address (address.c_str ());
+    }
+
+    inout->detach (reconnecter);
     unplug ();
     delete this;
 }
diff --git a/src/zmq_engine.hpp b/src/zmq_engine.hpp
index 8d9e4b9..63e1539 100644
--- a/src/zmq_engine.hpp
+++ b/src/zmq_engine.hpp
@@ -37,7 +37,7 @@ namespace zmq
     public:
 
         zmq_engine_t (class io_thread_t *parent_, fd_t fd_,
-            const options_t &options_);
+            const options_t &options_, bool reconnect_, const char *address_);
         ~zmq_engine_t ();
 
         //  i_engine interface implementation.
@@ -69,6 +69,9 @@ namespace zmq
 
         options_t options;
 
+        bool reconnect;
+        std::string address;
+
         zmq_engine_t (const zmq_engine_t&);
         void operator = (const zmq_engine_t&);
     };
diff --git a/src/zmq_listener_init.cpp b/src/zmq_listener_init.cpp
index 0c9f0ee..a463dc8 100644
--- a/src/zmq_listener_init.cpp
+++ b/src/zmq_listener_init.cpp
@@ -29,7 +29,7 @@ zmq::zmq_listener_init_t::zmq_listener_init_t (io_thread_t *parent_,
     has_peer_identity (false)
 {
     //  Create associated engine object.
-    engine = new zmq_engine_t (parent_, fd_, options);
+    engine = new zmq_engine_t (parent_, fd_, options, false, NULL);
     zmq_assert (engine);
 }
 
@@ -93,8 +93,11 @@ void zmq::zmq_listener_init_t::flush ()
     term ();
 }
 
-void zmq::zmq_listener_init_t::detach ()
+void zmq::zmq_listener_init_t::detach (owned_t *reconnecter_)
 {
+    //  On the listening side of the connection we are never reconnecting.
+    zmq_assert (reconnecter_ == NULL);
+
     //  This function is called by engine when disconnection occurs.
     //  The engine will destroy itself, so we just drop the pointer here and
     //  start termination of the init object.
@@ -102,6 +105,21 @@ void zmq::zmq_listener_init_t::detach ()
     term ();
 }
 
+zmq::io_thread_t *zmq::zmq_listener_init_t::get_io_thread ()
+{
+    return choose_io_thread (options.affinity);
+}
+
+class zmq::socket_base_t *zmq::zmq_listener_init_t::get_owner ()
+{
+    return owner;
+}
+
+const char *zmq::zmq_listener_init_t::get_session_name ()
+{
+    zmq_assert (false);
+}
+
 void zmq::zmq_listener_init_t::process_plug ()
 {
     zmq_assert (engine);
diff --git a/src/zmq_listener_init.hpp b/src/zmq_listener_init.hpp
index 885b36b..d7fde02 100644
--- a/src/zmq_listener_init.hpp
+++ b/src/zmq_listener_init.hpp
@@ -49,7 +49,10 @@ namespace zmq
         bool read (::zmq_msg_t *msg_);
         bool write (::zmq_msg_t *msg_);
         void flush ();
-        void detach ();
+        void detach (owned_t *reconnecter_);
+        class io_thread_t *get_io_thread ();
+        class socket_base_t *get_owner ();
+        const char *get_session_name ();
 
         //  Handlers for incoming commands.
         void process_plug ();