Discovery.hh

Go to the documentation of this file.

/*
 * Copyright (C) 2014 Open Source Robotics Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
*/

#ifndef IGNITION_TRANSPORT_DISCOVERY_HH_
#define IGNITION_TRANSPORT_DISCOVERY_HH_

#ifdef _WIN32
  // For socket(), connect(), send(), and recv().
  #include <Winsock2.h>
  #include <Ws2def.h>
  #include <Ws2ipdef.h>
  #include <Ws2tcpip.h>
  // Type used for raw data on this platform.
  using raw_type = char;
#else
  // For data types
  #include <sys/types.h>
  // For socket(), connect(), send(), and recv()
  #include <sys/socket.h>
  // For gethostbyname()
  #include <netdb.h>
  // For inet_addr()
  #include <arpa/inet.h>
  // For close()
  #include <unistd.h>
  // For sockaddr_in
  #include <netinet/in.h>
  // Type used for raw data on this platform
  using raw_type = void;
#endif

#ifdef _WIN32
  #pragma warning(push, 0)
#endif
#include <zmq.hpp>
#ifdef _WIN32
  #pragma warning(pop)
  // Suppress "decorated name length exceed" warning in STL.
  #pragma warning(disable: 4503)
  // Suppress "depreted API warnings" in WINSOCK.
  #pragma warning(disable: 4996)
#endif

#include <algorithm>
#include <condition_variable>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include "ignition/transport/Helpers.hh"
#include "ignition/transport/NetUtils.hh"
#include "ignition/transport/Packet.hh"
#include "ignition/transport/Publisher.hh"
#include "ignition/transport/TopicStorage.hh"
#include "ignition/transport/TransportTypes.hh"

namespace ignition
{
  namespace transport
  {
    template<typename Pub>
    class IGNITION_TRANSPORT_VISIBLE Discovery
    {
      public: Discovery(const std::string &_pUuid,
                        const int _port,
                        const bool _verbose = false)
        : port(_port),
          hostAddr(determineHost()),
          pUuid(_pUuid),
          silenceInterval(kDefSilenceInterval),
          activityInterval(kDefActivityInterval),
          heartbeatInterval(kDefHeartbeatInterval),
          connectionCb(nullptr),
          disconnectionCb(nullptr),
          verbose(_verbose),
          initialized(false),
          numHeartbeatsUninitialized(0),
          exit(false),
          enabled(false)
      {
        std::string ignIp;
        if (env("IGN_IP", ignIp) && !ignIp.empty())
          this->hostInterfaces = {ignIp};
        else
        {
          // Get the list of network interfaces in this host.
          this->hostInterfaces = determineInterfaces();
        }

#ifdef _WIN32
        WORD wVersionRequested;
        WSADATA wsaData;

        // Request WinSock v2.2.
        wVersionRequested = MAKEWORD(2, 2);
        // Load WinSock DLL.
        if (WSAStartup(wVersionRequested, &wsaData) != 0)
        {
          std::cerr << "Unable to load WinSock DLL" << std::endl;
          return;
        }
#endif
        for (const auto &netIface : this->hostInterfaces)
        {
          auto succeed = this->RegisterNetIface(netIface);

          // If the IP address that we're selecting as the main IP address of
          // the host is invalid, we change it to 127.0.0.1 .
          // This is probably because IGN_IP is set to a wrong value.
          if (netIface == this->hostAddr && !succeed)
          {
            this->RegisterNetIface("127.0.0.1");
            std::cerr << "Did you set the environment variable IGN_IP with a "
                      << "correct IP address? " << std::endl
                      << "  [" << netIface << "] seems an invalid local IP "
                      << "address." << std::endl
                      << "  Using 127.0.0.1 as hostname." << std::endl;
            this->hostAddr = "127.0.0.1";
          }
        }

        // Socket option: SO_REUSEADDR. This options is used only for receiving
        // data. We can reuse the same socket for receiving multicast data from
        // multiple interfaces. We will use the socket at position 0 for
        // receiving data.
        int reuseAddr = 1;
        if (setsockopt(this->sockets.at(0), SOL_SOCKET, SO_REUSEADDR,
            reinterpret_cast<const char *>(&reuseAddr), sizeof(reuseAddr)) != 0)
        {
          std::cerr << "Error setting socket option (SO_REUSEADDR)."
                    << std::endl;
          return;
        }

#ifdef SO_REUSEPORT
        // Socket option: SO_REUSEPORT. This options is used only for receiving
        // data. We can reuse the same socket for receiving multicast data from
        // multiple interfaces. We will use the socket at position 0 for
        // receiving data.
        int reusePort = 1;
        if (setsockopt(this->sockets.at(0), SOL_SOCKET, SO_REUSEPORT,
            reinterpret_cast<const char *>(&reusePort), sizeof(reusePort)) != 0)
        {
          std::cerr << "Error setting socket option (SO_REUSEPORT)."
                    << std::endl;
          return;
        }
#endif
        // Bind the first socket to the discovery port.
        sockaddr_in localAddr;
        memset(&localAddr, 0, sizeof(localAddr));
        localAddr.sin_family = AF_INET;
        localAddr.sin_addr.s_addr = htonl(INADDR_ANY);
        localAddr.sin_port = htons(static_cast<u_short>(this->port));

        if (bind(this->sockets.at(0),
          reinterpret_cast<sockaddr *>(&localAddr), sizeof(sockaddr_in)) < 0)
        {
          std::cerr << "Binding to a local port failed." << std::endl;
          return;
        }

        // Set 'mcastAddr' to the multicast discovery group.
        memset(&this->mcastAddr, 0, sizeof(this->mcastAddr));
        this->mcastAddr.sin_family = AF_INET;
        this->mcastAddr.sin_addr.s_addr =
          inet_addr(this->kMulticastGroup.c_str());
        this->mcastAddr.sin_port = htons(static_cast<u_short>(this->port));

        if (this->verbose)
          this->PrintCurrentState();
      }

      public: virtual ~Discovery()
      {
        // Tell the service thread to terminate.
        this->exitMutex.lock();
        this->exit = true;
        this->exitMutex.unlock();

        // Don't join on Windows, because it can hang when this object
        // is destructed on process exit (e.g., when it's a global static).
        // I think that it's due to this bug:
        // https://connect.microsoft.com/VisualStudio/feedback/details/747145/std-thread-join-hangs-if-called-after-main-exits-when-using-vs2012-rc
#ifndef _WIN32
        // Wait for the service threads to finish before exit.
        if (this->threadReception.joinable())
          this->threadReception.join();
#else
        bool exitLoop = false;
        while (!exitLoop)
        {
          std::lock_guard<std::mutex> lock(this->exitMutex);
          {
            if (this->threadReceptionExiting)
            {
              exitLoop = true;
            }
          }
          std::this_thread::sleep_for(std::chrono::milliseconds(50));
        }
#endif
        // Broadcast a BYE message to trigger the remote cancellation of
        // all our advertised topics.
        this->SendMsg(ByeType,
          Publisher("", "", this->pUuid, "", AdvertiseOptions()));

        // Close sockets.
        for (const auto &sock : this->sockets)
        {
#ifdef _WIN32
          closesocket(sock);
          WSACleanup();
#else
          close(sock);
#endif
        }
      }

      public: void Start()
      {
        {
          std::lock_guard<std::mutex> lock(this->mutex);

          // The service is already running.
          if (this->enabled)
            return;

          this->enabled = true;
        }

        auto now = std::chrono::steady_clock::now();
        this->timeNextHeartbeat = now;
        this->timeNextActivity = now;

        // Start the thread that receives discovery information.
        this->threadReception = std::thread(&Discovery::RecvMessages, this);

#ifdef _WIN32
        this->threadReceptionExiting = false;
        this->threadReception.detach();
#endif
      }

      public: bool Advertise(const Pub &_publisher)
      {
        {
          std::lock_guard<std::mutex> lock(this->mutex);

          if (!this->enabled)
            return false;

          // Add the addressing information (local publisher).
          if (!this->info.AddPublisher(_publisher))
            return false;
        }

        // Only advertise a message outside this process if the scope
        // is not 'Process'
        if (_publisher.Options().Scope() != Scope_t::PROCESS)
          this->SendMsg(AdvType, _publisher);

        return true;
      }

      public: bool Discover(const std::string &_topic) const
      {
        DiscoveryCallback<Pub> cb;
        bool found;
        Addresses_M<Pub> addresses;

        {
          std::lock_guard<std::mutex> lock(this->mutex);

          if (!this->enabled)
            return false;

          cb = this->connectionCb;
        }

        Pub pub;
        pub.SetTopic(_topic);
        pub.SetPUuid(this->pUuid);

        // Send a discovery request.
        this->SendMsg(SubType, pub);

        {
          std::lock_guard<std::mutex> lock(this->mutex);
          found = this->info.Publishers(_topic, addresses);
        }

        if (found)
        {
          // I already have information about this topic.
          for (const auto &proc : addresses)
          {
            for (const auto &node : proc.second)
            {
              if (cb)
              {
                // Execute the user's callback for a service request. Notice
                // that we only execute one callback for preventing receive
                //  multiple service responses for a single request.
                cb(node);
              }
            }
          }
        }

        return true;
      }

      public: const TopicStorage<Pub> &Info() const
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        return this->info;
      }

      public: bool Publishers(const std::string &_topic,
                              Addresses_M<Pub> &_publishers) const
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        return this->info.Publishers(_topic, _publishers);
      }

      public: bool Unadvertise(const std::string &_topic,
                               const std::string &_nUuid)
      {
        Pub inf;
        {
          std::lock_guard<std::mutex> lock(this->mutex);

          if (!this->enabled)
            return false;

          // Don't do anything if the topic is not advertised by any of my nodes
          if (!this->info.Publisher(_topic, this->pUuid, _nUuid, inf))
            return true;

          // Remove the topic information.
          this->info.DelPublisherByNode(_topic, this->pUuid, _nUuid);
        }

        // Only unadvertise a message outside this process if the scope
        // is not 'Process'.
        if (inf.Options().Scope() != Scope_t::PROCESS)
          this->SendMsg(UnadvType, inf);

        return true;
      }

      public: std::string HostAddr() const
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        return this->hostAddr;
      }

      public: unsigned int ActivityInterval() const
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        return this->activityInterval;
      }

      public: unsigned int HeartbeatInterval() const
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        return this->heartbeatInterval;
      }

      public: unsigned int SilenceInterval() const
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        return this->silenceInterval;
      }

      public: void SetActivityInterval(const unsigned int _ms)
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        this->activityInterval = _ms;
      }

      public: void SetHeartbeatInterval(const unsigned int _ms)
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        this->heartbeatInterval = _ms;
      }

      public: void SetSilenceInterval(const unsigned int _ms)
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        this->silenceInterval = _ms;
      }

      public: void ConnectionsCb(const DiscoveryCallback<Pub> &_cb)
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        this->connectionCb = _cb;
      }

      public: void DisconnectionsCb(const DiscoveryCallback<Pub> &_cb)
      {
        std::lock_guard<std::mutex> lock(this->mutex);
        this->disconnectionCb = _cb;
      }

      public: void PrintCurrentState() const
      {
        std::lock_guard<std::mutex> lock(this->mutex);

        std::cout << "---------------" << std::endl;
        std::cout << std::boolalpha << "Enabled: "
                  << this->enabled << std::endl;
        std::cout << "Discovery state" << std::endl;
        std::cout << "\tUUID: " << this->pUuid << std::endl;
        std::cout << "Settings" << std::endl;
        std::cout << "\tActivity: " << this->activityInterval
                  << " ms." << std::endl;
        std::cout << "\tHeartbeat: " << this->heartbeatInterval
                  << "ms." << std::endl;
        std::cout << "\tSilence: " << this->silenceInterval
                  << " ms." << std::endl;
        std::cout << "Known information: " << std::endl;
        this->info.Print();

        // Used to calculate the elapsed time.
        Timestamp now = std::chrono::steady_clock::now();

        std::cout << "Activity" << std::endl;
        if (this->activity.empty())
          std::cout << "\t<empty>" << std::endl;
        else
        {
          for (auto &proc : this->activity)
          {
            // Elapsed time since the last update from this publisher.
            std::chrono::duration<double> elapsed = now - proc.second;

            std::cout << "\t" << proc.first << std::endl;
            std::cout << "\t\t" << "Since: " << std::chrono::duration_cast<
              std::chrono::milliseconds>(elapsed).count() << " ms. ago. "
              << std::endl;
          }
        }
        std::cout << "---------------" << std::endl;
      }

      public: void TopicList(std::vector<std::string> &_topics) const
      {
        this->WaitForInit();
        std::lock_guard<std::mutex> lock(this->mutex);
        this->info.TopicList(_topics);
      }

      public: void WaitForInit() const
      {
        std::unique_lock<std::mutex> lk(this->mutex);

        if (!this->initialized)
        {
          this->initializedCv.wait(lk, [this]{return this->initialized;});
        }
      }

      private: void UpdateActivity()
      {
        Timestamp now = std::chrono::steady_clock::now();

        std::lock_guard<std::mutex> lock(this->mutex);

        if (now < this->timeNextActivity)
          return;

        for (auto it = this->activity.cbegin(); it != this->activity.cend();)
        {
          // Elapsed time since the last update from this publisher.
          auto elapsed = now - it->second;

          // This publisher has expired.
          if (std::chrono::duration_cast<std::chrono::milliseconds>
               (elapsed).count() > this->silenceInterval)
          {
            // Remove all the info entries for this process UUID.
            this->info.DelPublishersByProc(it->first);

            // Notify without topic information. This is useful to inform the
            // client that a remote node is gone, even if we were not
            // interested in its topics.
            Pub publisher;
            publisher.SetPUuid(it->first);
            this->disconnectionCb(publisher);

            // Remove the activity entry.
            this->activity.erase(it++);
          }
          else
            ++it;
        }

        this->timeNextActivity = std::chrono::steady_clock::now() +
          std::chrono::milliseconds(this->activityInterval);
      }

      private: void UpdateHeartbeat()
      {
        Timestamp now = std::chrono::steady_clock::now();

        {
          std::lock_guard<std::mutex> lock(this->mutex);

          if (now < this->timeNextHeartbeat)
            return;
        }

        Publisher pub("", "", this->pUuid, "", AdvertiseOptions());
        this->SendMsg(HeartbeatType, pub);

        std::map<std::string, std::vector<Pub>> nodes;
        {
          std::lock_guard<std::mutex> lock(this->mutex);

          // Re-advertise topics that are advertised inside this process.
          this->info.PublishersByProc(this->pUuid, nodes);
        }

        for (const auto &topic : nodes)
        {
          for (const auto &node : topic.second)
            this->SendMsg(AdvType, node);
        }

        {
          std::lock_guard<std::mutex> lock(this->mutex);
          if (!this->initialized)
          {
            ++this->numHeartbeatsUninitialized;
            if (this->numHeartbeatsUninitialized == 2)
            {
              // We consider the discovery initialized after two cycles of
              // heartbeats sent.
              this->initialized = true;

              // Notify anyone waiting for the initialization phase to finish.
              this->initializedCv.notify_all();
            }
          }

          this->timeNextHeartbeat = std::chrono::steady_clock::now() +
            std::chrono::milliseconds(this->heartbeatInterval);
        }
      }

      private: int NextTimeout() const
      {
        auto now = std::chrono::steady_clock::now();
        auto timeUntilNextHeartbeat = this->timeNextHeartbeat - now;
        auto timeUntilNextActivity = this->timeNextActivity - now;

        int t = static_cast<int>(
          std::chrono::duration_cast<std::chrono::milliseconds>
            (std::min(timeUntilNextHeartbeat, timeUntilNextActivity)).count());
        int t2 = std::min(t, this->kTimeout);
        return std::max(t2, 0);
      }

      private: void RecvMessages()
      {
        bool timeToExit = false;
        while (!timeToExit)
        {
          // Poll socket for a reply, with timeout.
          zmq::pollitem_t items[] =
          {
            {0, this->sockets.at(0), ZMQ_POLLIN, 0},
          };

          // Calculate the timeout.
          int timeout = this->NextTimeout();

          try
          {
            zmq::poll(&items[0], sizeof(items) / sizeof(items[0]), timeout);
          }
          catch(...)
          {
            continue;
          }

          //  If we got a reply, process it.
          if (items[0].revents & ZMQ_POLLIN)
          {
            this->RecvDiscoveryUpdate();

            if (this->verbose)
              this->PrintCurrentState();
          }

          this->UpdateHeartbeat();
          this->UpdateActivity();

          // Is it time to exit?
          {
            std::lock_guard<std::mutex> lock(this->exitMutex);
            if (this->exit)
              timeToExit = true;
          }
        }
#ifdef _WIN32
        std::lock_guard<std::mutex> lock(this->exitMutex);
        this->threadReceptionExiting = true;
#endif
      }

      private: void RecvDiscoveryUpdate()
      {
        char rcvStr[Discovery::kMaxRcvStr];
        std::string srcAddr;
        uint16_t srcPort;
        sockaddr_in clntAddr;
        socklen_t addrLen = sizeof(clntAddr);

        if ((recvfrom(this->sockets.at(0),
              reinterpret_cast<raw_type *>(rcvStr),
              this->kMaxRcvStr, 0,
              reinterpret_cast<sockaddr *>(&clntAddr),
              reinterpret_cast<socklen_t *>(&addrLen))) < 0)
        {
          std::cerr << "Discovery::RecvDiscoveryUpdate() recvfrom error"
                    << std::endl;
          return;
        }
        srcAddr = inet_ntoa(clntAddr.sin_addr);
        srcPort = ntohs(clntAddr.sin_port);

        if (this->verbose)
        {
          std::cout << "\nReceived discovery update from " << srcAddr << ": "
                    << srcPort << std::endl;
        }

        this->DispatchDiscoveryMsg(srcAddr, rcvStr);
      }


      private: void DispatchDiscoveryMsg(const std::string &_fromIp,
                                         char *_msg)
      {
        Header header;
        char *pBody = _msg;

        // Create the header from the raw bytes.
        header.Unpack(_msg);
        pBody += header.HeaderLength();

        // Discard the message if the wire protocol is different than mine.
        if (this->kWireVersion != header.Version())
          return;

        auto recvPUuid = header.PUuid();

        // Discard our own discovery messages.
        if (recvPUuid == this->pUuid)
          return;

        // Update timestamp and cache the callbacks.
        DiscoveryCallback<Pub> connectCb;
        DiscoveryCallback<Pub> disconnectCb;
        {
          std::lock_guard<std::mutex> lock(this->mutex);
          this->activity[recvPUuid] = std::chrono::steady_clock::now();
          connectCb = this->connectionCb;
          disconnectCb = this->disconnectionCb;
        }

        switch (header.Type())
        {
          case AdvType:
          {
            // Read the rest of the fields.
            transport::AdvertiseMessage<Pub> advMsg;
            advMsg.Unpack(pBody);

            // Check scope of the topic.
            if ((advMsg.Publisher().Options().Scope() == Scope_t::PROCESS) ||
                (advMsg.Publisher().Options().Scope() == Scope_t::HOST &&
                 _fromIp != this->hostAddr))
            {
              return;
            }

            // Register an advertised address for the topic.
            bool added;
            {
              std::lock_guard<std::mutex> lock(this->mutex);
              added = this->info.AddPublisher(advMsg.Publisher());
            }

            if (added && connectCb)
            {
              // Execute the client's callback.
              connectCb(advMsg.Publisher());
            }

            break;
          }
          case SubType:
          {
            // Read the rest of the fields.
            SubscriptionMsg subMsg;
            subMsg.Unpack(pBody);
            auto recvTopic = subMsg.Topic();

            // Check if at least one of my nodes advertises the topic requested.
            Addresses_M<Pub> addresses;
            {
              std::lock_guard<std::mutex> lock(this->mutex);
              if (!this->info.HasAnyPublishers(recvTopic, this->pUuid))
              {
                break;
              }

              if (!this->info.Publishers(recvTopic, addresses))
                break;
            }

            for (const auto &nodeInfo : addresses[this->pUuid])
            {
              // Check scope of the topic.
              if ((nodeInfo.Options().Scope() == Scope_t::PROCESS) ||
                  (nodeInfo.Options().Scope() == Scope_t::HOST &&
                   _fromIp != this->hostAddr))
              {
                continue;
              }

              // Answer an ADVERTISE message.
              this->SendMsg(AdvType, nodeInfo);
            }

            break;
          }
          case HeartbeatType:
          {
            // The timestamp has already been updated.
            break;
          }
          case ByeType:
          {
            // Remove the activity entry for this publisher.
            {
              std::lock_guard<std::mutex> lock(this->mutex);
              this->activity.erase(recvPUuid);
            }

            if (disconnectCb)
            {
              Pub pub;
              pub.SetPUuid(recvPUuid);
              // Notify the new disconnection.
              disconnectCb(pub);
            }

            // Remove the address entry for this topic.
            {
              std::lock_guard<std::mutex> lock(this->mutex);
              this->info.DelPublishersByProc(recvPUuid);
            }

            break;
          }
          case UnadvType:
          {
            // Read the address.
            transport::AdvertiseMessage<Pub> advMsg;
            advMsg.Unpack(pBody);

            // Check scope of the topic.
            if ((advMsg.Publisher().Options().Scope() == Scope_t::PROCESS) ||
                (advMsg.Publisher().Options().Scope() == Scope_t::HOST &&
                 _fromIp != this->hostAddr))
            {
              return;
            }

            if (disconnectCb)
            {
              // Notify the new disconnection.
              disconnectCb(advMsg.Publisher());
            }

            // Remove the address entry for this topic.
            {
              std::lock_guard<std::mutex> lock(this->mutex);
              this->info.DelPublisherByNode(advMsg.Publisher().Topic(),
                advMsg.Publisher().PUuid(), advMsg.Publisher().NUuid());
            }

            break;
          }
          default:
          {
            std::cerr << "Unknown message type [" << header.Type() << "]\n";
            break;
          }
        }
      }

      private: template<typename T>
      void SendMsg(const uint8_t _type,
                   const T &_pub,
                   const uint16_t _flags = 0) const
      {
        // Create the header.
        Header header(this->Version(), _pub.PUuid(), _type, _flags);
        auto msgLength = 0;
        std::vector<char> buffer;

        std::string topic = _pub.Topic();

        switch (_type)
        {
          case AdvType:
          case UnadvType:
          {
            // Create the [UN]ADVERTISE message.
            transport::AdvertiseMessage<T> advMsg(header, _pub);

            // Allocate a buffer and serialize the message.
            buffer.resize(advMsg.MsgLength());
            advMsg.Pack(reinterpret_cast<char*>(&buffer[0]));
            msgLength = static_cast<int>(advMsg.MsgLength());
            break;
          }
          case SubType:
          {
            // Create the [UN]SUBSCRIBE message.
            SubscriptionMsg subMsg(header, topic);

            // Allocate a buffer and serialize the message.
            buffer.resize(subMsg.MsgLength());
            subMsg.Pack(reinterpret_cast<char*>(&buffer[0]));
            msgLength = static_cast<int>(subMsg.MsgLength());
            break;
          }
          case HeartbeatType:
          case ByeType:
          {
            // Allocate a buffer and serialize the message.
            buffer.resize(header.HeaderLength());
            header.Pack(reinterpret_cast<char*>(&buffer[0]));
            msgLength = header.HeaderLength();
            break;
          }
          default:
            std::cerr << "Discovery::SendMsg() error: Unrecognized message"
                      << " type [" << _type << "]" << std::endl;
            return;
        }

        // Send the discovery message to the multicast group through all the
        // sockets.
        for (const auto &sock : this->Sockets())
        {
          if (sendto(sock, reinterpret_cast<const raw_type *>(
            reinterpret_cast<unsigned char*>(&buffer[0])),
            msgLength, 0,
            reinterpret_cast<const sockaddr *>(this->MulticastAddr()),
            sizeof(*(this->MulticastAddr()))) != msgLength)
          {
            std::cerr << "Exception sending a message" << std::endl;
            return;
          }
        }

        if (this->Verbose())
        {
          std::cout << "\t* Sending " << MsgTypesStr[_type]
                    << " msg [" << topic << "]" << std::endl;
        }
      }

      private: const std::vector<int> &Sockets() const
      {
        return this->sockets;
      }

      private: const sockaddr_in *MulticastAddr() const
      {
        return &this->mcastAddr;
      }

      private: bool Verbose() const
      {
        return this->verbose;
      }

      private: uint8_t Version() const
      {
        return this->kWireVersion;
      }

      private: bool RegisterNetIface(const std::string &_ip)
      {
        // Make a new socket for sending discovery information.
        int sock = static_cast<int>(socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP));
        if (sock < 0)
        {
          std::cerr << "Socket creation failed." << std::endl;
          return false;
        }

        // Socket option: IP_MULTICAST_IF.
        // This socket option needs to be applied to each socket used to send
        // data. This option selects the source interface for outgoing messages.
        struct in_addr ifAddr;
        ifAddr.s_addr = inet_addr(_ip.c_str());
        if (setsockopt(sock, IPPROTO_IP, IP_MULTICAST_IF,
          reinterpret_cast<const char*>(&ifAddr), sizeof(ifAddr)) != 0)
        {
          std::cerr << "Error setting socket option (IP_MULTICAST_IF)."
                    << std::endl;
          return false;
        }

        this->sockets.push_back(sock);

        // Join the multicast group. We have to do it for each network interface
        // but we can do it on the same socket. We will use the socket at
        // position 0 for receiving multicast information.
        struct ip_mreq group;
        group.imr_multiaddr.s_addr =
          inet_addr(this->kMulticastGroup.c_str());
        group.imr_interface.s_addr = inet_addr(_ip.c_str());
        if (setsockopt(this->sockets.at(0), IPPROTO_IP, IP_ADD_MEMBERSHIP,
          reinterpret_cast<const char*>(&group), sizeof(group)) != 0)
        {
          std::cerr << "Error setting socket option (IP_ADD_MEMBERSHIP)."
                    << std::endl;
          return false;
        }

        return true;
      }

      private: static const unsigned int kDefActivityInterval = 100;

      private: static const unsigned int kDefHeartbeatInterval = 1000;

      private: static const unsigned int kDefSilenceInterval = 3000;

      private: const std::string kMulticastGroup = "224.0.0.7";

      private: const int kTimeout = 250;

      private: static const int kMaxRcvStr = 65536;

      private: static const uint8_t kWireVersion = 8;

      private: int port;

      private: std::string hostAddr;

      private: std::vector<std::string> hostInterfaces;

      private: std::string pUuid;

      private: unsigned int silenceInterval;

      private: unsigned int activityInterval;

      private: unsigned int heartbeatInterval;

      private: DiscoveryCallback<Pub> connectionCb;

      private: DiscoveryCallback<Pub> disconnectionCb;

      private: TopicStorage<Pub> info;

      protected: std::map<std::string, Timestamp> activity;

      private: bool verbose;

      private: std::vector<int> sockets;

      private: sockaddr_in mcastAddr;

      private: mutable std::mutex mutex;

      private: std::thread threadReception;

      private: Timestamp timeNextHeartbeat;

      private: Timestamp timeNextActivity;

      private: std::mutex exitMutex;

      private: bool initialized;

      private: unsigned int numHeartbeatsUninitialized;

      private: mutable std::condition_variable initializedCv;

      private: bool exit;

#ifdef _WIN32
      private: bool threadReceptionExiting = true;
#endif

      private: bool enabled;
    };

    using MsgDiscovery = Discovery<MessagePublisher>;

    using SrvDiscovery = Discovery<ServicePublisher>;
  }
}

#endif