generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-09-26 15:59:34 +02:00
Code Issues Projects Releases Wiki Activity

Fair queueing of subscriptions added to XPUB socket

Browse Source 
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
This commit is contained in:
Martin Sustrik 2011-01-14 12:38:07 +01:00
parent 58c9830d90
commit a348d94c09
2 changed files with 22 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/xpub.cpp
View File

@ -25,10 +25,11 @@
zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_) :
socket_base_t (parent_, tid_),
dist (this)
dist (this),
fq (this)
{
options.type = ZMQ_XPUB;
options.requires_in = false;
options.requires_in = true;
options.requires_out = true;
}
@ -39,14 +40,16 @@ zmq::xpub_t::~xpub_t ()
void zmq::xpub_t::xattach_pipes (class reader_t *inpipe_,
class writer_t *outpipe_, const blob_t &peer_identity_)
{
zmq_assert (!inpipe_);
zmq_assert (inpipe_ && outpipe_);
dist.attach (outpipe_);
fq.attach (inpipe_);
}
void zmq::xpub_t::process_term (int linger_)
{
// Terminate the outbound pipes.
dist.terminate ();
fq.terminate ();
// Continue with the termination immediately.
socket_base_t::process_term (linger_);
@ -62,3 +65,13 @@ bool zmq::xpub_t::xhas_out ()
return dist.has_out ();
}
int zmq::xpub_t::xrecv (zmq_msg_t *msg_, int flags_)
{
return fq.recv (msg_, flags_);
}
bool zmq::xpub_t::xhas_in ()
{
return fq.has_in ();
}

6
src/xpub.hpp
View File

@ -24,6 +24,7 @@
#include "array.hpp"
#include "pipe.hpp"
#include "dist.hpp"
#include "fq.hpp"
namespace zmq
{
@ -40,6 +41,8 @@ namespace zmq
const blob_t &peer_identity_);
int xsend (zmq_msg_t *msg_, int flags_);
bool xhas_out ();
int xrecv (zmq_msg_t *msg_, int flags_);
bool xhas_in ();
private:
@ -49,6 +52,9 @@ namespace zmq
// Distributor of messages holding the list of outbound pipes.
dist_t dist;
// Fair queuer for inbound subscriptions.
fq_t fq;
xpub_t (const xpub_t&);
const xpub_t &operator = (const xpub_t&);
};