mirror of
https://github.com/zeromq/libzmq.git
synced 2025-02-23 07:12:47 +01:00
Problem: duplication across ipc_listener_t, tcp_listener_t, tipc_listener_t
Solution: extract common base class stream_listener_base_t
This commit is contained in:
parent
a40a3b7a34
commit
93c1843f3e
@ -823,6 +823,8 @@ set(cxx-sources
|
||||
stream_engine.hpp
|
||||
stream_connecter_base.hpp
|
||||
stream_connecter_base.cpp
|
||||
stream_listener_base.hpp
|
||||
stream_listener_base.cpp
|
||||
sub.hpp
|
||||
tcp.hpp
|
||||
tcp_address.hpp
|
||||
|
@ -196,6 +196,8 @@ src_libzmq_la_SOURCES = \
|
||||
src/stream.hpp \
|
||||
src/stream_connecter_base.cpp \
|
||||
src/stream_connecter_base.hpp \
|
||||
src/stream_listener_base.cpp \
|
||||
src/stream_listener_base.hpp \
|
||||
src/stream_engine.cpp \
|
||||
src/stream_engine.hpp \
|
||||
src/sub.cpp \
|
||||
|
@ -132,34 +132,11 @@ int zmq::ipc_listener_t::create_wildcard_address (std::string &path_,
|
||||
zmq::ipc_listener_t::ipc_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_),
|
||||
_has_file (false),
|
||||
_s (retired_fd),
|
||||
_handle (static_cast<handle_t> (NULL)),
|
||||
_socket (socket_)
|
||||
stream_listener_base_t (io_thread_, socket_, options_),
|
||||
_has_file (false)
|
||||
{
|
||||
}
|
||||
|
||||
zmq::ipc_listener_t::~ipc_listener_t ()
|
||||
{
|
||||
zmq_assert (_s == retired_fd);
|
||||
}
|
||||
|
||||
void zmq::ipc_listener_t::process_plug ()
|
||||
{
|
||||
// Start polling for incoming connections.
|
||||
_handle = add_fd (_s);
|
||||
set_pollin (_handle);
|
||||
}
|
||||
|
||||
void zmq::ipc_listener_t::process_term (int linger_)
|
||||
{
|
||||
rm_fd (_handle);
|
||||
close ();
|
||||
own_t::process_term (linger_);
|
||||
}
|
||||
|
||||
void zmq::ipc_listener_t::in_event ()
|
||||
{
|
||||
fd_t fd = accept ();
|
||||
|
@ -36,22 +36,16 @@
|
||||
#include <string>
|
||||
|
||||
#include "fd.hpp"
|
||||
#include "own.hpp"
|
||||
#include "stdint.hpp"
|
||||
#include "io_object.hpp"
|
||||
#include "stream_listener_base.hpp"
|
||||
|
||||
namespace zmq
|
||||
{
|
||||
class io_thread_t;
|
||||
class socket_base_t;
|
||||
|
||||
class ipc_listener_t : public own_t, public io_object_t
|
||||
class ipc_listener_t : public stream_listener_base_t
|
||||
{
|
||||
public:
|
||||
ipc_listener_t (zmq::io_thread_t *io_thread_,
|
||||
zmq::socket_base_t *socket_,
|
||||
const options_t &options_);
|
||||
~ipc_listener_t ();
|
||||
|
||||
// Set address to listen on.
|
||||
int set_address (const char *addr_);
|
||||
@ -60,16 +54,9 @@ class ipc_listener_t : public own_t, public io_object_t
|
||||
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.
|
||||
int close ();
|
||||
|
||||
// Create wildcard path address
|
||||
static int create_wildcard_address (std::string &path_, std::string &file_);
|
||||
|
||||
@ -79,6 +66,8 @@ class ipc_listener_t : public own_t, public io_object_t
|
||||
bool filter (fd_t sock_);
|
||||
#endif
|
||||
|
||||
int 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.
|
||||
@ -94,18 +83,6 @@ class ipc_listener_t : public own_t, public io_object_t
|
||||
// Name of the file associated with the UNIX domain address.
|
||||
std::string _filename;
|
||||
|
||||
// Underlying socket.
|
||||
fd_t _s;
|
||||
|
||||
// Handle corresponding to the listening socket.
|
||||
handle_t _handle;
|
||||
|
||||
// Socket the listener belongs to.
|
||||
zmq::socket_base_t *_socket;
|
||||
|
||||
// String representation of endpoint to bind to
|
||||
std::string _endpoint;
|
||||
|
||||
// Acceptable temporary directory environment variables
|
||||
static const char *tmp_env_vars[];
|
||||
|
||||
|
94
src/stream_listener_base.cpp
Normal file
94
src/stream_listener_base.cpp
Normal file
@ -0,0 +1,94 @@
|
||||
/*
|
||||
Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
|
||||
|
||||
This file is part of libzmq, the ZeroMQ core engine in C++.
|
||||
|
||||
libzmq is free software; you can redistribute it and/or modify it under
|
||||
the terms of the GNU Lesser General Public License (LGPL) as published
|
||||
by the Free Software Foundation; either version 3 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
As a special exception, the Contributors give you permission to link
|
||||
this library with independent modules to produce an executable,
|
||||
regardless of the license terms of these independent modules, and to
|
||||
copy and distribute the resulting executable under terms of your choice,
|
||||
provided that you also meet, for each linked independent module, the
|
||||
terms and conditions of the license of that module. An independent
|
||||
module is a module which is not derived from or based on this library.
|
||||
If you modify this library, you must extend this exception to your
|
||||
version of the library.
|
||||
|
||||
libzmq 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 "precompiled.hpp"
|
||||
#include "stream_listener_base.hpp"
|
||||
#include "socket_base.hpp"
|
||||
|
||||
zmq::stream_listener_base_t::stream_listener_base_t (
|
||||
zmq::io_thread_t *io_thread_,
|
||||
zmq::socket_base_t *socket_,
|
||||
const zmq::options_t &options_) :
|
||||
own_t (io_thread_, options_),
|
||||
io_object_t (io_thread_),
|
||||
_s (retired_fd),
|
||||
_handle (static_cast<handle_t> (NULL)),
|
||||
_socket (socket_)
|
||||
{
|
||||
}
|
||||
|
||||
zmq::stream_listener_base_t::~stream_listener_base_t ()
|
||||
{
|
||||
zmq_assert (_s == retired_fd);
|
||||
zmq_assert (!_handle);
|
||||
}
|
||||
|
||||
zmq::zmq_socklen_t
|
||||
zmq::stream_listener_base_t::get_socket_address (sockaddr_storage *ss_) const
|
||||
{
|
||||
zmq_socklen_t sl = sizeof (*ss_);
|
||||
|
||||
const int rc =
|
||||
getsockname (_s, reinterpret_cast<struct sockaddr *> (ss_), &sl);
|
||||
|
||||
return rc != 0 ? 0 : sl;
|
||||
}
|
||||
|
||||
void zmq::stream_listener_base_t::process_plug ()
|
||||
{
|
||||
// Start polling for incoming connections.
|
||||
_handle = add_fd (_s);
|
||||
set_pollin (_handle);
|
||||
}
|
||||
|
||||
void zmq::stream_listener_base_t::process_term (int linger_)
|
||||
{
|
||||
rm_fd (_handle);
|
||||
_handle = static_cast<handle_t> (NULL);
|
||||
close ();
|
||||
own_t::process_term (linger_);
|
||||
}
|
||||
|
||||
int zmq::stream_listener_base_t::close ()
|
||||
{
|
||||
// TODO this is identical to stream_connector_base_t::close
|
||||
|
||||
zmq_assert (_s != retired_fd);
|
||||
#ifdef ZMQ_HAVE_WINDOWS
|
||||
const int rc = closesocket (_s);
|
||||
wsa_assert (rc != SOCKET_ERROR);
|
||||
#else
|
||||
const int rc = ::close (_s);
|
||||
errno_assert (rc == 0);
|
||||
#endif
|
||||
_socket->event_closed (_endpoint, _s);
|
||||
_s = retired_fd;
|
||||
|
||||
return 0;
|
||||
}
|
90
src/stream_listener_base.hpp
Normal file
90
src/stream_listener_base.hpp
Normal file
@ -0,0 +1,90 @@
|
||||
/*
|
||||
Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
|
||||
|
||||
This file is part of libzmq, the ZeroMQ core engine in C++.
|
||||
|
||||
libzmq is free software; you can redistribute it and/or modify it under
|
||||
the terms of the GNU Lesser General Public License (LGPL) as published
|
||||
by the Free Software Foundation; either version 3 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
As a special exception, the Contributors give you permission to link
|
||||
this library with independent modules to produce an executable,
|
||||
regardless of the license terms of these independent modules, and to
|
||||
copy and distribute the resulting executable under terms of your choice,
|
||||
provided that you also meet, for each linked independent module, the
|
||||
terms and conditions of the license of that module. An independent
|
||||
module is a module which is not derived from or based on this library.
|
||||
If you modify this library, you must extend this exception to your
|
||||
version of the library.
|
||||
|
||||
libzmq 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_STREAM_LISTENER_BASE_HPP_INCLUDED__
|
||||
#define __ZMQ_STREAM_LISTENER_BASE_HPP_INCLUDED__
|
||||
|
||||
#include <string>
|
||||
|
||||
#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;
|
||||
|
||||
#if defined(ZMQ_HAVE_HPUX) || defined(ZMQ_HAVE_VXWORKS)
|
||||
typedef int zmq_socklen_t;
|
||||
#else
|
||||
typedef socklen_t zmq_socklen_t;
|
||||
#endif
|
||||
|
||||
class stream_listener_base_t : public own_t, public io_object_t
|
||||
{
|
||||
public:
|
||||
stream_listener_base_t (zmq::io_thread_t *io_thread_,
|
||||
zmq::socket_base_t *socket_,
|
||||
const options_t &options_);
|
||||
~stream_listener_base_t ();
|
||||
|
||||
protected:
|
||||
zmq_socklen_t get_socket_address (sockaddr_storage *ss_) const;
|
||||
|
||||
private:
|
||||
// Handlers for incoming commands.
|
||||
void process_plug ();
|
||||
void process_term (int linger_);
|
||||
|
||||
protected:
|
||||
// Close the listening socket.
|
||||
virtual int close ();
|
||||
|
||||
// Underlying socket.
|
||||
fd_t _s;
|
||||
|
||||
// Handle corresponding to the listening socket.
|
||||
handle_t _handle;
|
||||
|
||||
// Socket the listener belongs to.
|
||||
zmq::socket_base_t *_socket;
|
||||
|
||||
// String representation of endpoint to bind to
|
||||
std::string _endpoint;
|
||||
|
||||
private:
|
||||
stream_listener_base_t (const stream_listener_base_t &);
|
||||
const stream_listener_base_t &operator= (const stream_listener_base_t &);
|
||||
};
|
||||
}
|
||||
|
||||
#endif
|
@ -63,35 +63,10 @@
|
||||
zmq::tcp_listener_t::tcp_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),
|
||||
_handle (static_cast<handle_t> (NULL)),
|
||||
_socket (socket_)
|
||||
stream_listener_base_t (io_thread_, socket_, options_)
|
||||
{
|
||||
}
|
||||
|
||||
zmq::tcp_listener_t::~tcp_listener_t ()
|
||||
{
|
||||
zmq_assert (_s == retired_fd);
|
||||
zmq_assert (!_handle);
|
||||
}
|
||||
|
||||
void zmq::tcp_listener_t::process_plug ()
|
||||
{
|
||||
// Start polling for incoming connections.
|
||||
_handle = add_fd (_s);
|
||||
set_pollin (_handle);
|
||||
}
|
||||
|
||||
void zmq::tcp_listener_t::process_term (int linger_)
|
||||
{
|
||||
rm_fd (_handle);
|
||||
_handle = static_cast<handle_t> (NULL);
|
||||
close ();
|
||||
own_t::process_term (linger_);
|
||||
}
|
||||
|
||||
void zmq::tcp_listener_t::in_event ()
|
||||
{
|
||||
fd_t fd = accept ();
|
||||
@ -134,34 +109,14 @@ void zmq::tcp_listener_t::in_event ()
|
||||
_socket->event_accepted (_endpoint, fd);
|
||||
}
|
||||
|
||||
void zmq::tcp_listener_t::close ()
|
||||
{
|
||||
zmq_assert (_s != retired_fd);
|
||||
#ifdef ZMQ_HAVE_WINDOWS
|
||||
int rc = closesocket (_s);
|
||||
wsa_assert (rc != SOCKET_ERROR);
|
||||
#else
|
||||
int rc = ::close (_s);
|
||||
errno_assert (rc == 0);
|
||||
#endif
|
||||
_socket->event_closed (_endpoint, _s);
|
||||
_s = retired_fd;
|
||||
}
|
||||
|
||||
int zmq::tcp_listener_t::get_address (std::string &addr_)
|
||||
{
|
||||
// Get the details of the TCP socket
|
||||
struct sockaddr_storage ss;
|
||||
#if defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_VXWORKS
|
||||
int sl = sizeof (ss);
|
||||
#else
|
||||
socklen_t sl = sizeof (ss);
|
||||
#endif
|
||||
int rc = getsockname (_s, reinterpret_cast<struct sockaddr *> (&ss), &sl);
|
||||
|
||||
if (rc != 0) {
|
||||
const zmq_socklen_t sl = get_socket_address (&ss);
|
||||
if (!sl) {
|
||||
addr_.clear ();
|
||||
return rc;
|
||||
return -1;
|
||||
}
|
||||
|
||||
tcp_address_t addr (reinterpret_cast<struct sockaddr *> (&ss), sl);
|
||||
|
@ -31,23 +31,17 @@
|
||||
#define __ZMQ_TCP_LISTENER_HPP_INCLUDED__
|
||||
|
||||
#include "fd.hpp"
|
||||
#include "own.hpp"
|
||||
#include "stdint.hpp"
|
||||
#include "io_object.hpp"
|
||||
#include "tcp_address.hpp"
|
||||
#include "stream_listener_base.hpp"
|
||||
|
||||
namespace zmq
|
||||
{
|
||||
class io_thread_t;
|
||||
class socket_base_t;
|
||||
|
||||
class tcp_listener_t : public own_t, public io_object_t
|
||||
class tcp_listener_t : public stream_listener_base_t
|
||||
{
|
||||
public:
|
||||
tcp_listener_t (zmq::io_thread_t *io_thread_,
|
||||
zmq::socket_base_t *socket_,
|
||||
const options_t &options_);
|
||||
~tcp_listener_t ();
|
||||
|
||||
// Set address to listen on.
|
||||
int set_address (const char *addr_);
|
||||
@ -56,16 +50,9 @@ class tcp_listener_t : public own_t, public io_object_t
|
||||
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
|
||||
@ -75,18 +62,6 @@ class tcp_listener_t : public own_t, public io_object_t
|
||||
// Address to listen on.
|
||||
tcp_address_t _address;
|
||||
|
||||
// Underlying socket.
|
||||
fd_t _s;
|
||||
|
||||
// Handle corresponding to the listening socket.
|
||||
handle_t _handle;
|
||||
|
||||
// Socket the listener belongs to.
|
||||
zmq::socket_base_t *_socket;
|
||||
|
||||
// String representation of endpoint to bind to
|
||||
std::string _endpoint;
|
||||
|
||||
tcp_listener_t (const tcp_listener_t &);
|
||||
const tcp_listener_t &operator= (const tcp_listener_t &);
|
||||
};
|
||||
|
@ -59,32 +59,10 @@
|
||||
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_)
|
||||
stream_listener_base_t (io_thread_, socket_, options_)
|
||||
{
|
||||
}
|
||||
|
||||
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 ();
|
||||
@ -119,16 +97,10 @@ void zmq::tipc_listener_t::in_event ()
|
||||
int zmq::tipc_listener_t::get_address (std::string &addr_)
|
||||
{
|
||||
struct sockaddr_storage ss;
|
||||
socklen_t sl = sizeof (ss);
|
||||
|
||||
#ifdef ZMQ_HAVE_VXWORKS
|
||||
int rc = getsockname (_s, (sockaddr *) &ss, (int *) &sl);
|
||||
#else
|
||||
int rc = getsockname (_s, (sockaddr *) &ss, &sl);
|
||||
#endif
|
||||
if (rc != 0) {
|
||||
const zmq_socklen_t sl = get_socket_address (&ss);
|
||||
if (!sl) {
|
||||
addr_.clear ();
|
||||
return rc;
|
||||
return -1;
|
||||
}
|
||||
|
||||
tipc_address_t addr ((struct sockaddr *) &ss, sl);
|
||||
@ -198,15 +170,6 @@ error:
|
||||
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.
|
||||
|
@ -37,23 +37,17 @@
|
||||
#include <string>
|
||||
|
||||
#include "fd.hpp"
|
||||
#include "own.hpp"
|
||||
#include "stdint.hpp"
|
||||
#include "io_object.hpp"
|
||||
#include "stream_listener_base.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
|
||||
class tipc_listener_t : public stream_listener_base_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_);
|
||||
@ -62,16 +56,9 @@ class tipc_listener_t : public own_t, public io_object_t
|
||||
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.
|
||||
@ -80,18 +67,6 @@ class tipc_listener_t : public own_t, public io_object_t
|
||||
// Address to listen on
|
||||
tipc_address_t _address;
|
||||
|
||||
// Underlying socket.
|
||||
fd_t _s;
|
||||
|
||||
// Handle corresponding to the listening socket.
|
||||
handle_t _handle;
|
||||
|
||||
// Socket the listener 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 &);
|
||||
};
|
||||
|
Loading…
x
Reference in New Issue
Block a user