generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-10-27 19:10:22 +01:00
Code Issues Projects Releases Wiki Activity

Problem: can no longer send user data from XSUB to XPUB

Browse Source 
Solution: fix regression introduced by:
https://github.com/zeromq/libzmq/pull/3168

Correctly fall back to user message if the first byte is neither 0 nor
1, and add a simple unit test

Fixes https://github.com/zeromq/libzmq/issues/3656
This commit is contained in:
Luca Boccassi
2019-09-02 11:31:26 +01:00
parent c737e5720f
commit 6fb8ef0711
2 changed files with 27 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/xpub.cpp
View File

@@ -104,13 +104,10 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
data = static_cast<unsigned char *> (sub.command_body ());
size = sub.command_body_size ();
subscribe = sub.is_subscribe ();
} else if (sub.size () > 0) {
unsigned char first = *msg_data;
if (first == 0 || first == 1) {
data = msg_data + 1;
size = sub.size () - 1;
subscribe = first == 1;
}
} else if (sub.size () > 0 && (*msg_data == 0 || *msg_data == 1)) {
data = msg_data + 1;
size = sub.size () - 1;
subscribe = *msg_data == 1;
} else {
// Process user message coming upstream from xsub socket
_pending_data.push_back (blob_t (msg_data, sub.size ()));

23
tests/test_xpub_manual.cpp
View File

@@ -434,6 +434,28 @@ void test_unsubscribe_cleanup ()
test_context_socket_close (sub);
}
void test_user_message ()
{
// Create a publisher
void *pub = test_context_socket (ZMQ_XPUB);
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, "inproc://soname"));
// Create a subscriber
void *sub = test_context_socket (ZMQ_XSUB);
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, "inproc://soname"));
// Send some data that is neither sub nor unsub
const char subscription[] = {2, 'A', 0};
send_string_expect_success (sub, subscription, 0);
// Receive subscriptions from subscriber
recv_string_expect_success (pub, subscription, 0);
// Clean up.
test_context_socket_close (pub);
test_context_socket_close (sub);
}
int main ()
{
setup_test_environment ();
@@ -444,6 +466,7 @@ int main ()
RUN_TEST (test_xpub_proxy_unsubscribe_on_disconnect);
RUN_TEST (test_missing_subscriptions);
RUN_TEST (test_unsubscribe_cleanup);
RUN_TEST (test_user_message);
return UNITY_END ();
}