Merge pull request #546 from mtconnect/538_fix_check_timestamps

Author: wsobel

src/mtconnect/pipeline/correct_timestamp.hpp

@@ -74,7 +74,7 @@ namespace mtconnect::pipeline {
         }
       }
 
-      m_state->m_timestamps.emplace(id, ts);
+      m_state->m_timestamps.insert_or_assign(id, ts);
 
       return next(std::move(obs));
     }

test_package/correct_timestamp_test.cpp

@@ -165,3 +165,51 @@ TEST_F(ValidateTimestampTest, should_change_timestamp_if_time_is_moving_backward
   ASSERT_NE(now - 1s, obs2->getTimestamp());
   ASSERT_LE(now, obs2->getTimestamp());
 }
+
+TEST_F(ValidateTimestampTest, should_handle_timestamp_in_the_future)
+{
+  makeDataItem({{"id", "a"s}, {"type", "EXECUTION"s}, {"category", "EVENT"s}});
+  
+  auto filter = make_shared<CorrectTimestamp>(m_context);
+  m_mapper->bind(filter);
+  filter->bind(make_shared<DeliverObservation>(m_context));
+  
+  auto now = chrono::system_clock::now();
+  
+  {
+    auto os = observe({"a", "READY"}, now - 1s);
+    auto list = os->getValue<EntityList>();
+    ASSERT_EQ(1, list.size());
+    
+    auto obs = dynamic_pointer_cast<Observation>(list.front());
+    ASSERT_EQ(now - 1s, obs->getTimestamp());
+  }
+  
+  {
+    auto os = observe({"a", "ACTIVE"}, now + 1s);
+    auto list = os->getValue<EntityList>();
+    ASSERT_EQ(1, list.size());
+    
+    auto obs = dynamic_pointer_cast<Observation>(list.front());
+    ASSERT_EQ(now + 1s, obs->getTimestamp());
+  }
+  
+  {
+    auto os = observe({"a", "READY"}, now);
+    auto list = os->getValue<EntityList>();
+    ASSERT_EQ(1, list.size());
+    
+    auto obs = dynamic_pointer_cast<Observation>(list.front());
+    ASSERT_LT(now, obs->getTimestamp());
+    ASSERT_GT(now + 10ms, obs->getTimestamp());
+  }
+  
+  {
+    auto os = observe({"a", "ACTIVE"}, now + 2s);
+    auto list = os->getValue<EntityList>();
+    ASSERT_EQ(1, list.size());
+    
+    auto obs = dynamic_pointer_cast<Observation>(list.front());
+    ASSERT_EQ(now + 2s, obs->getTimestamp());
+  }
+}