Ref. 7288 Change removed when not on instance [7289]

src/cpp/subscriber/SubscriberHistory.cpp

@@ -415,37 +415,34 @@ bool SubscriberHistory::remove_change_sub(
     }
 
     std::lock_guard<RecursiveTimedMutex> guard(*mp_mutex);
-    if (mp_subImpl->getAttributes().topic.getTopicKind() == NO_KEY)
-    {
-        if (remove_change(change))
-        {
-            m_isHistoryFull = false;
-            return true;
-        }
-        return false;
-    }
-    else
+    if (mp_subImpl->getAttributes().topic.getTopicKind() == WITH_KEY)
     {
+        bool found = false;
         t_m_Inst_Caches::iterator vit;
-        if (!find_key(change, &vit))
+        if (find_key(change, &vit))
         {
-            return false;
-        }
-
-        for (auto chit = vit->second.cache_changes.begin(); chit != vit->second.cache_changes.end(); ++chit)
-        {
-            if ((*chit)->sequenceNumber == change->sequenceNumber && (*chit)->writerGUID == change->writerGUID)
+            for (auto chit = vit->second.cache_changes.begin(); chit != vit->second.cache_changes.end(); ++chit)
             {
-                if (remove_change(change))
+                if ((*chit)->sequenceNumber == change->sequenceNumber && (*chit)->writerGUID == change->writerGUID)
                 {
                     vit->second.cache_changes.erase(chit);
-                    m_isHistoryFull = false;
-                    return true;
+                    found = true;
+                    break;
                 }
             }
         }
-        logError(SUBSCRIBER, "Change not found, something is wrong");
+        if (!found)
+        {
+            logWarning(SUBSCRIBER, "Change not found, something is wrong");
+        }
     }
+
+    if (remove_change(change))
+    {
+        m_isHistoryFull = false;
+        return true;
+    }
+
     return false;
 }
 

test/blackbox/BlackboxTests.hpp

@@ -59,6 +59,9 @@ void default_receive_print(const HelloWorld& hello);
 template<>
 void default_receive_print(const FixedSized& hello);
 
+template<>
+void default_receive_print(const KeyedHelloWorld& str);
+
 template<>
 void default_receive_print(const String& str);
 
@@ -83,6 +86,9 @@ void default_send_print(const HelloWorld& hello);
 template<>
 void default_send_print(const FixedSized& hello);
 
+template<>
+void default_send_print(const KeyedHelloWorld& str);
+
 template<>
 void default_send_print(const String& str);
 

test/blackbox/BlackboxTestsPubSubHistory.cpp

@@ -472,6 +472,94 @@ TEST_P(PubSubHistory, PubSubAsReliableKeepLastReaderSmallDepthTwoPublishers)
     ASSERT_EQ(received.index(), 3u);
 }
 
+TEST_P(PubSubHistory, PubSubAsReliableKeepLastWithKey)
+{
+    PubSubReader<KeyedHelloWorldType> reader(TEST_TOPIC_NAME);
+    PubSubWriter<KeyedHelloWorldType> writer(TEST_TOPIC_NAME);
+
+    uint32_t keys = 2;
+
+    reader.resource_limits_max_instances(keys).
+    reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+    history_kind(eprosima::fastrtps::KEEP_LAST_HISTORY_QOS).
+    key(true).init();
+
+    ASSERT_TRUE(reader.isInitialized());
+
+    writer.resource_limits_max_instances(keys).
+    reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+    key(true).init();
+
+    ASSERT_TRUE(writer.isInitialized());
+
+    // Wait for discovery.
+    writer.wait_discovery();
+    reader.wait_discovery();
+
+    auto data = default_keyedhelloworld_data_generator();
+    reader.startReception(data);
+
+    // Send data
+    writer.send(data);
+    ASSERT_TRUE(data.empty());
+
+    reader.block_for_all();
+    reader.stopReception();
+}
+
+TEST_P(PubSubHistory, PubSubAsReliableKeepLastReaderSmallDepthWithKey)
+{
+    PubSubReader<KeyedHelloWorldType> reader(TEST_TOPIC_NAME);
+    PubSubWriter<KeyedHelloWorldType> writer(TEST_TOPIC_NAME);
+
+    uint32_t keys = 2;
+    uint32_t depth = 2;
+
+    reader.history_depth(depth).
+    resource_limits_max_samples(keys*depth).
+    resource_limits_allocated_samples(keys*depth).
+    resource_limits_max_instances(keys).
+    resource_limits_max_samples_per_instance(depth).
+    reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+    history_kind(eprosima::fastrtps::KEEP_LAST_HISTORY_QOS).
+    key(true).init();
+
+    ASSERT_TRUE(reader.isInitialized());
+
+    writer.history_depth(depth).
+    resource_limits_max_samples(keys*depth).
+    resource_limits_allocated_samples(keys*depth).
+    resource_limits_max_instances(keys).
+    resource_limits_max_samples_per_instance(depth).
+    reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+    key(true).init();
+
+    ASSERT_TRUE(writer.isInitialized());
+
+    // Wait for discovery.
+    writer.wait_discovery();
+    reader.wait_discovery();
+
+    //We want the number of messages to be multiple of keys*depth
+    auto data = default_keyedhelloworld_data_generator(3*keys*depth);
+    while(data.size() > 1)
+    {
+        auto expected_data(data);
+
+        // Send data
+        writer.send(data);
+        ASSERT_TRUE(data.empty());
+        std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+        reader.startReception(expected_data);
+        size_t current_received = reader.block_for_at_least(keys*depth);
+        reader.stopReception();
+        ASSERT_EQ(current_received, keys*depth);
+
+        data = reader.data_not_received();
+    }
+}
+
 INSTANTIATE_TEST_CASE_P(PubSubHistory,
         PubSubHistory,
         testing::Values(false, true),

test/blackbox/PubSubReader.hpp

@@ -546,6 +546,18 @@ class PubSubReader
         return *this;
     }
 
+    PubSubReader& resource_limits_max_instances(const int32_t max)
+    {
+        subscriber_attr_.topic.resourceLimitsQos.max_instances = max;
+        return *this;
+    }
+
+    PubSubReader& resource_limits_max_samples_per_instance(const int32_t max)
+    {
+        subscriber_attr_.topic.resourceLimitsQos.max_samples_per_instance = max;
+        return *this;
+    }
+
     PubSubReader& matched_writers_allocation(size_t initial, size_t maximum)
     {
         subscriber_attr_.matched_publisher_allocation.initial = initial;

test/blackbox/PubSubWriter.hpp

@@ -576,6 +576,18 @@ class PubSubWriter
         return *this;
     }
 
+    PubSubWriter& resource_limits_max_instances(const int32_t max)
+    {
+        publisher_attr_.topic.resourceLimitsQos.max_instances = max;
+        return *this;
+    }
+
+    PubSubWriter& resource_limits_max_samples_per_instance(const int32_t max)
+    {
+        publisher_attr_.topic.resourceLimitsQos.max_samples_per_instance = max;
+        return *this;
+    }
+
     PubSubWriter& matched_readers_allocation(size_t initial, size_t maximum)
     {
         publisher_attr_.matched_subscriber_allocation.initial = initial;

test/blackbox/types/KeyedHelloWorld.cpp

@@ -35,8 +35,8 @@ using namespace eprosima::fastcdr::exception;
 
 KeyedHelloWorld::KeyedHelloWorld()
 {
+    m_index = 0;
     m_key = 0;
-
 }
 
 KeyedHelloWorld::~KeyedHelloWorld()
@@ -45,18 +45,21 @@ KeyedHelloWorld::~KeyedHelloWorld()
 
 KeyedHelloWorld::KeyedHelloWorld(const KeyedHelloWorld &x)
 {
+    m_index = x.m_index;
     m_key = x.m_key;
     m_message = x.m_message;
 }
 
 KeyedHelloWorld::KeyedHelloWorld(KeyedHelloWorld &&x)
 {
+    m_index = x.m_index;
     m_key = x.m_key;
     m_message = std::move(x.m_message);
 }
 
 KeyedHelloWorld& KeyedHelloWorld::operator=(const KeyedHelloWorld &x)
 {
+    m_index = x.m_index;
     m_key = x.m_key;
     m_message = x.m_message;
 
@@ -65,6 +68,7 @@ KeyedHelloWorld& KeyedHelloWorld::operator=(const KeyedHelloWorld &x)
 
 KeyedHelloWorld& KeyedHelloWorld::operator=(KeyedHelloWorld &&x)
 {
+    m_index = x.m_index;
     m_key = x.m_key;
     m_message = std::move(x.m_message);
 
@@ -73,7 +77,8 @@ KeyedHelloWorld& KeyedHelloWorld::operator=(KeyedHelloWorld &&x)
 
 bool KeyedHelloWorld::operator==(const KeyedHelloWorld &x) const
 {
-    if(m_message == x.m_message &&
+    if(m_index == x.m_index &&
+            m_message == x.m_message &&
             m_key == x.m_key)
         return true;
 
@@ -99,6 +104,7 @@ size_t KeyedHelloWorld::getCdrSerializedSize(const KeyedHelloWorld& data, size_t
 void KeyedHelloWorld::serialize(eprosima::fastcdr::Cdr &scdr) const
 {
     scdr << m_key;
+    scdr << m_index;
 
     if(m_message.length() <= 256)
     {
@@ -113,6 +119,7 @@ void KeyedHelloWorld::serialize(eprosima::fastcdr::Cdr &scdr) const
 void KeyedHelloWorld::deserialize(eprosima::fastcdr::Cdr &dcdr)
 {
     dcdr >> m_key;
+    dcdr >> m_index;
     dcdr >> m_message;
 }
 

test/blackbox/types/KeyedHelloWorld.h

@@ -114,6 +114,33 @@ class KeyedHelloWorld
     {
         return m_key;
     }
+
+    /*
+     * @brief This function sets a value in member index
+     * @param _index New value for member index
+     */
+   inline eProsima_user_DllExport void index(uint16_t _index)
+   {
+       m_index = _index;
+   }
+
+   /*!
+    * @brief This function returns the value of member index
+    * @return Value of member index
+    */
+   inline eProsima_user_DllExport uint16_t index() const
+   {
+       return m_index;
+   }
+
+   /*!
+    * @brief This function returns a reference to member index
+    * @return Reference to member index
+    */
+   inline eProsima_user_DllExport uint16_t& index()
+   {
+       return m_index;
+   }
     /*!
      * @brief This function copies the value in member message
      * @param _message New value to be copied in member message
@@ -201,6 +228,7 @@ class KeyedHelloWorld
     eProsima_user_DllExport void serializeKey(eprosima::fastcdr::Cdr &cdr) const;
 
 private:
+    uint16_t m_index;
     uint16_t m_key;
     std::string m_message;
 };

test/blackbox/utils/data_generators.cpp

@@ -60,6 +60,7 @@ std::list<KeyedHelloWorld> default_keyedhelloworld_data_generator(size_t max)
     std::generate(returnedValue.begin(), returnedValue.end(), [&index]
     {
         KeyedHelloWorld hello;
+        hello.index(index);
         hello.key(index % 2);
         std::stringstream ss;
         ss << "HelloWorld " << index;

test/blackbox/utils/print_functions.cpp

@@ -20,6 +20,12 @@ void default_receive_print(const HelloWorld& hello)
     std::cout << "Received HelloWorld " << hello.index() << std::endl;
 }
 
+template<>
+void default_receive_print(const KeyedHelloWorld& hello)
+{
+    std::cout << "Received HelloWorld " << hello.index() << " with key " << hello.key() << std::endl;
+}
+
 template<>
 void default_receive_print(const FixedSized& hello)
 {
@@ -57,6 +63,12 @@ void default_send_print(const HelloWorld& hello)
     std::cout << "Sent HelloWorld " << hello.index() << std::endl;
 }
 
+template<>
+void default_send_print(const KeyedHelloWorld& hello)
+{
+    std::cout << "Sent HelloWorld " << hello.index() << " with key " << hello.key() << std::endl;
+}
+
 template<>
 void default_send_print(const FixedSized& hello)
 {