From 84f00c06d62338e78fb1e24d12ed78945ca08ad9 Mon Sep 17 00:00:00 2001 From: Pieter Hintjens Date: Wed, 6 Nov 2013 20:22:06 +0100 Subject: [PATCH] Revert "zmq: add support for TIPC transport" This reverts commit dbc42ee9af098183714845a75139997dcc34c7a0. --- src/Makefile.am | 8 +- src/address.cpp | 17 --- src/address.hpp | 6 - src/session_base.cpp | 11 -- src/socket_base.cpp | 43 +------ src/tipc_address.cpp | 105 ----------------- src/tipc_address.hpp | 67 ----------- src/tipc_connecter.cpp | 257 ----------------------------------------- src/tipc_connecter.hpp | 127 -------------------- src/tipc_listener.cpp | 178 ---------------------------- src/tipc_listener.hpp | 96 --------------- 11 files changed, 3 insertions(+), 912 deletions(-) delete mode 100644 src/tipc_address.cpp delete mode 100644 src/tipc_address.hpp delete mode 100644 src/tipc_connecter.cpp delete mode 100644 src/tipc_connecter.hpp delete mode 100644 src/tipc_listener.cpp delete mode 100644 src/tipc_listener.hpp diff --git a/src/Makefile.am b/src/Makefile.am index 3df44a40..340c997b 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -146,13 +146,7 @@ libzmq_la_SOURCES = \ v1_protocol.hpp \ xsub.cpp \ zmq.cpp \ - zmq_utils.cpp \ - tipc_address.cpp \ - tipc_address.hpp \ - tipc_listener.cpp \ - tipc_listener.hpp \ - tipc_connecter.cpp\ - tipc_connecter.hpp + zmq_utils.cpp if ON_MINGW libzmq_la_LDFLAGS = -no-undefined -avoid-version -version-info @LTVER@ @LIBZMQ_EXTRA_LDFLAGS@ diff --git a/src/address.cpp b/src/address.cpp index a38c9870..6dc86dd1 100644 --- a/src/address.cpp +++ b/src/address.cpp @@ -18,12 +18,10 @@ along with this program. If not, see . */ -#include "platform.hpp" #include "address.hpp" #include "err.hpp" #include "tcp_address.hpp" #include "ipc_address.hpp" -#include "tipc_address.hpp" #include #include @@ -52,14 +50,6 @@ zmq::address_t::~address_t () } } #endif -#if defined ZMQ_HAVE_LINUX - else if (protocol == "tipc") { - if (resolved.tipc_addr) { - delete resolved.tipc_addr; - resolved.tipc_addr = 0; - } - } -#endif } int zmq::address_t::to_string (std::string &addr_) const @@ -76,13 +66,6 @@ int zmq::address_t::to_string (std::string &addr_) const } } #endif -#if defined ZMQ_HAVE_LINUX - else if (protocol == "tipc") { - if (resolved.tipc_addr) { - return resolved.tipc_addr->to_string(addr_); - } - } -#endif if (!protocol.empty () && !address.empty ()) { std::stringstream s; diff --git a/src/address.hpp b/src/address.hpp index 2156ccf3..a8a4765d 100644 --- a/src/address.hpp +++ b/src/address.hpp @@ -28,9 +28,6 @@ namespace zmq class tcp_address_t; #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS class ipc_address_t; -#endif -#if defined ZMQ_HAVE_LINUX - class tipc_address_t; #endif struct address_t { address_t (const std::string &protocol_, const std::string &address_); @@ -45,9 +42,6 @@ namespace zmq tcp_address_t *tcp_addr; #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS ipc_address_t *ipc_addr; -#endif -#if defined ZMQ_HAVE_LINUX - tipc_address_t *tipc_addr; #endif } resolved; diff --git a/src/session_base.cpp b/src/session_base.cpp index dc85e6e6..b2676570 100644 --- a/src/session_base.cpp +++ b/src/session_base.cpp @@ -27,7 +27,6 @@ #include "err.hpp" #include "pipe.hpp" #include "likely.hpp" -#include "tipc_connecter.hpp" #include "tcp_connecter.hpp" #include "ipc_connecter.hpp" #include "pgm_sender.hpp" @@ -457,16 +456,6 @@ void zmq::session_base_t::start_connecting (bool wait_) } #endif -#if defined ZMQ_HAVE_LINUX - if (addr->protocol == "tipc") { - tipc_connecter_t *connecter = new (std::nothrow) tipc_connecter_t ( - io_thread, this, options, addr, wait_); - alloc_assert (connecter); - launch_child(connecter); - return; - } -#endif - #if defined ZMQ_HAVE_OPENPGM // Both PGM and EPGM transports are using the same infrastructure. diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 4fea5448..7fe40089 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -55,8 +55,6 @@ #include "address.hpp" #include "ipc_address.hpp" #include "tcp_address.hpp" -#include "tipc_address.hpp" -#include "tipc_listener.hpp" #ifdef ZMQ_HAVE_OPENPGM #include "pgm_socket.hpp" #endif @@ -184,7 +182,7 @@ 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_ != "tipc") { + protocol_ != "pgm" && protocol_ != "epgm") { errno = EPROTONOSUPPORT; return -1; } @@ -206,13 +204,7 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_) return -1; } #endif - // TIPC transport is only available on Linux. -#if !defined ZMQ_HAVE_LINUX - if (protocol_ = "tipc") { - errno = EPROTONOSUPPORT; - return -1; - } -#endif + // Check whether socket type and transport protocol match. // Specifically, multicast protocols can't be combined with // bi-directional messaging patterns (socket types). @@ -395,25 +387,6 @@ int zmq::socket_base_t::bind (const char *addr_) return 0; } #endif -#if defined ZMQ_HAVE_LINUX - if (protocol == "tipc") { - tipc_listener_t *listener = new (std::nothrow) tipc_listener_t ( - io_thread, this, options); - alloc_assert (listener); - int rc = listener->set_address (address.c_str ()); - if (rc != 0) { - delete listener; - event_bind_failed (address, zmq_errno()); - return -1; - } - - // Save last endpoint URI - listener->get_address (options.last_endpoint); - - add_endpoint (addr_, (own_t *) listener); - return 0; - } -#endif zmq_assert (false); return -1; @@ -544,18 +517,6 @@ int zmq::socket_base_t::connect (const char *addr_) } } #endif -#if defined ZMQ_HAVE_LINUX - else - if (protocol == "tipc") { - paddr->resolved.tipc_addr = new (std::nothrow) tipc_address_t (); - alloc_assert (paddr->resolved.tipc_addr); - int rc = paddr->resolved.tipc_addr->resolve (address.c_str()); - if (rc != 0) { - delete paddr; - return -1; - } - } -#endif #ifdef ZMQ_HAVE_OPENPGM if (protocol == "pgm" || protocol == "epgm") { struct pgm_addrinfo_t *res = NULL; diff --git a/src/tipc_address.cpp b/src/tipc_address.cpp deleted file mode 100644 index 555b1917..00000000 --- a/src/tipc_address.cpp +++ /dev/null @@ -1,105 +0,0 @@ -/* - Copyright (c) 2013 Ericsson AB - - 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 . -*/ - -#include "tipc_address.hpp" - -#if defined ZMQ_HAVE_LINUX - -#include "err.hpp" - -#include -#include - -zmq::tipc_address_t::tipc_address_t () -{ - memset (&address, 0, sizeof (address)); -} - -zmq::tipc_address_t::tipc_address_t (const sockaddr *sa, socklen_t sa_len) -{ - zmq_assert(sa && sa_len > 0); - - memset (&address, 0, sizeof (address)); - if (sa->sa_family == AF_TIPC) { - memcpy(&address, sa, sa_len); - } -} - -zmq::tipc_address_t::~tipc_address_t () -{ -} - -int zmq::tipc_address_t::resolve (const char *name) -{ - int res; - unsigned int type = 0; - unsigned int lower = 0; - unsigned int upper = 0; - - res = sscanf(name, "{%u,%u,%u}", &type, &lower, &upper); - if (res == 3) - goto nameseq; - else if (res == 2 && type > TIPC_RESERVED_TYPES) { - address.family = AF_TIPC; - address.addrtype = TIPC_ADDR_NAME; - address.addr.name.name.type = type; - address.addr.name.name.instance = lower; - address.scope = 0; - return 0; - } - else - return EINVAL; -nameseq: - if (type < TIPC_RESERVED_TYPES || upper < lower) - return EINVAL; - address.family = AF_TIPC; - address.addrtype = TIPC_ADDR_NAMESEQ; - address.addr.nameseq.type = type; - address.addr.nameseq.lower = lower; - address.addr.nameseq.upper = upper; - address.scope = TIPC_ZONE_SCOPE; - - return 0; -} - -int zmq::tipc_address_t::to_string (std::string &addr_) -{ - if (address.family != AF_TIPC) { - addr_.clear (); - return -1; - } - std::stringstream s; - s << "tipc://" << "{" << address.addr.nameseq.type; - s << ", " << address.addr.nameseq.lower; - s << ", " << address.addr.nameseq.upper << "}"; - addr_ = s.str (); - return 0; -} - -const sockaddr *zmq::tipc_address_t::addr () const -{ - return (sockaddr*) &address; -} - -socklen_t zmq::tipc_address_t::addrlen () const -{ - return (socklen_t) sizeof (address); -} - -#endif diff --git a/src/tipc_address.hpp b/src/tipc_address.hpp deleted file mode 100644 index d87dc43c..00000000 --- a/src/tipc_address.hpp +++ /dev/null @@ -1,67 +0,0 @@ -/* - Copyright (c) 2011 250bpm s.r.o. - Copyright (c) 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 . -*/ - -#ifndef __ZMQ_TIPC_ADDRESS_HPP_INCLUDED__ -#define __ZMQ_TIPC_ADDRESS_HPP_INCLUDED__ - -#include - -#include "platform.hpp" - -#if defined ZMQ_HAVE_LINUX - -#include -#include - -namespace zmq -{ - - class tipc_address_t - { - public: - - tipc_address_t (); - tipc_address_t (const sockaddr *sa, socklen_t sa_len); - ~tipc_address_t (); - - // This function sets up the address "{type, lower, upper}" for TIPC transport - int resolve (const char* name); - - // The opposite to resolve() - int to_string (std::string &addr_); - - const sockaddr *addr () const; - socklen_t addrlen () const; - - private: - - struct sockaddr_tipc address; - - tipc_address_t (const tipc_address_t&); - const tipc_address_t &operator = (const tipc_address_t&); - }; - -} - -#endif - -#endif - - diff --git a/src/tipc_connecter.cpp b/src/tipc_connecter.cpp deleted file mode 100644 index 7c927b2f..00000000 --- a/src/tipc_connecter.cpp +++ /dev/null @@ -1,257 +0,0 @@ -/* - Copyright (c) 2013 Ericsson AB - - 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 . -*/ - -#include "tipc_connecter.hpp" - -#if defined ZMQ_HAVE_LINUX - -#include -#include - -#include "stream_engine.hpp" -#include "io_thread.hpp" -#include "platform.hpp" -#include "random.hpp" -#include "err.hpp" -#include "ip.hpp" -#include "address.hpp" -#include "tipc_address.hpp" -#include "session_base.hpp" - -#include -#include -#include - -zmq::tipc_connecter_t::tipc_connecter_t (class io_thread_t *io_thread_, - class session_base_t *session_, const options_t &options_, - const address_t *addr_, bool delayed_start_) : - own_t (io_thread_, options_), - io_object_t (io_thread_), - addr (addr_), - s (retired_fd), - handle_valid (false), - delayed_start (delayed_start_), - timer_started (false), - session (session_), - current_reconnect_ivl(options.reconnect_ivl) -{ - zmq_assert (addr); - zmq_assert (addr->protocol == "tipc"); - addr->to_string (endpoint); - socket = session-> get_socket(); -} - -zmq::tipc_connecter_t::~tipc_connecter_t () -{ - zmq_assert (!timer_started); - zmq_assert (!handle_valid); - zmq_assert (s == retired_fd); -} - -void zmq::tipc_connecter_t::process_plug () -{ - if (delayed_start) - add_reconnect_timer (); - else - start_connecting (); -} - -void zmq::tipc_connecter_t::process_term (int linger_) -{ - if (timer_started) { - cancel_timer (reconnect_timer_id); - timer_started = false; - } - - if (handle_valid) { - rm_fd (handle); - handle_valid = false; - } - - if (s != retired_fd) - close (); - - own_t::process_term (linger_); -} - -void zmq::tipc_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::tipc_connecter_t::out_event () -{ - fd_t fd = connect (); - rm_fd (handle); - handle_valid = false; - - // Handle the error condition by attempt to reconnect. - if (fd == retired_fd) { - close (); - add_reconnect_timer(); - return; - } - // Create the engine object for this connection. - stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint); - alloc_assert (engine); - - // Attach the engine to the corresponding session object. - send_attach (session, engine); - - // Shut the connecter down. - terminate (); - - socket->event_connected (endpoint, fd); -} - -void zmq::tipc_connecter_t::timer_event (int id_) -{ - zmq_assert (id_ == reconnect_timer_id); - timer_started = false; - start_connecting (); -} - -void zmq::tipc_connecter_t::start_connecting () -{ - // Open the connecting socket. - int rc = open (); - - // Connect may succeed in synchronous manner. - if (rc == 0) { - handle = add_fd (s); - handle_valid = true; - out_event (); - } - - // Connection establishment may be delayed. Poll for its completion. - else - if (rc == -1 && errno == EINPROGRESS) { - handle = add_fd (s); - handle_valid = true; - set_pollout (handle); - socket->event_connect_delayed (endpoint, zmq_errno()); - } - - // Handle any other error condition by eventual reconnect. - else { - if (s != retired_fd) - close (); - add_reconnect_timer (); - } -} - -void zmq::tipc_connecter_t::add_reconnect_timer() -{ - int rc_ivl = get_new_reconnect_ivl(); - add_timer (rc_ivl, reconnect_timer_id); - socket->event_connect_retried (endpoint, rc_ivl); - timer_started = true; -} - -int zmq::tipc_connecter_t::get_new_reconnect_ivl () -{ - // The new interval is the current interval + random value. - int this_interval = current_reconnect_ivl + - (generate_random () % 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; -} - -int zmq::tipc_connecter_t::open () -{ - zmq_assert (s == retired_fd); - - // Create the socket. - s = open_socket (AF_TIPC, SOCK_STREAM, 0); - if (s == -1) - return -1; - - // Set the non-blocking flag. - unblock_socket (s); - // Connect to the remote peer. - int rc = ::connect ( - s, addr->resolved.tipc_addr->addr (), - addr->resolved.tipc_addr->addrlen ()); - - // Connect was successfull immediately. - if (rc == 0) - return 0; - - // Translate other error codes indicating asynchronous connect has been - // launched to a uniform EINPROGRESS. - if (rc == -1 && errno == EINTR) { - errno = EINPROGRESS; - return -1; - } - // Forward the error. - return -1; -} - -void zmq::tipc_connecter_t::close () -{ - zmq_assert (s != retired_fd); - int rc = ::close (s); - errno_assert (rc == 0); - socket->event_closed (endpoint, s); - s = retired_fd; -} - -zmq::fd_t zmq::tipc_connecter_t::connect () -{ - // Following code should handle both Berkeley-derived socket - // implementations and Solaris. - int err = 0; - socklen_t len = sizeof (err); - - int rc = getsockopt (s, SOL_SOCKET, SO_ERROR, (char*) &err, &len); - if (rc == -1) - err = errno; - if (err != 0) { - - // Assert if the error was caused by 0MQ bug. - // Networking problems are OK. No need to assert. - errno = err; - errno_assert (errno == ECONNREFUSED || errno == ECONNRESET || - errno == ETIMEDOUT || errno == EHOSTUNREACH || - errno == ENETUNREACH || errno == ENETDOWN); - - return retired_fd; - } - - fd_t result = s; - s = retired_fd; - return result; -} - -#endif - diff --git a/src/tipc_connecter.hpp b/src/tipc_connecter.hpp deleted file mode 100644 index 5e4a3121..00000000 --- a/src/tipc_connecter.hpp +++ /dev/null @@ -1,127 +0,0 @@ -/* - Copyright (c) 2013 Ericsson AB - - 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 . -*/ - -#ifndef __TIPC_CONNECTER_HPP_INCLUDED__ -#define __TIPC_CONNECTER_HPP_INCLUDED__ - -#include "platform.hpp" - -#if defined ZMQ_HAVE_LINUX - -#include "fd.hpp" -#include "own.hpp" -#include "stdint.hpp" -#include "io_object.hpp" - -namespace zmq -{ - - class io_thread_t; - class session_base_t; - struct address_t; - - class tipc_connecter_t : public own_t, public io_object_t - { - public: - - // If 'delayed_start' is true connecter first waits for a while, - // then starts connection process. - tipc_connecter_t (zmq::io_thread_t *io_thread_, - zmq::session_base_t *session_, const options_t &options_, - const address_t *addr_, bool delayed_start_); - ~tipc_connecter_t (); - - private: - - // ID of the timer used to delay the reconnection. - enum {reconnect_timer_id = 1}; - - // Handlers for incoming commands. - void process_plug (); - void process_term (int linger_); - - // Handlers for I/O events. - void in_event (); - void out_event (); - void timer_event (int id_); - - // Internal function to start the actual connection establishment. - void start_connecting (); - - // Internal function to add a reconnect timer - void add_reconnect_timer(); - - // Internal function to return a reconnect backoff delay. - // Will modify the current_reconnect_ivl used for next call - // Returns the currently used interval - int get_new_reconnect_ivl (); - - // Open IPC connecting socket. Returns -1 in case of error, - // 0 if connect was successfull immediately. Returns -1 with - // EAGAIN errno if async connect was launched. - int open (); - - // Close the connecting socket. - void close (); - - // Get the file descriptor of newly created connection. Returns - // retired_fd if the connection was unsuccessfull. - fd_t connect (); - - // Address to connect to. Owned by session_base_t. - const address_t *addr; - - // Underlying socket. - fd_t s; - - // Handle corresponding to the listening socket. - handle_t handle; - - // If true file descriptor is registered with the poller and 'handle' - // contains valid value. - bool handle_valid; - - // If true, connecter is waiting a while before trying to connect. - const bool delayed_start; - - // True iff a timer has been started. - bool timer_started; - - // Reference to the session we belong to. - zmq::session_base_t *session; - - // Current reconnect ivl, updated for backoff strategy - int current_reconnect_ivl; - - // String representation of endpoint to connect to - std::string endpoint; - - // Socket - zmq::socket_base_t *socket; - - tipc_connecter_t (const tipc_connecter_t&); - const tipc_connecter_t &operator = (const tipc_connecter_t&); - }; - -} - -#endif - -#endif - diff --git a/src/tipc_listener.cpp b/src/tipc_listener.cpp deleted file mode 100644 index 0b1a1b09..00000000 --- a/src/tipc_listener.cpp +++ /dev/null @@ -1,178 +0,0 @@ -/* - Copyright (c) 2011 250bpm s.r.o. - Copyright (c) 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 . -*/ - -#include "tipc_listener.hpp" - -#if defined ZMQ_HAVE_LINUX - -#include - -#include - -#include "stream_engine.hpp" -#include "tipc_address.hpp" -#include "io_thread.hpp" -#include "session_base.hpp" -#include "config.hpp" -#include "err.hpp" -#include "ip.hpp" -#include "socket_base.hpp" - -#include -#include -#include -#include - -zmq::tipc_listener_t::tipc_listener_t (io_thread_t *io_thread_, - socket_base_t *socket_, const options_t &options_) : - own_t (io_thread_, options_), - io_object_t (io_thread_), - s (retired_fd), - socket (socket_) -{ -} - -zmq::tipc_listener_t::~tipc_listener_t () -{ - zmq_assert (s == retired_fd); -} - -void zmq::tipc_listener_t::process_plug () -{ - // Start polling for incoming connections. - handle = add_fd (s); - set_pollin (handle); -} - -void zmq::tipc_listener_t::process_term (int linger_) -{ - rm_fd (handle); - close (); - own_t::process_term (linger_); -} - -void zmq::tipc_listener_t::in_event () -{ - fd_t fd = accept (); - - // If connection was reset by the peer in the meantime, just ignore it. - // TODO: Handle specific errors like ENFILE/EMFILE etc. - if (fd == retired_fd) { - socket->event_accept_failed (endpoint, zmq_errno()); - return; - } - - // Create the engine object for this connection. - stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint); - 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_base_t *session = session_base_t::create (io_thread, false, socket, - options, NULL); - errno_assert (session); - session->inc_seqnum (); - launch_child (session); - send_attach (session, engine, false); - socket->event_accepted (endpoint, fd); -} - -int zmq::tipc_listener_t::get_address (std::string &addr_) -{ - struct sockaddr_storage ss; - socklen_t sl = sizeof (ss); - - int rc = getsockname (s, (sockaddr *) &ss, &sl); - if (rc != 0) { - addr_.clear (); - return rc; - } - - tipc_address_t addr ((struct sockaddr *) &ss, sl); - return addr.to_string (addr_); -} - -int zmq::tipc_listener_t::set_address (const char *addr_) -{ - //convert str to address struct - int rc = address.resolve(addr_); - if (rc != 0) - return -1; - // Create a listening socket. - s = open_socket (AF_TIPC, SOCK_STREAM, 0); - if (s == -1) - return -1; - - address.to_string (endpoint); - - // Bind the socket to tipc name. - rc = bind (s, address.addr (), address.addrlen ()); - if (rc != 0) - goto error; - - // Listen for incomming connections. - rc = listen (s, options.backlog); - if (rc != 0) - goto error; - - socket->event_listening (endpoint, s); - return 0; - -error: - int err = errno; - close (); - errno = err; - return -1; -} - -void zmq::tipc_listener_t::close () -{ - zmq_assert (s != retired_fd); - int rc = ::close (s); - errno_assert (rc == 0); - s = retired_fd; - socket->event_closed (endpoint, s); -} - -zmq::fd_t zmq::tipc_listener_t::accept () -{ - // Accept one connection and deal with different failure modes. - // The situation where connection cannot be accepted due to insufficient - // resources is considered valid and treated by ignoring the connection. - struct sockaddr_storage ss = {}; - socklen_t ss_len = sizeof(ss); - - zmq_assert (s != retired_fd); - fd_t sock = ::accept (s, (struct sockaddr *) &ss, &ss_len); - if (sock == -1) { - errno_assert (errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS || - errno == EINTR || errno == ECONNABORTED || errno == EPROTO || errno == EMFILE || - errno == ENFILE); - return retired_fd; - } - /*FIXME Accept filters?*/ - return sock; -} - -#endif diff --git a/src/tipc_listener.hpp b/src/tipc_listener.hpp deleted file mode 100644 index d5d3e293..00000000 --- a/src/tipc_listener.hpp +++ /dev/null @@ -1,96 +0,0 @@ -/* - Copyright (c) 2013 Ericsson AB - - 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 . -*/ - -#ifndef __ZMQ_TIPC_LISTENER_HPP_INCLUDED__ -#define __ZMQ_TIPC_LISTENER_HPP_INCLUDED__ - -#include "platform.hpp" - -#if defined ZMQ_HAVE_LINUX - -#include - -#include "fd.hpp" -#include "own.hpp" -#include "stdint.hpp" -#include "io_object.hpp" -#include "tipc_address.hpp" - -namespace zmq -{ - - class io_thread_t; - class socket_base_t; - - class tipc_listener_t : public own_t, public io_object_t - { - public: - - tipc_listener_t (zmq::io_thread_t *io_thread_, - zmq::socket_base_t *socket_, const options_t &options_); - ~tipc_listener_t (); - - // Set address to listen on. - int set_address (const char *addr_); - - // Get the bound address for use with wildcards - int get_address (std::string &addr_); - - private: - - // Handlers for incoming commands. - void process_plug (); - void process_term (int linger_); - - // Handlers for I/O events. - void in_event (); - - // Close the listening socket. - void close (); - - // Accept the new connection. Returns the file descriptor of the - // newly created connection. The function may return retired_fd - // if the connection was dropped while waiting in the listen backlog. - fd_t accept (); - - // Address to listen on - tipc_address_t address; - - // Underlying socket. - fd_t s; - - // Handle corresponding to the listening socket. - handle_t handle; - - // Socket the listerner belongs to. - zmq::socket_base_t *socket; - - // String representation of endpoint to bind to - std::string endpoint; - - tipc_listener_t (const tipc_listener_t&); - const tipc_listener_t &operator = (const tipc_listener_t&); - }; - -} - -#endif - -#endif -