fix minor failures

Signed-off-by: Alberto Soragna <[email protected]>

Author: alsora

rclcpp/include/rclcpp/subscription.hpp

@@ -171,6 +171,10 @@ class Subscription : public SubscriptionBase
 
       // First create a SubscriptionIntraProcess which will be given to the intra-process manager.
       auto context = node_base->get_context();
+      using SubscriptionIntraProcessT = rclcpp::experimental::SubscriptionIntraProcess<
+        CallbackMessageT,
+        AllocatorT,
+        typename MessageUniquePtr::deleter_type>;
       subscription_intra_process_ = std::make_shared<SubscriptionIntraProcessT>(
         callback,
         options.get_allocator(),

rclcpp/include/rclcpp/subscription_base.hpp

@@ -374,7 +374,13 @@ class SubscriptionBase : public std::enable_shared_from_this<SubscriptionBase>
       return;
     }
 
-    // The on_ready_callback signature has an extra `int` argument used to disambiguate between 
+    if (!callback) {
+      throw std::invalid_argument(
+              "The callback passed to set_on_new_intra_process_message_callback "
+              "is not callable.");
+    }
+
+    // The on_ready_callback signature has an extra `int` argument used to disambiguate between
     // possible different entities within a generic waitable.
     // We hide that detail to users of this method.
     std::function<void(size_t, int)> new_callback = std::bind(callback, std::placeholders::_1);
@@ -435,7 +441,7 @@ class SubscriptionBase : public std::enable_shared_from_this<SubscriptionBase>
   bool use_intra_process_;
   IntraProcessManagerWeakPtr weak_ipm_;
   uint64_t intra_process_subscription_id_;
-  std::shared_ptr<SubscriptionIntraProcessBase> subscription_intra_process_;
+  std::shared_ptr<rclcpp::experimental::SubscriptionIntraProcessBase> subscription_intra_process_;
 
 private:
   RCLCPP_DISABLE_COPY(SubscriptionBase)

rclcpp/test/rclcpp/test_subscription.cpp

@@ -448,7 +448,7 @@ TEST_F(TestSubscription, on_new_message_callback) {
   using test_msgs::msg::Empty;
 
   auto do_nothing = [](std::shared_ptr<const test_msgs::msg::Empty>) {FAIL();};
-  auto sub = node->create_subscription<test_msgs::msg::Empty>("~/test_take", 1, do_nothing);
+  auto sub = node->create_subscription<test_msgs::msg::Empty>("~/test_take", 10, do_nothing);
 
   std::atomic<size_t> c1 {0};
   auto increase_c1_cb = [&c1](size_t count_msgs) {c1 += count_msgs;};
@@ -518,7 +518,7 @@ TEST_F(TestSubscription, on_new_intra_process_message_callback) {
   using test_msgs::msg::Empty;
 
   auto do_nothing = [](std::shared_ptr<const test_msgs::msg::Empty>) {FAIL();};
-  auto sub = node->create_subscription<test_msgs::msg::Empty>("~/test_take", 1, do_nothing);
+  auto sub = node->create_subscription<test_msgs::msg::Empty>("~/test_take", 10, do_nothing);
 
   std::atomic<size_t> c1 {0};
   auto increase_c1_cb = [&c1](size_t count_msgs) {c1 += count_msgs;};