generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-10-16 18:56:55 +02:00
Code Issues Projects Releases Wiki Activity

Problem: xpub_t.xattach_pipe no handle msg.copy or pipe_t.write fails.

Browse Source
This commit is contained in:
evoskuil
2017-03-30 14:45:40 -07:00
parent 6436bc51e6
commit 1dd9cac24d
1 changed files with 33 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
src/xpub.cpp
View File

@@ -41,18 +41,18 @@ zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
verbose_unsubs (false), verbose_unsubs (false),
more (false), more (false),
lossy (true), lossy (true),
manual(false), manual (false),
pending_pipes (), pending_pipes (),
welcome_msg () welcome_msg ()
{ {
last_pipe = NULL; last_pipe = NULL;
options.type = ZMQ_XPUB; options.type = ZMQ_XPUB;
welcome_msg.init(); welcome_msg.init ();
} }
zmq::xpub_t::~xpub_t () zmq::xpub_t::~xpub_t ()
{ {
welcome_msg.close(); welcome_msg.close ();
} }
void zmq::xpub_t::xattach_pipe (pipe_t *pipe_, bool subscribe_to_all_) void zmq::xpub_t::xattach_pipe (pipe_t *pipe_, bool subscribe_to_all_)
@@ -65,15 +65,16 @@ void zmq::xpub_t::xattach_pipe (pipe_t *pipe_, bool subscribe_to_all_)
if (subscribe_to_all_) if (subscribe_to_all_)
subscriptions.add (NULL, 0, pipe_); subscriptions.add (NULL, 0, pipe_);
// if welcome message exist // if welcome message exists, send a copy of it
if (welcome_msg.size() > 0) if (welcome_msg.size () > 0)
{ {
msg_t copy; msg_t copy;
copy.init(); copy.init ();
copy.copy(welcome_msg); int rc = copy.copy (welcome_msg);
errno_assert (rc == 0);
pipe_->write(&copy); bool ok = pipe_->write (&copy);
pipe_->flush(); zmq_assert (ok);
pipe_->flush ();
} }
// The pipe is active when attached. Let's read the subscriptions from // The pipe is active when attached. Let's read the subscriptions from
@@ -95,35 +96,35 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
{ {
// Store manual subscription to use on termination // Store manual subscription to use on termination
if (*data == 0) if (*data == 0)
manual_subscriptions.rm(data + 1, size - 1, pipe_); manual_subscriptions.rm (data + 1, size - 1, pipe_);
else else
manual_subscriptions.add(data + 1, size - 1, pipe_); manual_subscriptions.add (data + 1, size - 1, pipe_);
pending_pipes.push_back(pipe_); pending_pipes.push_back (pipe_);
pending_data.push_back(blob_t(data, size)); pending_data.push_back (blob_t (data, size));
if (metadata) if (metadata)
metadata->add_ref(); metadata->add_ref ();
pending_metadata.push_back(metadata); pending_metadata.push_back (metadata);
pending_flags.push_back(0); pending_flags.push_back (0);
} }
else else
{ {
bool unique; bool unique;
if (*data == 0) if (*data == 0)
unique = subscriptions.rm(data + 1, size - 1, pipe_); unique = subscriptions.rm (data + 1, size - 1, pipe_);
else else
unique = subscriptions.add(data + 1, size - 1, pipe_); unique = subscriptions.add (data + 1, size - 1, pipe_);
// If the (un)subscription is not a duplicate store it so that it can be // If the (un)subscription is not a duplicate store it so that it can be
// passed to the user on next recv call unless verbose mode is enabled // passed to the user on next recv call unless verbose mode is enabled
// which makes to pass always these messages. // which makes to pass always these messages.
if (options.type == ZMQ_XPUB && (unique || (*data == 1 && verbose_subs) || if (options.type == ZMQ_XPUB && (unique || (*data == 1 && verbose_subs) ||
(*data == 0 && verbose_unsubs && verbose_subs))) { (*data == 0 && verbose_unsubs && verbose_subs))) {
pending_data.push_back(blob_t(data, size)); pending_data.push_back (blob_t(data, size));
if (metadata) if (metadata)
metadata->add_ref(); metadata->add_ref ();
pending_metadata.push_back(metadata); pending_metadata.push_back (metadata);
pending_flags.push_back(0); pending_flags.push_back (0);
} }
} }
} }
@@ -131,7 +132,7 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
// Process user message coming upstream from xsub socket // Process user message coming upstream from xsub socket
pending_data.push_back (blob_t (data, size)); pending_data.push_back (blob_t (data, size));
if (metadata) if (metadata)
metadata->add_ref(); metadata->add_ref ();
pending_metadata.push_back (metadata); pending_metadata.push_back (metadata);
pending_flags.push_back (sub.flags ()); pending_flags.push_back (sub.flags ());
} }
@@ -151,7 +152,7 @@ int zmq::xpub_t::xsetsockopt (int option_, const void *optval_,
|| option_ == ZMQ_XPUB_VERBOSER || option_ == ZMQ_XPUB_VERBOSER
|| option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_NODROP
|| option_ == ZMQ_XPUB_MANUAL) { || option_ == ZMQ_XPUB_MANUAL) {
if (optvallen_ != sizeof(int) || *static_cast <const int*> (optval_) < 0) { if (optvallen_ != sizeof (int) || *static_cast <const int*> (optval_) < 0) {
errno = EINVAL; errno = EINVAL;
return -1; return -1;
} }
@@ -174,26 +175,26 @@ int zmq::xpub_t::xsetsockopt (int option_, const void *optval_,
else else
if (option_ == ZMQ_SUBSCRIBE && manual) { if (option_ == ZMQ_SUBSCRIBE && manual) {
if (last_pipe != NULL) if (last_pipe != NULL)
subscriptions.add ((unsigned char *)optval_, optvallen_, last_pipe); subscriptions.add ((unsigned char *) optval_, optvallen_, last_pipe);
} }
else else
if (option_ == ZMQ_UNSUBSCRIBE && manual) { if (option_ == ZMQ_UNSUBSCRIBE && manual) {
if (last_pipe != NULL) if (last_pipe != NULL)
subscriptions.rm ((unsigned char *)optval_, optvallen_, last_pipe); subscriptions.rm ((unsigned char *) optval_, optvallen_, last_pipe);
} }
else else
if (option_ == ZMQ_XPUB_WELCOME_MSG) { if (option_ == ZMQ_XPUB_WELCOME_MSG) {
welcome_msg.close(); welcome_msg.close ();
if (optvallen_ > 0) { if (optvallen_ > 0) {
int rc = welcome_msg.init_size(optvallen_); int rc = welcome_msg.init_size (optvallen_);
errno_assert(rc == 0); errno_assert(rc == 0);
unsigned char *data = (unsigned char*)welcome_msg.data(); unsigned char *data = (unsigned char*) welcome_msg.data ();
memcpy(data, optval_, optvallen_); memcpy (data, optval_, optvallen_);
} }
else else
welcome_msg.init(); welcome_msg.init ();
} }
else { else {
errno = EINVAL; errno = EINVAL;