Merge pull request #175 from jachiang/master

zmq socket subscription filter methods added.

Author: evoskuil

include/bitcoin/protocol/zmq/socket.hpp

@@ -115,11 +115,11 @@ class BCP_API socket
     /// Configure the socket to connect through the specified socks5 proxy.
     bool set_socks_proxy(const config::authority& socks_proxy);
 
-    /////// Configure subscriber socket to apply the message filter.
-    ////bool set_subscription(const data_chunk& filter);
+    /// Configure subscriber socket to apply the message filter.
+    bool set_subscription(const data_chunk& filter);
 
-    /////// Configure subscriber socket to remove the message filter.
-    ////bool set_unsubscription(const data_chunk& filter);
+    /// Configure subscriber socket to remove the message filter.
+    bool set_unsubscription(const data_chunk& filter);
 
     /// Send a message on this socket.
     code send(message& packet);
@@ -133,6 +133,7 @@ class BCP_API socket
     bool set32(int32_t option, int32_t value);
     bool set64(int32_t option, int64_t value);
     bool set(int32_t option, const std::string& value);
+    bool set(int32_t option, const data_chunk& value);
 
 private:
     void* self_;

src/zmq/socket.cpp

@@ -33,7 +33,7 @@
 namespace libbitcoin {
 namespace protocol {
 namespace zmq {
-    
+
 static const auto subscribe_all = "";
 static constexpr int32_t zmq_true = 1;
 static constexpr int32_t zmq_false = 0;
@@ -42,7 +42,7 @@ static constexpr int32_t reconnect_interval = 100;
 static const bc::protocol::settings default_settings;
 
 // Linger
-// The default value of -1 specifies an infinite linger period. Pending 
+// The default value of -1 specifies an infinite linger period. Pending
 // messages shall not be discarded after a call to zmq_close(); attempting to
 // terminate the socket's context with zmq_term() shall block until all pending
 // messages have been sent to a peer. The value 0 specifies no linger period.
@@ -241,6 +241,13 @@ bool socket::set(int32_t option, const std::string& value)
     return zmq_setsockopt(self_, option, buffer, value.size()) != zmq_fail;
 }
 
+// private
+bool socket::set(int32_t option, const data_chunk& value)
+{
+    return zmq_setsockopt(self_, option, value.data(), value.size())
+        != zmq_fail;
+}
+
 // For NULL security, ZAP calls are only made for non-empty domain.
 // For PLAIN/CURVE, calls are always made if ZAP handler is present.
 bool socket::set_authentication_domain(const std::string& domain)
@@ -288,6 +295,16 @@ bool socket::set_socks_proxy(const config::authority& socks_proxy)
     return socks_proxy && set(ZMQ_SOCKS_PROXY, socks_proxy.to_string());
 }
 
+bool socket::set_subscription(const data_chunk& filter)
+{
+    return set(ZMQ_SUBSCRIBE, filter);
+}
+
+bool socket::set_unsubscription(const data_chunk& filter)
+{
+    return set(ZMQ_UNSUBSCRIBE, filter);
+}
+
 code socket::send(message& packet)
 {
     return packet.send(*this);

test/utility.hpp

@@ -26,6 +26,7 @@
 
 #define TEST_DOMAIN "testing"
 #define TEST_MESSAGE "hello world!"
+#define TEST_TOPIC "hello"
 #define TEST_HOST "127.0.0.1"
 #define TEST_HOST_BAD "127.0.0.42"
 #define TEST_PUBLIC_ENDPOINT "tcp://" TEST_HOST ":9000"
@@ -75,4 +76,4 @@ class simple_thread
     }
 };
 
-#endif
+#endif

test/zmq/socket.cpp

@@ -406,6 +406,71 @@ BOOST_AUTO_TEST_CASE(socket__pub_sub__grasslands_asynchronous_connect_first__rec
     SEND_MESSAGES_UNTIL(publisher, received);
 }
 
+// Default receive-all filter is removed. No message received.
+BOOST_AUTO_TEST_CASE(socket__pub_sub__grasslands_asynchronous__no_subscription)
+{
+    zmq::context context;
+    BOOST_REQUIRE(context);
+
+    std::promise<bool> received;
+
+    simple_thread publisher_thread([&]()
+    {
+        zmq::socket publisher(context, role::publisher);
+        BOOST_REQUIRE(publisher);
+        BC_REQUIRE_SUCCESS(publisher.bind({ TEST_PUBLIC_ENDPOINT }));
+        SEND_MESSAGES_UNTIL(publisher, received); //begin sending now, needs own thread.
+    });
+
+    simple_thread subscriber_thread([&]()
+    {
+        zmq::socket subscriber(context, role::subscriber);
+        BOOST_REQUIRE(subscriber);
+
+        data_chunk subscribe_all;
+        BOOST_REQUIRE(subscriber.set_unsubscription(subscribe_all));
+
+        BC_REQUIRE_SUCCESS(subscriber.connect({ TEST_PUBLIC_ENDPOINT }));
+
+        RECEIVE_FAILURE(subscriber);
+        received.set_value(true);
+    });
+}
+
+// Default receive-all filter is removed. "hello" filter is added.
+BOOST_AUTO_TEST_CASE(socket__pub_sub__grasslands_asynchronous__hello_subscription_only)
+{
+    zmq::context context;
+    BOOST_REQUIRE(context);
+
+    std::promise<bool> received;
+
+    simple_thread publisher_thread([&]()
+    {
+        zmq::socket publisher(context, role::publisher);
+        BOOST_REQUIRE(publisher);
+        BC_REQUIRE_SUCCESS(publisher.bind({ TEST_PUBLIC_ENDPOINT }));
+        SEND_MESSAGES_UNTIL(publisher, received);
+    });
+
+    simple_thread subscriber_thread([&]()
+    {
+        zmq::socket subscriber(context, role::subscriber);
+        BOOST_REQUIRE(subscriber);
+
+        data_chunk subscribe_all;
+        BOOST_REQUIRE(subscriber.set_unsubscription(subscribe_all));
+
+        std::string topic = TEST_TOPIC;
+        BOOST_REQUIRE(subscriber.set_subscription(to_chunk(topic)));
+
+        BC_REQUIRE_SUCCESS(subscriber.connect({ TEST_PUBLIC_ENDPOINT }));
+
+        RECEIVE_MESSAGE(subscriber);
+        received.set_value(true);
+    });
+}
+
 BOOST_AUTO_TEST_CASE(socket__xpub_xsub__grasslands_two_threads__subscribed)
 {
     zmq::context context;