summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorMartin Sustrik <sustrik@250bpm.com>2011-07-26 18:35:40 +0200
committerMartin Sustrik <sustrik@250bpm.com>2011-07-26 18:35:40 +0200
commit279302c5f54ddf8a23b1eaacee63c3158850d9ff (patch)
tree501200a6a2189fb9ed0a158cb825c653343729d9 /src
parent9906c652ca01698dcc429c9045eb040c24c4bb8d (diff)
Experimental VTCP listener added
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
Diffstat (limited to 'src')
-rw-r--r--src/Makefile.am4
-rw-r--r--src/socket_base.cpp53
-rw-r--r--src/tcp_connecter.cpp46
-rw-r--r--src/tcp_connecter.hpp12
-rw-r--r--src/tcp_listener.cpp1
-rw-r--r--src/vtcp_listener.cpp111
-rw-r--r--src/vtcp_listener.hpp71
7 files changed, 278 insertions, 20 deletions
diff --git a/src/Makefile.am b/src/Makefile.am
index 1354174..1da5be9 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -64,6 +64,8 @@ libzmq_la_SOURCES = \
tcp_listener.hpp \
thread.hpp \
trie.hpp \
+ vtcp_connecter.hpp \
+ vtcp_listener.hpp \
windows.hpp \
wire.hpp \
xpub.hpp \
@@ -117,6 +119,8 @@ libzmq_la_SOURCES = \
tcp_listener.cpp \
thread.cpp \
trie.cpp \
+ vtcp_connecter.cpp \
+ vtcp_listener.cpp \
xpub.cpp \
xrep.cpp \
xreq.cpp \
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index b97e3a7..fb0f5fd 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -35,6 +35,7 @@
#include "socket_base.hpp"
#include "tcp_listener.hpp"
+#include "vtcp_listener.hpp"
#include "tcp_connecter.hpp"
#include "io_thread.hpp"
#include "session.hpp"
@@ -174,7 +175,8 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_)
{
// First check out whether the protcol is something we are aware of.
if (protocol_ != "inproc" && protocol_ != "ipc" && protocol_ != "tcp" &&
- protocol_ != "pgm" && protocol_ != "epgm" && protocol_ != "sys") {
+ protocol_ != "pgm" && protocol_ != "epgm" && protocol_ != "sys" &&
+ protocol_ != "vtcp") {
errno = EPROTONOSUPPORT;
return -1;
}
@@ -188,6 +190,14 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_)
}
#endif
+ // If 0MQ is not compiled with VTCP, vtcp transport is not avaialble.
+#if !defined ZMQ_HAVE_VTCP
+ if (protocol_ == "vtcp") {
+ errno = EPROTONOSUPPORT;
+ return -1;
+ }
+#endif
+
// IPC transport is not available on Windows and OpenVMS.
#if defined ZMQ_HAVE_WINDOWS || defined ZMQ_HAVE_OPENVMS
if (protocol_ == "ipc") {
@@ -338,16 +348,22 @@ int zmq::socket_base_t::bind (const char *addr_)
return register_endpoint (addr_, endpoint);
}
- if (protocol == "tcp" || protocol == "ipc") {
+ if (protocol == "pgm" || protocol == "epgm") {
- // Choose I/O thread to run the listerner in.
- io_thread_t *io_thread = choose_io_thread (options.affinity);
- if (!io_thread) {
- errno = EMTHREAD;
- return -1;
- }
+ // For convenience's sake, bind can be used interchageable with
+ // connect for PGM and EPGM transports.
+ return connect (addr_);
+ }
- // Create and run the listener.
+ // Remaining trasnports require to be run in an I/O thread, so at this
+ // point we'll choose one.
+ io_thread_t *io_thread = choose_io_thread (options.affinity);
+ if (!io_thread) {
+ errno = EMTHREAD;
+ return -1;
+ }
+
+ if (protocol == "tcp" || protocol == "ipc") {
tcp_listener_t *listener = new (std::nothrow) tcp_listener_t (
io_thread, this, options);
alloc_assert (listener);
@@ -357,16 +373,23 @@ int zmq::socket_base_t::bind (const char *addr_)
return -1;
}
launch_child (listener);
-
return 0;
}
- if (protocol == "pgm" || protocol == "epgm") {
-
- // For convenience's sake, bind can be used interchageable with
- // connect for PGM and EPGM transports.
- return connect (addr_);
+#if defined ZMQ_HAVE_VTCP
+ if (protocol == "vtcp") {
+ vtcp_listener_t *listener = new (std::nothrow) vtcp_listener_t (
+ io_thread, this, options);
+ alloc_assert (listener);
+ int rc = listener->set_address (address.c_str ());
+ if (rc != 0) {
+ delete listener;
+ return -1;
+ }
+ launch_child (listener);
+ return 0;
}
+#endif
zmq_assert (false);
return -1;
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp
index 678d488..b530e8e 100644
--- a/src/tcp_connecter.cpp
+++ b/src/tcp_connecter.cpp
@@ -23,6 +23,52 @@
#include "tcp_connecter.hpp"
#include "tcp_engine.hpp"
+#include "platform.hpp"
+#include "ip.hpp"
+#include "err.hpp"
+
+#if defined ZMQ_HAVE_WINDOWS
+#include "windows.hpp"
+#else
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <arpa/inet.h>
+#include <netinet/tcp.h>
+#include <netinet/in.h>
+#include <netdb.h>
+#include <fcntl.h>
+#ifdef ZMQ_HAVE_OPENVMS
+#include <ioctl.h>
+#endif
+#endif
+
+
+/*
+ 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 <http://www.gnu.org/licenses/>.
+*/
+
+#include <new>
+#include <string>
+
+#include "tcp_connecter.hpp"
+#include "tcp_engine.hpp"
#include "io_thread.hpp"
#include "platform.hpp"
#include "ip.hpp"
diff --git a/src/tcp_connecter.hpp b/src/tcp_connecter.hpp
index 2168ca7..524e9cc 100644
--- a/src/tcp_connecter.hpp
+++ b/src/tcp_connecter.hpp
@@ -18,8 +18,8 @@
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
-#ifndef __ZMQ_TCP_CONNECTER_HPP_INCLUDED__
-#define __ZMQ_TCP_CONNECTER_HPP_INCLUDED__
+#ifndef __TCP_CONNECTER_HPP_INCLUDED__
+#define __TCP_CONNECTER_HPP_INCLUDED__
#include "fd.hpp"
#include "ip.hpp"
@@ -30,11 +30,14 @@
namespace zmq
{
+ // A base class for different connecters. It handles auto-reconnection
+ // on behalf of the derived class.
+
class tcp_connecter_t : public own_t, public io_object_t
{
public:
- // If 'wait' is true connecter first waits for a while, then starts
+ // If 'delay' is true connecter first waits for a while, then starts
// connection process.
tcp_connecter_t (class io_thread_t *io_thread_,
class session_t *session_, const options_t &options_,
@@ -68,8 +71,7 @@ namespace zmq
// Set address to connect to.
int set_address (const char *protocol, const char *addr_);
- // Open TCP connecting socket. Address is in
- // <hostname>:<port-number> format. Returns -1 in case of error,
+ // Open TCP connecting socket. Returns -1 in case of error,
// 0 if connect was successfull immediately and 1 if async connect
// was launched.
int open ();
diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp
index b409891..fbb1b85 100644
--- a/src/tcp_listener.cpp
+++ b/src/tcp_listener.cpp
@@ -86,6 +86,7 @@ void zmq::tcp_listener_t::in_event ()
// TODO: Handle specific errors like ENFILE/EMFILE etc.
if (fd == retired_fd)
return;
+
// Create the engine object for this connection.
tcp_engine_t *engine = new (std::nothrow) tcp_engine_t (fd, options);
alloc_assert (engine);
diff --git a/src/vtcp_listener.cpp b/src/vtcp_listener.cpp
new file mode 100644
index 0000000..31fb9ac
--- /dev/null
+++ b/src/vtcp_listener.cpp
@@ -0,0 +1,111 @@
+/*
+ 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 <http://www.gnu.org/licenses/>.
+*/
+
+#include "vtcp_listener.hpp"
+
+#if defined ZMQ_HAVE_VTCP
+
+#include <string>
+#include <string.h>
+#include <vtcp.h>
+
+#include "tcp_engine.hpp"
+#include "session.hpp"
+#include "stdint.hpp"
+#include "err.hpp"
+
+zmq::vtcp_listener_t::vtcp_listener_t (io_thread_t *io_thread_,
+ socket_base_t *socket_, options_t &options_) :
+ own_t (io_thread_, options_),
+ io_object_t (io_thread_),
+ s (retired_fd),
+ socket (socket_)
+{
+}
+
+zmq::vtcp_listener_t::~vtcp_listener_t ()
+{
+ zmq_assert (s != retired_fd);
+ int rc = ::close (s);
+ errno_assert (rc == 0);
+ s = retired_fd;
+}
+
+int zmq::vtcp_listener_t::set_address (const char *addr_)
+{
+ // Find the '.' at end that separates NIC name from service.
+ const char *delimiter = strrchr (addr_, '.');
+ if (!delimiter) {
+ errno = EINVAL;
+ return -1;
+ }
+
+ // Parse port and subport.
+ std::string port_str (addr_, delimiter - addr_);
+ std::string subport_str (delimiter + 1);
+ uint16_t port = (uint16_t) atoi (port_str.c_str ());
+ uint32_t subport = (uint32_t) atoi (subport_str.c_str ());
+
+ // Srart listening.
+ s = vtcp_bind (port, subport);
+ if (s == retired_fd)
+ return -1;
+
+ return 0;
+}
+
+void zmq::vtcp_listener_t::process_plug ()
+{
+ // Start polling for incoming connections.
+ handle = add_fd (s);
+ set_pollin (handle);
+}
+
+void zmq::vtcp_listener_t::process_term (int linger_)
+{
+ rm_fd (handle);
+ own_t::process_term (linger_);
+}
+
+void zmq::vtcp_listener_t::in_event ()
+{
+ fd_t fd = vtcp_acceptb (s);
+ if (fd == retired_fd)
+ return;
+
+ // Create the engine object for this connection.
+ tcp_engine_t *engine = new (std::nothrow) tcp_engine_t (fd, options);
+ alloc_assert (engine);
+
+ // Choose I/O thread to run connecter in. Given that we are already
+ // running in an I/O thread, there must be at least one available.
+ io_thread_t *io_thread = choose_io_thread (options.affinity);
+ zmq_assert (io_thread);
+
+ // Create and launch a session object.
+ session_t *session = new (std::nothrow)
+ session_t (io_thread, false, socket, options, NULL, NULL);
+ alloc_assert (session);
+ session->inc_seqnum ();
+ launch_child (session);
+ send_attach (session, engine, false);
+}
+
+#endif
diff --git a/src/vtcp_listener.hpp b/src/vtcp_listener.hpp
new file mode 100644
index 0000000..78f3b51
--- /dev/null
+++ b/src/vtcp_listener.hpp
@@ -0,0 +1,71 @@
+/*
+ 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 <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_VTCP_LISTENER_HPP_INCLUDED__
+#define __ZMQ_VTCP_LISTENER_HPP_INCLUDED__
+
+#include "platform.hpp"
+
+#if defined ZMQ_HAVE_VTCP
+
+#include "own.hpp"
+#include "io_object.hpp"
+#include "fd.hpp"
+
+namespace zmq
+{
+
+ class vtcp_listener_t : public own_t, public io_object_t
+ {
+ public:
+
+ vtcp_listener_t (class io_thread_t *io_thread_,
+ class socket_base_t *socket_, class options_t &options_);
+ ~vtcp_listener_t ();
+
+ int set_address (const char *addr_);
+
+ private:
+
+ // Handlers for incoming commands.
+ void process_plug ();
+ void process_term (int linger_);
+
+ // Handlers for I/O events.
+ void in_event ();
+
+ // VTCP listener socket.
+ fd_t s;
+
+ // Handle corresponding to the listening socket.
+ handle_t handle;
+
+ // Socket the listerner belongs to.
+ class socket_base_t *socket;
+
+ vtcp_listener_t (const vtcp_listener_t&);
+ const vtcp_listener_t &operator = (const vtcp_listener_t&);
+ };
+
+}
+
+#endif
+
+#endif