summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMartin Sustrik <sustrik@fastmq.commkdir>2009-08-09 16:12:09 +0200
committerMartin Sustrik <sustrik@fastmq.commkdir>2009-08-09 16:12:09 +0200
commit9f1f823b7b69ced56bdb0416feef71230cc7fd55 (patch)
treee5aa6ca2b943c62afcc4ca427102693d8f88a14e
parent3147ff8523d9736039c4582f5f62c323b23940d1 (diff)
zmq_listener/zmq_connecter implemented
-rw-r--r--src/Makefile.am2
-rw-r--r--src/io_object.cpp13
-rw-r--r--src/io_object.hpp18
-rw-r--r--src/socket_base.cpp12
-rw-r--r--src/tcp_connecter.cpp24
-rw-r--r--src/tcp_connecter.hpp9
-rw-r--r--src/zmq_connecter.cpp110
-rw-r--r--src/zmq_connecter.hpp70
-rw-r--r--src/zmq_listener.cpp9
-rw-r--r--src/zmq_listener.hpp5
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.