{

INVALID = 0,

ROS_MESSAGE = 1, // The subscription delivers a ROS message to its callback

SERIALIZED_MESSAGE = 2, // The subscription delivers a serialized message to its callback

DYNAMIC_MESSAGE = 3, // The subscription delivers a dynamic message to its callback

{

RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(SubscriptionBase)

/// Constructor.

/**

RCLCPP_PUBLIC

SubscriptionBase(

rclcpp::node_interfaces::NodeBaseInterface * node_base,

const rosidl_message_type_support_t & type_support_handle,

const std::string & topic_name,

const rcl_subscription_options_t & subscription_options,

const SubscriptionEventCallbacks & event_callbacks,

bool use_default_callbacks,

DeliveredMessageKind delivered_message_kind = DeliveredMessageKind::ROS_MESSAGE);

/// Destructor.

RCLCPP_PUBLIC

virtual ~SubscriptionBase();

/// Add event handlers for passed in event_callbacks.

RCLCPP_PUBLIC

void

bind_event_callbacks(

const SubscriptionEventCallbacks & event_callbacks, bool use_default_callbacks);

/// Get the topic that this subscription is subscribed on.

RCLCPP_PUBLIC

const char *

get_topic_name() const;

RCLCPP_PUBLIC

std::shared_ptr<rcl_subscription_t>

get_subscription_handle();

RCLCPP_PUBLIC

std::shared_ptr<const rcl_subscription_t>

get_subscription_handle() const;

/// Get all the QoS event handlers associated with this subscription.

/** \return The map of QoS event handlers. */

RCLCPP_PUBLIC

const

std::unordered_map<rcl_subscription_event_type_t, std::shared_ptr<rclcpp::EventHandlerBase>> &

get_event_handlers() const;

/// Get the actual QoS settings, after the defaults have been determined.

/**

RCLCPP_PUBLIC

rclcpp::QoS

get_actual_qos() const;

/// Take the next inter-process message from the subscription as a type erased pointer.

/**

RCLCPP_PUBLIC

bool

take_type_erased(void * message_out, rclcpp::MessageInfo & message_info_out);

/// Take the next inter-process message, in its serialized form, from the subscription.

/**

RCLCPP_PUBLIC

bool

take_serialized(rclcpp::SerializedMessage & message_out, rclcpp::MessageInfo & message_info_out);

/// Borrow a new message.

/** \return Shared pointer to the fresh message. */

RCLCPP_PUBLIC

virtual

std::shared_ptr<void>

create_message() = 0;

/// Borrow a new serialized message

/** \return Shared pointer to a rcl_message_serialized_t. */

RCLCPP_PUBLIC

virtual

std::shared_ptr<rclcpp::SerializedMessage>

create_serialized_message() = 0;

/// Check if we need to handle the message, and execute the callback if we do.

/**

RCLCPP_PUBLIC

virtual

void

handle_message(std::shared_ptr<void> & message, const rclcpp::MessageInfo & message_info) = 0;

RCLCPP_PUBLIC

virtual

void

handle_serialized_message(

const std::shared_ptr<rclcpp::SerializedMessage> & serialized_message,

const rclcpp::MessageInfo & message_info) = 0;

RCLCPP_PUBLIC

virtual

void

handle_loaned_message(void * loaned_message, const rclcpp::MessageInfo & message_info) = 0;

/// Return the message borrowed in create_message.

/** \param[in] message Shared pointer to the returned message. */

RCLCPP_PUBLIC

virtual

void

return_message(std::shared_ptr<void> & message) = 0;

/// Return the message borrowed in create_serialized_message.

/** \param[in] message Shared pointer to the returned message. */

RCLCPP_PUBLIC

virtual

void

return_serialized_message(std::shared_ptr<rclcpp::SerializedMessage> & message) = 0;

RCLCPP_PUBLIC

const rosidl_message_type_support_t &

get_message_type_support_handle() const;

/// Return if the subscription is serialized

/**

RCLCPP_PUBLIC

bool

is_serialized() const;

/// Return the delivered message kind.

/**

RCLCPP_PUBLIC

DeliveredMessageKind

get_delivered_message_kind() const;

/// Get matching publisher count.

/** \return The number of publishers on this topic. */

RCLCPP_PUBLIC

size_t

get_publisher_count() const;

/// Check if subscription instance can loan messages.

/**

RCLCPP_PUBLIC

bool

can_loan_messages() const;

using IntraProcessManagerWeakPtr =

std::weak_ptr<rclcpp::experimental::IntraProcessManager>;

/// Implemenation detail.

RCLCPP_PUBLIC

void

setup_intra_process(

uint64_t intra_process_subscription_id,

IntraProcessManagerWeakPtr weak_ipm);

/// Return the waitable for intra-process

/**

RCLCPP_PUBLIC

rclcpp::Waitable::SharedPtr

get_intra_process_waitable() const;

/// Exchange state of whether or not a part of the subscription is used by a wait set.

/**

RCLCPP_PUBLIC

bool

exchange_in_use_by_wait_set_state(void * pointer_to_subscription_part, bool in_use_state);

/// Get network flow endpoints

/**

RCLCPP_PUBLIC

std::vector<rclcpp::NetworkFlowEndpoint>

get_network_flow_endpoints() const;

/// Set a callback to be called when each new message is received.

/**

void

set_on_new_message_callback(std::function<void(size_t)> callback)

{

if (!callback) {

throw std::invalid_argument(

"The callback passed to set_on_new_message_callback "

"is not callable.");

}

auto new_callback =

[callback, this](size_t number_of_messages) {

try {

callback(number_of_messages);

} catch (const std::exception & exception) {

RCLCPP_ERROR_STREAM(

node_logger_,

"rclcpp::SubscriptionBase@" << this <<

" caught " << rmw::impl::cpp::demangle(exception) <<

" exception in user-provided callback for the 'on new message' callback: " <<

exception.what());

} catch (...) {

RCLCPP_ERROR_STREAM(

node_logger_,

"rclcpp::SubscriptionBase@" << this <<

" caught unhandled exception in user-provided callback " <<

"for the 'on new message' callback");

}

};

std::lock_guard<std::recursive_mutex> lock(callback_mutex_);

// Set it temporarily to the new callback, while we replace the old one.

// This two-step setting, prevents a gap where the old std::function has

// been replaced but the middleware hasn't been told about the new one yet.

set_on_new_message_callback(

rclcpp::detail::cpp_callback_trampoline<decltype(new_callback), const void *, size_t>,

static_cast<const void *>(&new_callback));

// Store the std::function to keep it in scope, also overwrites the existing one.

on_new_message_callback_ = new_callback;

// Set it again, now using the permanent storage.

set_on_new_message_callback(

rclcpp::detail::cpp_callback_trampoline<

decltype(on_new_message_callback_), const void *, size_t>,

static_cast<const void *>(&on_new_message_callback_));

}

/// Unset the callback registered for new messages, if any.

void

clear_on_new_message_callback()

{

std::lock_guard<std::recursive_mutex> lock(callback_mutex_);

if (on_new_message_callback_) {

set_on_new_message_callback(nullptr, nullptr);

on_new_message_callback_ = nullptr;

}

}

/// Set a callback to be called when each new intra-process message is received.

/**

void

set_on_new_intra_process_message_callback(std::function<void(size_t)> callback)

{

if (!use_intra_process_) {

RCLCPP_WARN(

rclcpp::get_logger("rclcpp"),

"Calling set_on_new_intra_process_message_callback for subscription with IPC disabled");

return;

}

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);

subscription_intra_process_->set_on_ready_callback(new_callback);

}

/// Unset the callback registered for new intra-process messages, if any.

void

clear_on_new_intra_process_message_callback()

{

if (!use_intra_process_) {

RCLCPP_WARN(

rclcpp::get_logger("rclcpp"),

"Calling clear_on_new_intra_process_message_callback for subscription with IPC disabled");

return;

}

subscription_intra_process_->clear_on_ready_callback();

}

/// Set a callback to be called when each new qos event instance occurs.

/**

void

set_on_new_qos_event_callback(

std::function<void(size_t)> callback,

rcl_subscription_event_type_t event_type)

{

if (event_handlers_.count(event_type) == 0) {

RCLCPP_WARN(

rclcpp::get_logger("rclcpp"),

"Calling set_on_new_qos_event_callback for non registered subscription event_type");

return;

}

if (!callback) {

throw std::invalid_argument(

"The callback passed to set_on_new_qos_event_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);

event_handlers_[event_type]->set_on_ready_callback(new_callback);

}

/// Unset the callback registered for new qos events, if any.

void

clear_on_new_qos_event_callback(rcl_subscription_event_type_t event_type)

{

if (event_handlers_.count(event_type) == 0) {

RCLCPP_WARN(

rclcpp::get_logger("rclcpp"),

"Calling clear_on_new_qos_event_callback for non registered event_type");

return;

}

event_handlers_[event_type]->clear_on_ready_callback();

}

/// Check if content filtered topic feature of the subscription instance is enabled.

/**

RCLCPP_PUBLIC

bool

is_cft_enabled() const;

/// Set the filter expression and expression parameters for the subscription.

/**

RCLCPP_PUBLIC

void

set_content_filter(

const std::string & filter_expression,

const std::vector<std::string> & expression_parameters = {});

/// Get the filter expression and expression parameters for the subscription.

/**

RCLCPP_PUBLIC

rclcpp::ContentFilterOptions

get_content_filter() const;

// DYNAMIC TYPE ==================================================================================

// TODO(methylDragon): Reorder later

RCLCPP_PUBLIC

virtual

rclcpp::dynamic_typesupport::DynamicMessageType::SharedPtr

get_shared_dynamic_message_type() = 0;

RCLCPP_PUBLIC

virtual

rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr

get_shared_dynamic_message() = 0;

RCLCPP_PUBLIC

virtual

rclcpp::dynamic_typesupport::DynamicSerializationSupport::SharedPtr

get_shared_dynamic_serialization_support() = 0;

/// Borrow a new serialized message (this clones!)

/** \return Shared pointer to a rclcpp::dynamic_typesupport::DynamicMessage. */

RCLCPP_PUBLIC

virtual

rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr

create_dynamic_message() = 0;

RCLCPP_PUBLIC

virtual

void

return_dynamic_message(rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message) = 0;

RCLCPP_PUBLIC

virtual

void

handle_dynamic_message(

const rclcpp::dynamic_typesupport::DynamicMessage::SharedPtr & message,

const rclcpp::MessageInfo & message_info) = 0;

RCLCPP_PUBLIC

bool

take_dynamic_message(

rclcpp::dynamic_typesupport::DynamicMessage & message_out,

rclcpp::MessageInfo & message_info_out);

// ===============================================================================================

template<typename EventCallbackT>

void

add_event_handler(

const EventCallbackT & callback,

const rcl_subscription_event_type_t event_type)

{

auto handler = std::make_shared<EventHandler<EventCallbackT,

std::shared_ptr<rcl_subscription_t>>>(

callback,

rcl_subscription_event_init,

get_subscription_handle(),

event_type);

qos_events_in_use_by_wait_set_.insert(std::make_pair(handler.get(), false));

event_handlers_.insert(std::make_pair(event_type, handler));

}

RCLCPP_PUBLIC

void default_incompatible_qos_callback(QOSRequestedIncompatibleQoSInfo & info) const;

RCLCPP_PUBLIC

void default_incompatible_type_callback(IncompatibleTypeInfo & info) const;

RCLCPP_PUBLIC

bool

matches_any_intra_process_publishers(const rmw_gid_t * sender_gid) const;

RCLCPP_PUBLIC

void

set_on_new_message_callback(rcl_event_callback_t callback, const void * user_data);

rclcpp::node_interfaces::NodeBaseInterface * const node_base_;

std::shared_ptr<rcl_node_t> node_handle_;

std::recursive_mutex callback_mutex_;

// It is important to declare on_new_message_callback_ before

// subscription_handle_, so on destruction the subscription is

// destroyed first. Otherwise, the rmw subscription callback

// would point briefly to a destroyed function.

std::function<void(size_t)> on_new_message_callback_{nullptr};

// Declare subscription_handle_ after callback

std::shared_ptr<rcl_subscription_t> subscription_handle_;

std::shared_ptr<rcl_subscription_t> intra_process_subscription_handle_;

rclcpp::Logger node_logger_;

std::unordered_map<rcl_subscription_event_type_t,

std::shared_ptr<rclcpp::EventHandlerBase>> event_handlers_;

bool use_intra_process_;

IntraProcessManagerWeakPtr weak_ipm_;

uint64_t intra_process_subscription_id_;

std::shared_ptr<rclcpp::experimental::SubscriptionIntraProcessBase> subscription_intra_process_;

const SubscriptionEventCallbacks event_callbacks_;

RCLCPP_DISABLE_COPY(SubscriptionBase)

rosidl_message_type_support_t type_support_;

DeliveredMessageKind delivered_message_kind_;

std::atomic<bool> subscription_in_use_by_wait_set_{false};

std::atomic<bool> intra_process_subscription_waitable_in_use_by_wait_set_{false};

std::unordered_map<rclcpp::EventHandlerBase *,

std::atomic<bool>> qos_events_in_use_by_wait_set_;