Enable deadline tests for FastRTPS (#438)

* improve robustness of test_events

Signed-off-by: Miaofei <miaofei@amazon.com>

* refactor code

Signed-off-by: Miaofei <miaofei@amazon.com>

* enable testing of fastrtps

Signed-off-by: Miaofei <miaofei@amazon.com>

* fix failing tests and remove useless tests

Signed-off-by: Miaofei <miaofei@amazon.com>

* disable liveliness tests for fastrtps as it is not supported

Signed-off-by: Miaofei <miaofei@amazon.com>

rcl/test/rcl/test_events.cpp

@@ -27,11 +27,14 @@
 
 #include "osrf_testing_tools_cpp/scope_exit.hpp"
 
+using namespace std::chrono_literals;
 using std::chrono::milliseconds;
 using std::chrono::seconds;
+using std::chrono::duration_cast;
 
-constexpr seconds LIVELINESS_LEASE_DURATION_IN_S(1);
+constexpr seconds LIVELINESS_LEASE_DURATION_IN_S = 1s;
-constexpr seconds DEADLINE_PERIOD_IN_S(1);
+constexpr seconds DEADLINE_PERIOD_IN_S = 2s;
+constexpr seconds MAX_WAIT_PER_TESTCASE = 10s;
 
 #ifdef RMW_IMPLEMENTATION
 # define CLASSNAME_(NAME, SUFFIX) NAME ## __ ## SUFFIX
@@ -47,9 +50,7 @@ public:
   {
     is_opensplice = (std::string(rmw_get_implementation_identifier()).find("rmw_opensplice") == 0);
     is_fastrtps = (std::string(rmw_get_implementation_identifier()).find("rmw_fastrtps") == 0);
-
+    is_manual_liveliness_unsupported = is_fastrtps;
-    // TODO(mm318): Revisit once FastRTPS supports these QoS policies
-    is_unsupported = is_fastrtps;
 
     rcl_ret_t ret;
     {
@@ -82,6 +83,7 @@ public:
     // init publisher
     publisher = rcl_get_zero_initialized_publisher();
     rcl_publisher_options_t publisher_options = rcl_publisher_get_default_options();
+    publisher_options.qos.reliability = RMW_QOS_POLICY_RELIABILITY_BEST_EFFORT;
     publisher_options.qos.deadline = deadline;
     publisher_options.qos.lifespan = lifespan;
     publisher_options.qos.liveliness = liveliness_policy;
@@ -103,6 +105,7 @@ public:
     // init publisher
     subscription = rcl_get_zero_initialized_subscription();
     rcl_subscription_options_t subscription_options = rcl_subscription_get_default_options();
+    subscription_options.qos.reliability = RMW_QOS_POLICY_RELIABILITY_BEST_EFFORT;
     subscription_options.qos.deadline = deadline;
     subscription_options.qos.lifespan = lifespan;
     subscription_options.qos.liveliness = liveliness_policy;
@@ -125,7 +128,8 @@ public:
     rmw_time_t lifespan {0, 0};
     rmw_time_t deadline {DEADLINE_PERIOD_IN_S.count(), 0};
     rmw_time_t lease_duration {LIVELINESS_LEASE_DURATION_IN_S.count(), 0};
-    rmw_qos_liveliness_policy_t liveliness_policy = RMW_QOS_POLICY_LIVELINESS_AUTOMATIC;
+    rmw_qos_liveliness_policy_t liveliness_policy = is_manual_liveliness_unsupported ?
+      RMW_QOS_POLICY_LIVELINESS_AUTOMATIC : RMW_QOS_POLICY_LIVELINESS_MANUAL_BY_TOPIC;
 
     // init publisher
     ret = setup_publisher(deadline, lifespan, lease_duration, liveliness_policy);
@@ -205,21 +209,21 @@ protected:
   rcl_event_t subscription_event;
   bool is_fastrtps;
   bool is_opensplice;
-  bool is_unsupported;
+  bool is_manual_liveliness_unsupported;
   const char * topic = "rcl_test_publisher_subscription_events";
   const rosidl_message_type_support_t * ts;
 };
 
 rcl_ret_t
 wait_for_msgs_and_events(
+  rcl_context_t * context,
   rcl_subscription_t * subscription,
   rcl_event_t * subscription_event,
   rcl_event_t * publisher_event,
-  rcl_context_t * context,
-  int64_t period_ms,
   bool * msg_ready,
   bool * subscription_event_ready,
-  bool * publisher_event_ready)
+  bool * publisher_event_ready,
+  seconds period = 1s)
 {
   *msg_ready = false;
   *subscription_event_ready = false;
@@ -256,7 +260,7 @@ wait_for_msgs_and_events(
     EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
   }
 
-  ret = rcl_wait(&wait_set, RCL_MS_TO_NS(period_ms));
+  ret = rcl_wait(&wait_set, RCL_S_TO_NS(period.count()));
   if (ret == RCL_RET_TIMEOUT) {
     return ret;
   }
@@ -277,7 +281,7 @@ wait_for_msgs_and_events(
     }
   }
 
-  return RCL_RET_OK;
+  return ret;
 }
 
 /// Conditional function for determining when the wait_for_msgs_and_events loop is complete.
@@ -297,79 +301,46 @@ using WaitConditionPredicate = std::function<
   )>;
 
 // Wait for msgs and events until all conditions are met or a timeout has occured.
-template<typename S, typename P>
 rcl_ret_t
 conditional_wait_for_msgs_and_events(
+  rcl_context_t * context,
+  seconds timeout,
   const WaitConditionPredicate & events_ready,
   rcl_subscription_t * subscription,
   rcl_event_t * subscription_event,
   rcl_event_t * publisher_event,
-  rcl_context_t * context,
-  int64_t period_ms,
   bool * msg_persist_ready,
   bool * subscription_persist_ready,
-  bool * publisher_persist_ready,
+  bool * publisher_persist_ready)
-  test_msgs__msg__Strings * msg,
-  S * subscription_discrete_event,
-  P * publisher_discrete_event)
 {
   *msg_persist_ready = false;
   *subscription_persist_ready = false;
   *publisher_persist_ready = false;
-  auto timeout = milliseconds(2000);
+
   auto start_time = std::chrono::system_clock::now();
+  while (std::chrono::system_clock::now() - start_time < timeout) {
     bool msg_ready, subscription_event_ready, publisher_event_ready;
-  do {
+
     // wait for msg and events
-    wait_for_msgs_and_events(subscription, subscription_event, publisher_event,
+    rcl_ret_t ret = wait_for_msgs_and_events(
-      context, period_ms, &msg_ready, &subscription_event_ready, &publisher_event_ready);
+      context, subscription, subscription_event, publisher_event,
-    // test that the message published to topic is as expected
+      &msg_ready, &subscription_event_ready, &publisher_event_ready);
-    if (msg_ready) {
+    if (RCL_RET_OK != ret) {
-      EXPECT_EQ(rcl_take(subscription, msg, nullptr, nullptr), RCL_RET_OK);
+      continue;
-    }
-    if (subscription_event_ready && subscription_discrete_event) {
-      EXPECT_EQ(rcl_take_event(subscription_event, subscription_discrete_event), RCL_RET_OK);
-    }
-    if (publisher_event_ready && publisher_discrete_event) {
-      EXPECT_EQ(rcl_take_event(publisher_event, publisher_discrete_event), RCL_RET_OK);
     }
+
     *msg_persist_ready |= msg_ready;
     *subscription_persist_ready |= subscription_event_ready;
     *publisher_persist_ready |= publisher_event_ready;
-    if (std::chrono::system_clock::now() - start_time > timeout) {
+    if (events_ready(*msg_persist_ready, *subscription_persist_ready, *publisher_persist_ready)) {
-      return RCL_RET_TIMEOUT;
-    }
-  } while (!(events_ready(*msg_persist_ready,
-    *subscription_persist_ready,
-    *publisher_persist_ready)));
       return RCL_RET_OK;
-}
-
-TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_unsupported_lifespan) {
-  if (is_unsupported) {
-    rmw_time_t deadline {0, 0};
-    rmw_time_t lifespan {1, 0};
-    rmw_time_t lease_duration {1, 0};
-    rmw_qos_liveliness_policy_t liveliness_policy = RMW_QOS_POLICY_LIVELINESS_AUTOMATIC;
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_subscriber(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized subscriber lifespan when unsupported";
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_publisher(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized publisher lifespan when unsupported";
-
-    lifespan = {0, 1};
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_subscriber(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized subscriber lifespan when unsupported";
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_publisher(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized publisher lifespan when unsupported";
     }
+  }
+  return RCL_RET_TIMEOUT;
 }
 
 TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_unsupported_liveliness) {
-  if (is_unsupported) {
+  if (is_manual_liveliness_unsupported) {
     rmw_time_t deadline {0, 0};
     rmw_time_t lifespan {0, 0};
     rmw_time_t lease_duration {0, 0};
@@ -395,192 +366,11 @@ TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_unsupported_livelin
   }
 }
 
-TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_unsupported_unsupported_deadline) {
-  if (is_unsupported) {
-    rmw_time_t deadline {1, 0};
-    rmw_time_t lifespan {0, 0};
-    rmw_time_t lease_duration {0, 0};
-    rmw_qos_liveliness_policy_t liveliness_policy = RMW_QOS_POLICY_LIVELINESS_AUTOMATIC;
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_subscriber(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized subscriber deadline when unsupported";
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_publisher(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized publisher deadline when unsupported";
-
-    deadline = {0, 1};
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_subscriber(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized subscriber deadline when unsupported";
-    EXPECT_EQ(RCL_RET_ERROR,
-      setup_publisher(deadline, lifespan, lease_duration,
-      liveliness_policy)) << "Initialized publisher deadline when unsupported";
-  }
-}
-
-/*
- * Basic test of publisher and subscriber liveliness events, with publisher killed
- */
-TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_pubsub_liveliness_kill_pub)
-{
-  if (is_unsupported) {
-    return;
-  }
-  setup_publisher_and_subscriber(RCL_PUBLISHER_LIVELINESS_LOST,
-    RCL_SUBSCRIPTION_LIVELINESS_CHANGED);
-  rcl_ret_t ret;
-
-  // publish message to topic
-  const char * test_string = "testing";
-  {
-    test_msgs__msg__Strings msg;
-    test_msgs__msg__Strings__init(&msg);
-    ASSERT_TRUE(rosidl_generator_c__String__assign(&msg.string_value, test_string));
-    ret = rcl_publish(&publisher, &msg, nullptr);
-    test_msgs__msg__Strings__fini(&msg);
-    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-  }
-
-  // kill the publisher
-  ret = rcl_event_fini(&publisher_event);
-  EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-  ret = rcl_publisher_fini(&publisher, this->node_ptr);
-  EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-
-  WaitConditionPredicate msg_and_subevent_ready = [](
-    const bool & msg_persist_ready,
-    const bool & subscription_persist_ready,
-    const bool &) {
-      return msg_persist_ready && subscription_persist_ready;
-    };
-  test_msgs__msg__Strings msg;
-  test_msgs__msg__Strings__init(&msg);
-  OSRF_TESTING_TOOLS_CPP_SCOPE_EXIT({
-    test_msgs__msg__Strings__fini(&msg);
-  });
-  rmw_liveliness_changed_status_t liveliness_status;
-  bool msg_persist_ready, subscription_persist_ready, publisher_persist_ready;
-  rcl_ret_t wait_res = conditional_wait_for_msgs_and_events<rmw_liveliness_changed_status_t, void>(
-    msg_and_subevent_ready, &subscription, &subscription_event, nullptr, context_ptr, 1000,
-    &msg_persist_ready, &subscription_persist_ready, &publisher_persist_ready,
-    &msg, &liveliness_status, nullptr);
-  EXPECT_EQ(wait_res, RCL_RET_OK);
-
-  // test that the message published to topic is as expected
-  EXPECT_TRUE(msg_persist_ready);
-  // test subscriber/datareader liveliness changed status
-  EXPECT_TRUE(subscription_persist_ready);
-  // test that the killed publisher/datawriter has no active events
-  EXPECT_FALSE(publisher_persist_ready);
-
-  if (msg_persist_ready) {
-    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-    EXPECT_EQ(std::string(msg.string_value.data, msg.string_value.size),
-      std::string(test_string));
-  }
-  if (subscription_persist_ready) {
-    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-    EXPECT_EQ(liveliness_status.alive_count, 0);
-    // TODO(mm3188): Connext and OpenSplice seem to be tracking alive_count_change differently.
-    //               Issue has been raised at https://github.com/ADLINK-IST/opensplice/issues/88
-    if (is_opensplice) {
-      EXPECT_EQ(liveliness_status.alive_count_change, 2);
-    } else {
-      EXPECT_EQ(liveliness_status.alive_count_change, 0);
-    }
-    EXPECT_EQ(liveliness_status.not_alive_count, 0);
-    EXPECT_EQ(liveliness_status.not_alive_count_change, 0);
-  }
-
-  // clean up
-  ret = rcl_event_fini(&subscription_event);
-  EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-  ret = rcl_subscription_fini(&subscription, this->node_ptr);
-  EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-}
-
-/*
- * Basic test of publisher and subscriber deadline events, with first message sent after deadline
- */
-TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_pubsub_deadline_missed)
-{
-  if (is_unsupported) {
-    return;
-  }
-  setup_publisher_and_subscriber(RCL_PUBLISHER_OFFERED_DEADLINE_MISSED,
-    RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED);
-  rcl_ret_t ret;
-
-  // publish message to topic
-  const char * test_string = "testing";
-  {
-    test_msgs__msg__Strings msg;
-    test_msgs__msg__Strings__init(&msg);
-    ASSERT_TRUE(rosidl_generator_c__String__assign(&msg.string_value, test_string));
-    ret = rcl_publish(&publisher, &msg, nullptr);
-    test_msgs__msg__Strings__fini(&msg);
-    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-  }
-
-  WaitConditionPredicate all_ready = [](
-    const bool & msg_persist_ready,
-    const bool & subscription_persist_ready,
-    const bool & publisher_persist_ready) {
-      return msg_persist_ready && subscription_persist_ready && publisher_persist_ready;
-    };
-  test_msgs__msg__Strings msg;
-  test_msgs__msg__Strings__init(&msg);
-  OSRF_TESTING_TOOLS_CPP_SCOPE_EXIT({
-    test_msgs__msg__Strings__fini(&msg);
-  });
-  rmw_offered_deadline_missed_status_t offered_deadline_status;
-  rmw_requested_deadline_missed_status_t requested_deadline_status;
-  bool msg_persist_ready, subscription_persist_ready, publisher_persist_ready;
-  rcl_ret_t wait_res = conditional_wait_for_msgs_and_events(
-    all_ready, &subscription, &subscription_event, &publisher_event, context_ptr, 1000,
-    &msg_persist_ready, &subscription_persist_ready, &publisher_persist_ready,
-    &msg, &requested_deadline_status, &offered_deadline_status);
-  EXPECT_EQ(wait_res, RCL_RET_OK);
-
-  // test that the message published to topic is as expected
-  if (msg_persist_ready) {
-    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-    EXPECT_EQ(std::string(msg.string_value.data, msg.string_value.size),
-      std::string(test_string));
-  }
-
-  if (subscription_persist_ready) {
-    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-    EXPECT_EQ(requested_deadline_status.total_count, 1);
-    EXPECT_EQ(requested_deadline_status.total_count_change, 1);
-  }
-
-  if (publisher_persist_ready) {
-    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
-    EXPECT_EQ(offered_deadline_status.total_count, 1);
-    EXPECT_EQ(offered_deadline_status.total_count_change, 1);
-  }
-  // test that the message published to topic is as expected
-  EXPECT_TRUE(msg_persist_ready);
-
-  // test subscriber/datareader deadline missed status
-  EXPECT_TRUE(subscription_persist_ready);
-
-  // test publisher/datawriter deadline missed status
-  EXPECT_TRUE(publisher_persist_ready);
-
-  // clean up
-  tear_down_publisher_subscriber();
-}
-
 /*
  * Basic test of publisher and subscriber deadline events, with first message sent before deadline
  */
 TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_pubsub_no_deadline_missed)
 {
-  if (is_unsupported) {
-    return;
-  }
   setup_publisher_and_subscriber(RCL_PUBLISHER_OFFERED_DEADLINE_MISSED,
     RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED);
   rcl_ret_t ret;
@@ -598,12 +388,14 @@ TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_pubsub_no_deadline_
 
   // wait for msg and events
   bool msg_ready, subscription_event_ready, publisher_event_ready;
-  EXPECT_EQ(wait_for_msgs_and_events(&subscription, &subscription_event, &publisher_event,
+  rcl_ret_t wait_res = wait_for_msgs_and_events(
-    context_ptr, 1000, &msg_ready, &subscription_event_ready, &publisher_event_ready), RCL_RET_OK);
+    context_ptr, &subscription, &subscription_event, &publisher_event,
+    &msg_ready, &subscription_event_ready, &publisher_event_ready, DEADLINE_PERIOD_IN_S);
+  EXPECT_EQ(wait_res, RCL_RET_OK);
 
   // test that the message published to topic is as expected
   EXPECT_TRUE(msg_ready);
-  {
+  if (msg_ready) {
     test_msgs__msg__Strings msg;
     test_msgs__msg__Strings__init(&msg);
     OSRF_TESTING_TOOLS_CPP_SCOPE_EXIT({
@@ -637,3 +429,159 @@ TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_pubsub_no_deadline_
   // clean up
   tear_down_publisher_subscriber();
 }
+
+/*
+ * Basic test of publisher and subscriber deadline events, with first message sent after deadline
+ */
+TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_pubsub_deadline_missed)
+{
+  setup_publisher_and_subscriber(RCL_PUBLISHER_OFFERED_DEADLINE_MISSED,
+    RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED);
+  rcl_ret_t ret;
+
+  // publish message to topic
+  const char * test_string = "testing";
+  {
+    test_msgs__msg__Strings msg;
+    test_msgs__msg__Strings__init(&msg);
+    ASSERT_TRUE(rosidl_generator_c__String__assign(&msg.string_value, test_string));
+    ret = rcl_publish(&publisher, &msg, nullptr);
+    test_msgs__msg__Strings__fini(&msg);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+  }
+
+  WaitConditionPredicate all_ready = [](
+    const bool & msg_persist_ready,
+    const bool & subscription_persist_ready,
+    const bool & publisher_persist_ready) {
+      return msg_persist_ready && subscription_persist_ready && publisher_persist_ready;
+    };
+  bool msg_persist_ready, subscription_persist_ready, publisher_persist_ready;
+  rcl_ret_t wait_res = conditional_wait_for_msgs_and_events(
+    context_ptr, MAX_WAIT_PER_TESTCASE, all_ready,
+    &subscription, &subscription_event, &publisher_event,
+    &msg_persist_ready, &subscription_persist_ready, &publisher_persist_ready);
+  EXPECT_EQ(wait_res, RCL_RET_OK);
+
+  // test that the message published to topic is as expected
+  EXPECT_TRUE(msg_persist_ready);
+  if (msg_persist_ready) {
+    test_msgs__msg__Strings msg;
+    test_msgs__msg__Strings__init(&msg);
+    OSRF_TESTING_TOOLS_CPP_SCOPE_EXIT({
+      test_msgs__msg__Strings__fini(&msg);
+    });
+    ret = rcl_take(&subscription, &msg, nullptr, nullptr);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+    EXPECT_EQ(std::string(msg.string_value.data, msg.string_value.size),
+      std::string(test_string));
+  }
+
+  // test subscriber/datareader deadline missed status
+  EXPECT_TRUE(subscription_persist_ready);
+  if (subscription_persist_ready) {
+    rmw_requested_deadline_missed_status_t requested_deadline_status;
+    ret = rcl_take_event(&subscription_event, &requested_deadline_status);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+    EXPECT_EQ(requested_deadline_status.total_count, 1);
+    EXPECT_EQ(requested_deadline_status.total_count_change, 1);
+  }
+
+  // test publisher/datawriter deadline missed status
+  EXPECT_TRUE(publisher_persist_ready);
+  if (publisher_persist_ready) {
+    rmw_offered_deadline_missed_status_t offered_deadline_status;
+    ret = rcl_take_event(&publisher_event, &offered_deadline_status);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+    EXPECT_EQ(offered_deadline_status.total_count, 1);
+    EXPECT_EQ(offered_deadline_status.total_count_change, 1);
+  }
+
+  // clean up
+  tear_down_publisher_subscriber();
+}
+
+/*
+ * Basic test of publisher and subscriber liveliness events, with publisher killed
+ */
+TEST_F(CLASSNAME(TestEventFixture, RMW_IMPLEMENTATION), test_pubsub_liveliness_kill_pub)
+{
+  if (is_manual_liveliness_unsupported) {
+    return;
+  }
+
+  setup_publisher_and_subscriber(RCL_PUBLISHER_LIVELINESS_LOST,
+    RCL_SUBSCRIPTION_LIVELINESS_CHANGED);
+  rcl_ret_t ret;
+
+  // publish message to topic
+  const char * test_string = "testing";
+  {
+    test_msgs__msg__Strings msg;
+    test_msgs__msg__Strings__init(&msg);
+    ASSERT_TRUE(rosidl_generator_c__String__assign(&msg.string_value, test_string));
+    ret = rcl_publish(&publisher, &msg, nullptr);
+    test_msgs__msg__Strings__fini(&msg);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+  }
+
+  std::this_thread::sleep_for(2 * LIVELINESS_LEASE_DURATION_IN_S);
+
+  WaitConditionPredicate all_ready = [](
+    const bool & msg_persist_ready,
+    const bool & subscription_persist_ready,
+    const bool & publisher_persist_ready) {
+      return msg_persist_ready && subscription_persist_ready && publisher_persist_ready;
+    };
+  bool msg_persist_ready, subscription_persist_ready, publisher_persist_ready;
+  rcl_ret_t wait_res = conditional_wait_for_msgs_and_events(
+    context_ptr, MAX_WAIT_PER_TESTCASE, all_ready,
+    &subscription, &subscription_event, &publisher_event,
+    &msg_persist_ready, &subscription_persist_ready, &publisher_persist_ready);
+  EXPECT_EQ(wait_res, RCL_RET_OK);
+
+  // test that the message published to topic is as expected
+  EXPECT_TRUE(msg_persist_ready);
+  if (msg_persist_ready) {
+    test_msgs__msg__Strings msg;
+    test_msgs__msg__Strings__init(&msg);
+    OSRF_TESTING_TOOLS_CPP_SCOPE_EXIT({
+      test_msgs__msg__Strings__fini(&msg);
+    });
+    ret = rcl_take(&subscription, &msg, nullptr, nullptr);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+    EXPECT_EQ(std::string(msg.string_value.data, msg.string_value.size),
+      std::string(test_string));
+  }
+
+  // test subscriber/datareader liveliness changed status
+  EXPECT_TRUE(subscription_persist_ready);
+  if (subscription_persist_ready) {
+    rmw_liveliness_changed_status_t liveliness_status;
+    ret = rcl_take_event(&subscription_event, &liveliness_status);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+    EXPECT_EQ(liveliness_status.alive_count, 0);
+    // TODO(mm3188): Connext and OpenSplice seem to be tracking alive_count_change differently.
+    //               Issue has been raised at https://github.com/ADLINK-IST/opensplice/issues/88
+    if (is_opensplice) {
+      EXPECT_EQ(liveliness_status.alive_count_change, 2);
+    } else {
+      EXPECT_EQ(liveliness_status.alive_count_change, 0);
+    }
+    EXPECT_EQ(liveliness_status.not_alive_count, 1);
+    EXPECT_EQ(liveliness_status.not_alive_count_change, 1);
+  }
+
+  // test that the killed publisher/datawriter has no active events
+  EXPECT_TRUE(publisher_persist_ready);
+  if (publisher_persist_ready) {
+    rmw_liveliness_lost_status_t liveliness_status;
+    ret = rcl_take_event(&publisher_event, &liveliness_status);
+    EXPECT_EQ(ret, RCL_RET_OK) << rcl_get_error_string().str;
+    EXPECT_EQ(liveliness_status.total_count, 1);
+    EXPECT_EQ(liveliness_status.total_count_change, 1);
+  }
+
+  // clean up
+  tear_down_publisher_subscriber();
+}