CR HBPVIS#219 IV

tests/subscriber.cpp

@@ -27,7 +27,7 @@ BOOST_AUTO_TEST_CASE(construction)
 
 BOOST_AUTO_TEST_CASE(invalid_construction)
 {
-    BOOST_CHECK_THROW(zeroeq::Subscriber(zeroeq::NULL_SESSION),
+    BOOST_CHECK_THROW(zeroeq::Subscriber{zeroeq::NULL_SESSION},
                       std::runtime_error);
     BOOST_CHECK_THROW(zeroeq::Subscriber(""), std::runtime_error);
     BOOST_CHECK_THROW(zeroeq::Subscriber(zeroeq::URI("localhost")),
@@ -37,8 +37,7 @@ BOOST_AUTO_TEST_CASE(invalid_construction)
                       std::runtime_error);
 
     zeroeq::Subscriber shared;
-    BOOST_CHECK_THROW(zeroeq::Subscriber subscriber(zeroeq::NULL_SESSION,
-                                                    shared),
+    BOOST_CHECK_THROW(zeroeq::Subscriber(zeroeq::NULL_SESSION, shared),
                       std::runtime_error);
     BOOST_CHECK_THROW(zeroeq::Subscriber("", shared), std::runtime_error);
     BOOST_CHECK_THROW(zeroeq::Subscriber(zeroeq::URI("localhost"), shared),

zeroeq/CMakeLists.txt

@@ -18,11 +18,11 @@ set(ZEROEQ_PUBLIC_HEADERS
   uri.h)
 
 set(ZEROEQ_HEADERS
-  detail/browser.h
   detail/common.h
   detail/constants.h
   detail/context.h
   detail/port.h
+  detail/receiver.h
   detail/sender.h
   detail/socket.h)
 

zeroeq/client.cpp

@@ -5,38 +5,36 @@
 
 #include "client.h"
 
-#include "detail/browser.h"
 #include "detail/common.h"
+#include "detail/receiver.h"
 
 #include <servus/servus.h>
 #include <unordered_map>
 
 namespace zeroeq
 {
-class Client::Impl : public detail::Browser
+class Client::Impl : public detail::Receiver
 {
 public:
     explicit Impl(const std::string& session)
-        : Browser(SERVER_SERVICE,
-                  session == DEFAULT_SESSION ? getDefaultRepSession() : session)
+        : detail::Receiver(SERVER_SERVICE, session == DEFAULT_SESSION
+                                       ? getDefaultRepSession()
+                                       : session)
         , _servers(zmq_socket(getContext(), ZMQ_DEALER),
                    [](void* s) { ::zmq_close(s); })
     {
     }
 
     explicit Impl(const URIs& uris)
-        : Browser(SERVER_SERVICE)
+        : detail::Receiver(SERVER_SERVICE)
         , _servers(zmq_socket(getContext(), ZMQ_DEALER),
                    [](void* s) { ::zmq_close(s); })
     {
         for (const auto& uri : uris)
         {
-            if (uri.getScheme() == DEFAULT_SCHEMA &&
-                (uri.getHost().empty() || uri.getPort() == 0))
-            {
+            if (!uri.isFullyQualified())
                 ZEROEQTHROW(std::runtime_error(
                     std::string("Non-fully qualified URI used for server")));
-            }
 
             const auto& zmqURI = buildZmqURI(uri);
             if (!addConnection(zmqURI))

zeroeq/detail/browser.h → zeroeq/detail/receiver.h

@@ -19,10 +19,10 @@ namespace zeroeq
 namespace detail
 {
 /** Manages and updates a set of connections with a zeroconf browser. */
-class Browser
+class Receiver
 {
 public:
-    Browser(const std::string& service, const std::string session)
+    Receiver(const std::string& service, const std::string session)
         : _servus(service)
         , _session(session)
         , _context(detail::getContext())
@@ -39,14 +39,14 @@ class Browser
         update();
     }
 
-    Browser(const std::string& service)
+    Receiver(const std::string& service)
         : _servus(service)
         , _session(zeroeq::NULL_SESSION)
         , _context(detail::getContext())
     {
     }
 
-    virtual ~Browser()
+    virtual ~Receiver()
     {
         if (_servus.isBrowsing())
             _servus.endBrowsing();
@@ -77,7 +77,7 @@ class Browser
                 const uint128_t identifier(_servus.get(instance, KEY_INSTANCE));
                 zmq::SocketPtr socket = createSocket(identifier);
                 if (socket)
-                    _addConnection(zmqURI, socket);
+                    _connect(zmqURI, socket);
             }
         }
     }
@@ -86,16 +86,10 @@ class Browser
     {
         zmq::SocketPtr socket = createSocket(uint128_t());
         if (socket)
-            return _addConnection(zmqURI, socket);
+            return _connect(zmqURI, socket);
         return true;
     }
 
-    /**
-     * Create the socket for zmqURI, return nullptr if connection is to be
-     * ignored.
-     */
-    virtual zmq::SocketPtr createSocket(const uint128_t& instance) = 0;
-
     void addSockets(std::vector<detail::Socket>& entries)
     {
         entries.insert(entries.end(), _entries.begin(), _entries.end());
@@ -105,9 +99,16 @@ class Browser
     using SocketMap = std::map<std::string, zmq::SocketPtr>;
 
     void* getContext() { return _context.get(); }
+
+    /**
+     * Create the socket for the given instance, return nullptr if connection is
+     * to be ignored.
+     */
+    virtual zmq::SocketPtr createSocket(const uint128_t& instance) = 0;
+
     const SocketMap& getSockets() { return _sockets; }
 
-    bool _addConnection(const std::string& zmqURI, zmq::SocketPtr socket)
+    bool _connect(const std::string& zmqURI, zmq::SocketPtr socket)
     {
         if (zmq_connect(socket.get(), zmqURI.c_str()) == -1)
         {

zeroeq/detail/sender.cpp

@@ -25,6 +25,11 @@ namespace zeroeq
 {
 namespace detail
 {
+Sender::Sender(const URI& uri_, const int type)
+    : Sender(uri_, type, {}, {})
+{
+}
+
 Sender::Sender(const URI& uri_, const int type, const std::string service,
                const std::string& session)
     : _context(getContext())
@@ -37,11 +42,6 @@ Sender::Sender(const URI& uri_, const int type, const std::string service,
     zmq_setsockopt(socket.get(), ZMQ_SNDHWM, &hwm, sizeof(hwm));
 }
 
-Sender::Sender(const URI& uri_, const int type)
-    : Sender(uri_, type, {}, {})
-{
-}
-
 Sender::~Sender()
 {
     socket.reset();
@@ -62,23 +62,23 @@ void Sender::initURI()
         host.clear();
 
     uint16_t port = uri.getPort();
-    if (host.empty() || port == 0)
-    {
-        std::string hostStr, portStr;
-        _getEndPoint(hostStr, portStr);
-
-        if (port == 0)
-        {
-            // No overflow is possible unless ZMQ reports bad port number.
-            port = std::stoi(portStr);
-            uri.setPort(port);
-        }
+    if (!host.empty() && port != 0)
+        return;
 
-        if (host.empty())
-            uri.setHost(hostStr);
+    std::string hostStr, portStr;
+    _getEndPoint(hostStr, portStr);
 
-        ZEROEQINFO << "Bound to " << uri << std::endl;
+    if (port == 0)
+    {
+        // No overflow is possible unless ZMQ reports bad port number.
+        port = std::stoi(portStr);
+        uri.setPort(port);
     }
+
+    if (host.empty())
+        uri.setHost(hostStr);
+
+    ZEROEQINFO << "Bound to " << uri << std::endl;
 }
 
 void Sender::announce()

zeroeq/detail/sender.h

@@ -20,9 +20,9 @@ class Sender
     zmq::ContextPtr _context; // must be private before socket
 
 public:
+    Sender(const URI& uri_, const int type);
     Sender(const URI& uri_, const int type, const std::string service,
            const std::string& session);
-    Sender(const URI& uri_, const int type);
     ~Sender();
 
     std::string getAddress() const;

zeroeq/receiver.cpp

@@ -103,10 +103,9 @@ class Receiver::Impl
                 intervals.push_back(sockets.size() - before);
             }
 
-            const uint32_t remaining =
-                duration_cast<milliseconds>(high_resolution_clock::now() -
-                                            startTime)
-                    .count();
+            const auto remaining = duration_cast<milliseconds>(
+                                       high_resolution_clock::now() - startTime)
+                                       .count();
 
             switch (zmq_poll(sockets.data(), int(sockets.size()), remaining))
             {
@@ -152,10 +151,9 @@ class Receiver::Impl
                 }
             }
             }
-        } while (haveData &&
-                 duration_cast<milliseconds>(high_resolution_clock::now() -
-                                             startTime)
-                         .count() < timeout);
+        } while (haveData && duration_cast<milliseconds>(
+                                 high_resolution_clock::now() - startTime)
+                                     .count() < timeout);
         return hadData;
     }
 };

zeroeq/receiver.h

@@ -87,6 +87,7 @@ class Receiver
      * their list of sockets.
      */
     virtual void update() {}
+
     /**
      * Add the given connection to the list of receiving sockets.
      *

zeroeq/server.cpp

@@ -5,9 +5,11 @@
 
 #include "server.h"
 
-#include "detail/browser.h"
+#include "detail/receiver.h"
 #include "detail/sender.h"
 
+#include <zmq.h>
+
 #include <cassert>
 #include <unordered_map>
 

zeroeq/subscriber.cpp

@@ -6,10 +6,10 @@
 
 #include "subscriber.h"
 
-#include "detail/browser.h"
 #include "detail/byteswap.h"
 #include "detail/common.h"
 #include "detail/constants.h"
+#include "detail/receiver.h"
 #include "detail/sender.h"
 #include "detail/socket.h"
 #include "log.h"
@@ -24,26 +24,23 @@
 
 namespace zeroeq
 {
-class Subscriber::Impl : public detail::Browser
+class Subscriber::Impl : public detail::Receiver
 {
 public:
     Impl(const std::string& session)
-        : Browser(PUBLISHER_SERVICE,
+        : detail::Receiver(PUBLISHER_SERVICE,
                   session == DEFAULT_SESSION ? getDefaultPubSession() : session)
         , _selfInstance(detail::Sender::getUUID())
     {
     }
 
     Impl(const URI& uri)
-        : Browser(PUBLISHER_SERVICE)
+        : detail::Receiver(PUBLISHER_SERVICE)
         , _selfInstance(detail::Sender::getUUID())
     {
-        if (uri.getScheme() == DEFAULT_SCHEMA &&
-            (uri.getHost().empty() || uri.getPort() == 0))
-        {
+        if (!uri.isFullyQualified())
             ZEROEQTHROW(std::runtime_error(
                 std::string("Non-fully qualified URI used for subscriber")));
-        }
 
         const std::string& zmqURI = buildZmqURI(uri);
         if (!addConnection(zmqURI))

zeroeq/uri.cpp

@@ -96,4 +96,10 @@ bool URI::operator!=(const servus::URI& rhs) const
 {
     return servus::URI::operator!=(rhs);
 }
+
+bool URI::isFullyQualified() const
+{
+    return getScheme() != DEFAULT_SCHEMA ||
+           (!getHost().empty() && getPort() != 0);
+}
 }

zeroeq/uri.h

@@ -50,6 +50,10 @@ class URI : private servus::URI
 
     /** Convert this URI to a servus::URI */
     const servus::URI& toServusURI() const { return *this; }
+
+    /** @return true if the host and port are given for a tcp URI. */
+    ZEROEQ_API bool isFullyQualified() const;
+
     /** @name servus::URI API */
     //@{
     using servus::URI::getScheme;