generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-14 02:57:47 +01:00
Code Issues Projects Releases Wiki Activity

PGM transport reconciled with subscription forwarding

Browse Source 
As PGM is not capable of passing subscriptions upstream,
subscriptions are ignored at sub side and engine subscribes
for all messages on pub side.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
This commit is contained in:
Martin Sustrik 2011-07-11 10:18:30 +02:00
parent d7adc3f19a
commit c7542981d1
3 changed files with 24 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/pgm_receiver.cpp
View File

@ -68,6 +68,9 @@ void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_, i_engine_sink *sink_)
set_pollin (socket_handle);
sink = sink_;
// If there are any subscriptions already queued in the session, drop them.
drop_subscriptions ();
}
void zmq::pgm_receiver_t::unplug ()
@ -101,7 +104,7 @@ void zmq::pgm_receiver_t::terminate ()
void zmq::pgm_receiver_t::activate_out ()
{
zmq_assert (false);
drop_subscriptions ();
}
void zmq::pgm_receiver_t::activate_in ()
@ -255,5 +258,12 @@ void zmq::pgm_receiver_t::timer_event (int token)
in_event ();
}
void zmq::pgm_receiver_t::drop_subscriptions ()
{
msg_t msg;
while (sink->read (&msg))
msg.close ();
}
#endif

4
src/pgm_receiver.hpp
View File

@ -64,6 +64,10 @@ namespace zmq
private:
// PGM is not able to move subscriptions upstream. Thus, drop all
// the pending subscriptions.
void drop_subscriptions ();
// RX timeout timer ID.
enum {rx_timer_id = 0xa1};

9
src/pgm_sender.cpp
View File

@ -88,6 +88,15 @@ void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, i_engine_sink *sink_)
// Set POLLOUT for downlink_socket_handle.
set_pollout (handle);
// PGM is not able to pass subscriptions upstream, thus we have no idea
// what messages are peers interested in. Because of that we have to
// subscribe for all the messages.
msg_t msg;
msg.init ();
bool ok = sink_->write (&msg);
zmq_assert (ok);
sink_->flush ();
}
void zmq::pgm_sender_t::unplug ()