zmq_listener/zmq_connecter implemented

10 files changed, 239 insertions, 33 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 674f307..5d51725 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -44,6 +44,7 @@ libzmq_la_SOURCES = \
     ypipe.hpp \
     ypollset.hpp \
     yqueue.hpp \
+    zmq_connecter.hpp \
     zmq_engine.hpp \
     zmq_listener.hpp \
     app_thread.cpp \
@@ -67,6 +68,7 @@ libzmq_la_SOURCES = \
     uuid.cpp \
     ypollset.cpp \
     zmq.cpp \
+    zmq_connecter.cpp \
     zmq_engine.cpp \
     zmq_listener.cpp
 
diff --git a/src/io_object.cpp b/src/io_object.cpp
index b7c70d4..d8cc1c0 100644
--- a/src/io_object.cpp
+++ b/src/io_object.cpp
@@ -52,9 +52,9 @@ void zmq::io_object_t::process_plug ()
     plugged_in = true;
 }
 
-zmq::handle_t zmq::io_object_t::add_fd (fd_t fd_, i_poll_events *events_)
+zmq::handle_t zmq::io_object_t::add_fd (fd_t fd_)
 {
-    return poller->add_fd (fd_, events_);
+    return poller->add_fd (fd_, this);
 }
 
 void zmq::io_object_t::rm_fd (handle_t handle_)
@@ -82,14 +82,14 @@ void zmq::io_object_t::reset_pollout (handle_t handle_)
     poller->reset_pollout (handle_);
 }
 
-void zmq::io_object_t::add_timer (i_poll_events *events_)
+void zmq::io_object_t::add_timer ()
 {
-    poller->add_timer (events_);
+    poller->add_timer (this);
 }
 
-void zmq::io_object_t::cancel_timer (i_poll_events *events_)
+void zmq::io_object_t::cancel_timer ()
 {
-    poller->cancel_timer (events_);
+    poller->cancel_timer (this);
 }
 
 void zmq::io_object_t::in_event ()
@@ -126,5 +126,6 @@ void zmq::io_object_t::process_term ()
     //  Otherwise, destroy the object and acknowledge the termination
     //  straight away.
     send_term_ack (owner);
+    process_unplug ();
     delete this;
 }
diff --git a/src/io_object.hpp b/src/io_object.hpp
index 2029bfb..4f323ad 100644
--- a/src/io_object.hpp
+++ b/src/io_object.hpp
@@ -51,15 +51,20 @@ namespace zmq
         //  handler.
         void process_plug ();
 
+        //  io_object_t defines a new handler used to disconnect the object
+        //  from the poller object. Implement the handlen in the derived
+        //  classes to ensure sane cleanup.
+        virtual void process_unplug () = 0;
+
         //  Methods to access underlying poller object.
-        handle_t add_fd (fd_t fd_, struct i_poll_events *events_);
+        handle_t add_fd (fd_t fd_);
         void rm_fd (handle_t handle_);
         void set_pollin (handle_t handle_);
         void reset_pollin (handle_t handle_);
         void set_pollout (handle_t handle_);
         void reset_pollout (handle_t handle_);
-        void add_timer (struct i_poll_events *events_);
-        void cancel_timer (struct i_poll_events *events_);
+        void add_timer ();
+        void cancel_timer ();
 
         //  i_poll_events interface implementation.
         void in_event ();
@@ -70,12 +75,11 @@ namespace zmq
         //  it when it's being closed.
         object_t *owner;
 
-        //  Set to true when object is plugged in. It's responsibility
-        //  of derived object to set the property after the feat.
-        bool plugged_in;
-
     private:
 
+        //  Set to true when object is plugged in.
+        bool plugged_in;
+
         //  Set to true when object was terminated before it was plugged in.
         //  In such case destruction is delayed till 'plug' command arrives.
         bool terminated;
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index ac3b4b9..c179a93 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -25,6 +25,7 @@
 #include "app_thread.hpp"
 #include "err.hpp"
 #include "zmq_listener.hpp"
+#include "zmq_connecter.hpp"
 #include "io_thread.hpp"
 #include "config.hpp"
 
@@ -127,7 +128,6 @@ int zmq::socket_base_t::setsockopt (int option_, void *optval_,
 
 int zmq::socket_base_t::bind (const char *addr_)
 {
-    //  TODO: The taskset should be taken from socket options.
     zmq_listener_t *listener =
         new zmq_listener_t (choose_io_thread (affinity), this);
     int rc = listener->set_address (addr_);
@@ -141,7 +141,15 @@ int zmq::socket_base_t::bind (const char *addr_)
 
 int zmq::socket_base_t::connect (const char *addr_)
 {
-    zmq_assert (false);
+    zmq_connecter_t *connecter =
+        new zmq_connecter_t (choose_io_thread (affinity), this);
+    int rc = connecter->set_address (addr_);
+    if (rc != 0)
+        return -1;
+
+    send_plug (connecter);
+    send_own (this, connecter);
+    return 0;
 }
 
 int zmq::socket_base_t::subscribe (const char *criteria_)
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp
index 7d29019..23f3741 100644
--- a/src/tcp_connecter.cpp
+++ b/src/tcp_connecter.cpp
@@ -17,6 +17,8 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
+#include <string>
+
 #include "tcp_connecter.hpp"
 #include "platform.hpp"
 #include "ip.hpp"
@@ -40,6 +42,7 @@
 zmq::tcp_connecter_t::tcp_connecter_t () :
     s (retired_fd)
 {
+    memset (&addr, 0, sizeof (addr));
 }
 
 zmq::tcp_connecter_t::~tcp_connecter_t ()
@@ -48,15 +51,15 @@ zmq::tcp_connecter_t::~tcp_connecter_t ()
         close ();
 }
 
-int zmq::tcp_connecter_t::open (const char *addr_)
+int zmq::tcp_connecter_t::set_address (const char *addr_)
 {
-    zmq_assert (s == retired_fd);
-
     //  Convert the hostname into sockaddr_in structure.
-    sockaddr_in address;
-    int rc = resolve_ip_hostname (&address, addr_);
-    if (rc != 0)
-        return -1;
+    return resolve_ip_hostname (&addr, addr_);
+}
+
+int zmq::tcp_connecter_t::open ()
+{
+    zmq_assert (s == retired_fd);
 
     //  Create the socket.
     s = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
@@ -67,7 +70,7 @@ int zmq::tcp_connecter_t::open (const char *addr_)
     int flags = fcntl (s, F_GETFL, 0);
     if (flags == -1) 
         flags = 0;
-    rc = fcntl (s, F_SETFL, flags | O_NONBLOCK);
+    int rc = fcntl (s, F_SETFL, flags | O_NONBLOCK);
     errno_assert (rc != -1);
 
     //  Disable Nagle's algorithm.
@@ -83,7 +86,7 @@ int zmq::tcp_connecter_t::open (const char *addr_)
 #endif
 
     //  Connect to the remote peer.
-    rc = ::connect (s, (sockaddr*) &address, sizeof address);
+    rc = ::connect (s, (sockaddr*) &addr, sizeof (addr));
 
     //  Connect was successfull immediately.
     if (rc == 0)
@@ -91,7 +94,7 @@ int zmq::tcp_connecter_t::open (const char *addr_)
 
     //  Asynchronous connect was launched.
     if (rc == -1 && errno == EINPROGRESS)
-        return 1;
+        return -1;
 
     //  Error occured.
     int err = errno;
@@ -125,7 +128,6 @@ zmq::fd_t zmq::tcp_connecter_t::connect ()
     if (rc == -1)
         err = errno;
     if (err != 0) {
-        close ();
         errno = err;
         return retired_fd;
     }
diff --git a/src/tcp_connecter.hpp b/src/tcp_connecter.hpp
index aa1ef05..bb358fd 100644
--- a/src/tcp_connecter.hpp
+++ b/src/tcp_connecter.hpp
@@ -21,6 +21,7 @@
 #define __ZMQ_TCP_CONNECTER_HPP_INCLUDED__
 
 #include "fd.hpp"
+#include "ip.hpp"
 
 namespace zmq
 {
@@ -34,11 +35,14 @@ namespace zmq
         tcp_connecter_t ();
         ~tcp_connecter_t ();
 
+        //  Set IP address/port to connect to.
+        int set_address (const char *addr_);
+
         //  Open TCP connecting socket. Address is in
         //  <hostname>:<port-number> format. Returns -1 in case of error,
         //  0 if connect was successfull immediately and 1 if async connect
         //  was launched.
-        int open (const char *addr_);
+        int open ();
 
         //  Close the connecting socket.
         int close ();
@@ -53,6 +57,9 @@ namespace zmq
 
     private:
 
+        //  Address to connect to.
+        sockaddr_in addr;
+
         //  Underlying socket.
         fd_t s;
 
diff --git a/src/zmq_connecter.cpp b/src/zmq_connecter.cpp
new file mode 100644
index 0000000..513508d
--- /dev/null
+++ b/src/zmq_connecter.cpp
@@ -0,0 +1,110 @@
+/*
+    Copyright (c) 2007-2009 FastMQ Inc.
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the Lesser GNU 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
+    Lesser GNU General Public License for more details.
+
+    You should have received a copy of the Lesser GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "zmq_connecter.hpp"
+#include "err.hpp"
+
+zmq::zmq_connecter_t::zmq_connecter_t (io_thread_t *parent_, object_t *owner_) :
+    io_object_t (parent_, owner_),
+    waiting (false)
+{
+}
+
+zmq::zmq_connecter_t::~zmq_connecter_t ()
+{
+}
+
+int zmq::zmq_connecter_t::set_address (const char *addr_)
+{
+     return tcp_connecter.set_address (addr_);
+}
+
+void zmq::zmq_connecter_t::process_plug ()
+{
+    start_connecting ();
+    io_object_t::process_plug ();
+}
+
+void zmq::zmq_connecter_t::process_unplug ()
+{
+    if (!waiting)
+        rm_fd (handle);
+}
+
+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 ();
+
+    //  If there was error during the connecting, close the socket and wait
+    //  for a while before trying to reconnect.
+    if (fd == retired_fd) {
+        rm_fd (handle);
+        tcp_connecter.close ();
+        waiting = true;
+        add_timer ();
+        return;
+    }
+
+    zmq_assert (false);
+
+/*
+    object_t *engine = new zmq_engine_t (choose_io_thread (0), owner);
+    send_plug (engine);
+    send_own (owner, engine);
+*/
+}
+
+void zmq::zmq_connecter_t::timer_event ()
+{
+    //  Reconnect period have elapsed.
+    waiting = 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) {
+        out_event ();
+        return;
+    }
+
+    //  Connection establishment may be dealyed. Poll for its completion.
+    else if (rc == -1 && errno == EINPROGRESS) {
+        handle = add_fd (tcp_connecter.get_fd ());
+        set_pollout (handle);
+        return;
+    }
+
+    //  If none of the above is true, synchronous error occured.
+    //  Wait for a while and retry.
+    waiting = true;
+    add_timer ();
+}
diff --git a/src/zmq_connecter.hpp b/src/zmq_connecter.hpp
new file mode 100644
index 0000000..d346396
--- /dev/null
+++ b/src/zmq_connecter.hpp
@@ -0,0 +1,70 @@
+/*
+    Copyright (c) 2007-2009 FastMQ Inc.
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the Lesser GNU 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
+    Lesser GNU General Public License for more details.
+
+    You should have received a copy of the Lesser GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_ZMQ_CONNECTER_HPP_INCLUDED__
+#define __ZMQ_ZMQ_CONNECTER_HPP_INCLUDED__
+
+#include "io_object.hpp"
+#include "tcp_connecter.hpp"
+
+namespace zmq
+{
+
+    class zmq_connecter_t : public io_object_t
+    {
+    public:
+
+        zmq_connecter_t (class io_thread_t *parent_, object_t *owner_);
+
+        //  Set IP address to connect to.
+        int set_address (const char *addr_);
+
+    private:
+
+        ~zmq_connecter_t ();
+
+        //  Handlers for incoming commands.
+        void process_plug ();
+        void process_unplug ();
+
+        //  Handlers for I/O events.
+        void in_event ();
+        void out_event ();
+        void timer_event ();
+
+        //  Internal function to start the actual connection establishment.
+        void start_connecting ();
+
+        //  Actual connecting socket.
+        tcp_connecter_t tcp_connecter;
+
+        //  Handle corresponding to the listening socket.
+        handle_t handle;
+
+        //  True, if we are waiting for a period of time before trying to
+        //  reconnect.
+        bool waiting;
+
+        zmq_connecter_t (const zmq_connecter_t&);
+        void operator = (const zmq_connecter_t&);
+    };
+
+}
+
+#endif
diff --git a/src/zmq_listener.cpp b/src/zmq_listener.cpp
index 9c9bbe9..463a1dd 100644
--- a/src/zmq_listener.cpp
+++ b/src/zmq_listener.cpp
@@ -29,8 +29,6 @@ zmq::zmq_listener_t::zmq_listener_t (io_thread_t *parent_, object_t *owner_) :
 
 zmq::zmq_listener_t::~zmq_listener_t ()
 {
-    if (plugged_in)
-        rm_fd (handle);
 }
 
 int zmq::zmq_listener_t::set_address (const char *addr_)
@@ -45,12 +43,17 @@ void zmq::zmq_listener_t::process_plug ()
     zmq_assert (rc == 0);
 
     //  Start polling for incoming connections.
-    handle = add_fd (tcp_listener.get_fd (), this);
+    handle = add_fd (tcp_listener.get_fd ());
     set_pollin (handle);
 
     io_object_t::process_plug ();
 }
 
+void zmq::zmq_listener_t::process_unplug ()
+{
+    rm_fd (handle);
+}
+
 void zmq::zmq_listener_t::in_event ()
 {
     fd_t fd = tcp_listener.accept ();
diff --git a/src/zmq_listener.hpp b/src/zmq_listener.hpp
index 74d42e1..ef252b3 100644
--- a/src/zmq_listener.hpp
+++ b/src/zmq_listener.hpp
@@ -20,8 +20,6 @@
 #ifndef __ZMQ_ZMQ_LISTENER_HPP_INCLUDED__
 #define __ZMQ_ZMQ_LISTENER_HPP_INCLUDED__
 
-#include <string>
-
 #include "io_object.hpp"
 #include "tcp_listener.hpp"
 
@@ -43,8 +41,9 @@ namespace zmq
 
         //  Handlers for incoming commands.
         void process_plug ();
+        void process_unplug ();
 
-        //  Handle I/O events.
+        //  Handlers for I/O events.
         void in_event ();
 
         //  Actual listening socket.