mirror of
https://github.com/zeromq/libzmq.git
synced 2025-10-30 13:47:13 +01:00
Problem: duplication in *_event methods across subclasses of stream_connecter_base_t
Solution: pull up common code, introduce new create_engine function in base class
This commit is contained in:
@@ -62,14 +62,6 @@ zmq::ipc_connecter_t::ipc_connecter_t (class io_thread_t *io_thread_,
|
|||||||
zmq_assert (_addr->protocol == protocol_name::ipc);
|
zmq_assert (_addr->protocol == protocol_name::ipc);
|
||||||
}
|
}
|
||||||
|
|
||||||
void zmq::ipc_connecter_t::in_event ()
|
|
||||||
{
|
|
||||||
// We are not polling for incoming 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::ipc_connecter_t::out_event ()
|
void zmq::ipc_connecter_t::out_event ()
|
||||||
{
|
{
|
||||||
fd_t fd = connect ();
|
fd_t fd = connect ();
|
||||||
@@ -81,25 +73,8 @@ void zmq::ipc_connecter_t::out_event ()
|
|||||||
add_reconnect_timer ();
|
add_reconnect_timer ();
|
||||||
return;
|
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.
|
create_engine (fd);
|
||||||
send_attach (_session, engine);
|
|
||||||
|
|
||||||
// Shut the connecter down.
|
|
||||||
terminate ();
|
|
||||||
|
|
||||||
_socket->event_connected (_endpoint, fd);
|
|
||||||
}
|
|
||||||
|
|
||||||
void zmq::ipc_connecter_t::timer_event (int id_)
|
|
||||||
{
|
|
||||||
zmq_assert (id_ == reconnect_timer_id);
|
|
||||||
_reconnect_timer_started = false;
|
|
||||||
start_connecting ();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void zmq::ipc_connecter_t::start_connecting ()
|
void zmq::ipc_connecter_t::start_connecting ()
|
||||||
|
|||||||
@@ -57,9 +57,7 @@ class ipc_connecter_t : public stream_connecter_base_t
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Handlers for I/O events.
|
// Handlers for I/O events.
|
||||||
void in_event ();
|
|
||||||
void out_event ();
|
void out_event ();
|
||||||
void timer_event (int id_);
|
|
||||||
|
|
||||||
// Internal function to start the actual connection establishment.
|
// Internal function to start the actual connection establishment.
|
||||||
void start_connecting ();
|
void start_connecting ();
|
||||||
|
|||||||
@@ -134,3 +134,34 @@ void zmq::stream_connecter_base_t::close ()
|
|||||||
_socket->event_closed (_endpoint, _s);
|
_socket->event_closed (_endpoint, _s);
|
||||||
_s = retired_fd;
|
_s = retired_fd;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void zmq::stream_connecter_base_t::in_event ()
|
||||||
|
{
|
||||||
|
// We are not polling for incoming 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::stream_connecter_base_t::create_engine (fd_t fd)
|
||||||
|
{
|
||||||
|
// 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::stream_connecter_base_t::timer_event (int id_)
|
||||||
|
{
|
||||||
|
zmq_assert (id_ == reconnect_timer_id);
|
||||||
|
_reconnect_timer_started = false;
|
||||||
|
start_connecting ();
|
||||||
|
}
|
||||||
|
|||||||
@@ -73,6 +73,13 @@ class stream_connecter_base_t : public own_t, public io_object_t
|
|||||||
void process_plug ();
|
void process_plug ();
|
||||||
void process_term (int linger_);
|
void process_term (int linger_);
|
||||||
|
|
||||||
|
// Handlers for I/O events.
|
||||||
|
void in_event ();
|
||||||
|
void timer_event (int id_);
|
||||||
|
|
||||||
|
// Internal function to create the engine after connection was established.
|
||||||
|
void create_engine (fd_t fd);
|
||||||
|
|
||||||
// Internal function to add a reconnect timer
|
// Internal function to add a reconnect timer
|
||||||
void add_reconnect_timer ();
|
void add_reconnect_timer ();
|
||||||
|
|
||||||
|
|||||||
@@ -90,14 +90,6 @@ void zmq::tcp_connecter_t::process_term (int linger_)
|
|||||||
stream_connecter_base_t::process_term (linger_);
|
stream_connecter_base_t::process_term (linger_);
|
||||||
}
|
}
|
||||||
|
|
||||||
void zmq::tcp_connecter_t::in_event ()
|
|
||||||
{
|
|
||||||
// We are not polling for incoming 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::tcp_connecter_t::out_event ()
|
void zmq::tcp_connecter_t::out_event ()
|
||||||
{
|
{
|
||||||
if (_connect_timer_started) {
|
if (_connect_timer_started) {
|
||||||
@@ -105,6 +97,9 @@ void zmq::tcp_connecter_t::out_event ()
|
|||||||
_connect_timer_started = false;
|
_connect_timer_started = false;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TODO this is still very similar to (t)ipc_connecter_t, maybe the
|
||||||
|
// differences can be factored out
|
||||||
|
|
||||||
rm_handle ();
|
rm_handle ();
|
||||||
|
|
||||||
const fd_t fd = connect ();
|
const fd_t fd = connect ();
|
||||||
@@ -116,18 +111,7 @@ void zmq::tcp_connecter_t::out_event ()
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Create the engine object for this connection.
|
create_engine (fd);
|
||||||
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::tcp_connecter_t::timer_event (int id_)
|
void zmq::tcp_connecter_t::timer_event (int id_)
|
||||||
@@ -138,10 +122,8 @@ void zmq::tcp_connecter_t::timer_event (int id_)
|
|||||||
rm_handle ();
|
rm_handle ();
|
||||||
close ();
|
close ();
|
||||||
add_reconnect_timer ();
|
add_reconnect_timer ();
|
||||||
} else if (id_ == reconnect_timer_id) {
|
} else
|
||||||
_reconnect_timer_started = false;
|
stream_connecter_base_t::timer_event (id_);
|
||||||
start_connecting ();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void zmq::tcp_connecter_t::start_connecting ()
|
void zmq::tcp_connecter_t::start_connecting ()
|
||||||
|
|||||||
@@ -59,7 +59,6 @@ class tcp_connecter_t : public stream_connecter_base_t
|
|||||||
void process_term (int linger_);
|
void process_term (int linger_);
|
||||||
|
|
||||||
// Handlers for I/O events.
|
// Handlers for I/O events.
|
||||||
void in_event ();
|
|
||||||
void out_event ();
|
void out_event ();
|
||||||
void timer_event (int id_);
|
void timer_event (int id_);
|
||||||
|
|
||||||
|
|||||||
@@ -64,14 +64,6 @@ zmq::tipc_connecter_t::tipc_connecter_t (class io_thread_t *io_thread_,
|
|||||||
zmq_assert (_addr->protocol == "tipc");
|
zmq_assert (_addr->protocol == "tipc");
|
||||||
}
|
}
|
||||||
|
|
||||||
void zmq::tipc_connecter_t::in_event ()
|
|
||||||
{
|
|
||||||
// We are not polling for incoming 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 ()
|
void zmq::tipc_connecter_t::out_event ()
|
||||||
{
|
{
|
||||||
fd_t fd = connect ();
|
fd_t fd = connect ();
|
||||||
@@ -83,25 +75,8 @@ void zmq::tipc_connecter_t::out_event ()
|
|||||||
add_reconnect_timer ();
|
add_reconnect_timer ();
|
||||||
return;
|
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.
|
create_engine (fd);
|
||||||
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);
|
|
||||||
_reconnect_timer_started = false;
|
|
||||||
start_connecting ();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void zmq::tipc_connecter_t::start_connecting ()
|
void zmq::tipc_connecter_t::start_connecting ()
|
||||||
|
|||||||
@@ -58,9 +58,7 @@ class tipc_connecter_t : public stream_connecter_base_t
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Handlers for I/O events.
|
// Handlers for I/O events.
|
||||||
void in_event ();
|
|
||||||
void out_event ();
|
void out_event ();
|
||||||
void timer_event (int id_);
|
|
||||||
|
|
||||||
// Internal function to start the actual connection establishment.
|
// Internal function to start the actual connection establishment.
|
||||||
void start_connecting ();
|
void start_connecting ();
|
||||||
|
|||||||
Reference in New Issue
Block a user