Adding option to ignore local messages.

Signed-off-by: Carlos Agüero <[email protected]>

include/gz/transport/SubscribeOptions.hh

@@ -66,6 +66,18 @@ namespace gz
       /// \return The maximum number of messages per second.
       public: uint64_t MsgsPerSec() const;
 
+      /// \brief Set the value to ignore local messages or not.
+      /// \param[in] _ignore True when ignoring local messages
+      /// or false otherwise.
+      /// \sa IgnoreLocalMessages
+      public: void SetIgnoreLocalMessages(bool _ignore);
+
+      /// \brief Whether the local messages should be ignored.
+      /// \return true when the local messages should be ignored or
+      /// false otherwise.
+      /// \sa SetIgnoreLocalMessages
+      public: bool IgnoreLocalMessages() const;
+
 #ifdef _WIN32
 // Disable warning C4251 which is triggered by
 // std::unique_ptr

include/gz/transport/SubscriptionHandler.hh

@@ -80,6 +80,10 @@ namespace gz
       /// \return A string representation of the handler UUID.
       public: std::string HandlerUuid() const;
 
+      /// \brief Return whether local messages are ignored or not.
+      /// \return True when local messages are ignored or false otherwise.
+      public: bool IgnoreLocalMessages() const;
+
       /// \brief Check if message subscription is throttled. If so, verify
       /// whether the callback should be executed or not.
       /// \return true if the callback should be executed or false otherwise.

src/Node.cc

@@ -338,6 +338,8 @@ bool Node::Publisher::Publish(const ProtoMsg &_msg)
     pubMsgDetails->msgCopy.reset(_msg.New());
     pubMsgDetails->msgCopy->CopyFrom(_msg);
 
+    pubMsgDetails->publisherNodeUUID = this->dataPtr->publisher.NUuid();
+
     if (subscribers.haveLocal)
     {
       for (const auto &node : subscribers.localHandlers)

src/NodeShared.cc

@@ -1849,6 +1849,14 @@ void NodeSharedPrivate::PublishThread()
     // Send the message to all the local handlers.
     for (auto &handler : msgDetails->localHandlers)
     {
+
+      // Check here if we want to ignore local publications.
+      if (handler->IgnoreLocalMessages() &&
+          msgDetails->publisherNodeUUID == handler->NodeUuid())
+      {
+        continue;
+      }
+
       try
       {
         handler->RunLocalCallback(*(msgDetails->msgCopy.get()),

src/NodeSharedPrivate.hh

@@ -167,6 +167,9 @@ namespace gz
 
                 /// \brief Information about the topic and type.
                 public: MessageInfo info;
+
+                /// \brief Publisher's node UUID.
+                public: std::string publisherNodeUUID;
               };
 
       /// \brief Publish thread used to process the pubQueue.

src/Node_TEST.cc

@@ -2020,6 +2020,41 @@ TEST(NodeTest, PubThrottled)
   reset();
 }
 
+//////////////////////////////////////////////////
+/// \brief This test creates one local publisher and subscriber and
+/// checks that no messages are received when using SetIgnoreLocalMessages
+/// is set to true.
+TEST(NodeTest, IgnoreLocalMessages)
+{
+  reset();
+
+  msgs::Int32 msg;
+  msg.set_data(data);
+
+  transport::Node node;
+
+  auto pub = node.Advertise<msgs::Int32>(g_topic);
+  EXPECT_TRUE(pub);
+
+  transport::SubscribeOptions opts;
+  EXPECT_FALSE(opts.IgnoreLocalMessages())
+  opts.SetIgnoreLocalMessages(true);
+  EXPECT_TRUE(opts.IgnoreLocalMessages())
+  EXPECT_TRUE(node.Subscribe(g_topic, cb, opts));
+
+  // Should be true the first time
+  for (auto i = 0; i < 3; ++i)
+  {
+    EXPECT_TRUE(pub.Publish(msg));
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  }
+
+  // No messages should be received.
+  EXPECT_EQ(0, counter);
+
+  reset();
+}
+
 //////////////////////////////////////////////////
 /// \brief This test spawns a service responser and a service requester. The
 /// requester uses a wrong type for the request argument. The test should verify

src/SubscribeOptions.cc

@@ -36,6 +36,7 @@ SubscribeOptions::SubscribeOptions(const SubscribeOptions &_otherSubscribeOpts)
   : dataPtr(new SubscribeOptionsPrivate())
 {
   this->SetMsgsPerSec(_otherSubscribeOpts.MsgsPerSec());
+  this->SetIgnoreLocalMessages(_otherSubscribeOpts.IgnoreLocalMessages());
 }
 
 //////////////////////////////////////////////////
@@ -60,3 +61,15 @@ void SubscribeOptions::SetMsgsPerSec(const uint64_t _newMsgsPerSec)
 {
   this->dataPtr->msgsPerSec = _newMsgsPerSec;
 }
+
+//////////////////////////////////////////////////
+bool SubscribeOptions::IgnoreLocalMessages() const
+{
+  return this->dataPtr->ignoreLocalMessages;
+}
+
+//////////////////////////////////////////////////
+void SubscribeOptions::SetIgnoreLocalMessages(bool _ignore)
+{
+  this->dataPtr->ignoreLocalMessages = _ignore;
+}

src/SubscribeOptionsPrivate.hh

@@ -41,6 +41,9 @@ namespace gz
 
       /// \brief Default message subscription rate.
       public: uint64_t msgsPerSec = kUnthrottled;
+
+      /// \brief Whether local messages should be ignored or not.
+      public: bool ignoreLocalMessages = false;
     };
     }
   }

src/SubscriptionHandler.cc

@@ -49,6 +49,12 @@ namespace gz
       return this->hUuid;
     }
 
+    /////////////////////////////////////////////////
+    bool SubscriptionHandlerBase::IgnoreLocalMessages() const
+    {
+      return this->opts.IgnoreLocalMessages();
+    }
+
     /////////////////////////////////////////////////
     bool SubscriptionHandlerBase::UpdateThrottling()
     {