Skip to content

Event listeners distinct #109

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 12 commits into from
Mar 30, 2023
Merged

Event listeners distinct #109

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
73bc140
Add sequence numbers to message info structure (#74)
ivanpauno Mar 21, 2022
5fd0aab
add stub for content filtered topic (#77)
Mar 21, 2022
13dea22
Add support for user-specified content filters (#68)
asorbini Mar 22, 2022
3c8eb0e
0.10.0
paudrow Mar 28, 2022
8b70603
Update launch_testing_ros output filter prefixes for Connext6 (#80)
ivanpauno Apr 6, 2022
be74aa2
Properly initialize CDR stream before using it for filtering (#81)
asorbini Apr 7, 2022
d24d1ae
Exclude missing sample info fields when building rmw_connextddsmicro …
asorbini Apr 8, 2022
1b00f7b
0.11.0
paudrow Apr 9, 2022
5270943
Resolve build error with RTI Connext DDS 5.3.1 (#82)
asorbini Apr 18, 2022
84384b0
Changelog.
clalancette Apr 26, 2022
2aad9a6
0.11.1
clalancette Apr 26, 2022
77ca294
Use destinct callbacks for each event type
cwecht Mar 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 15 additions & 14 deletions rmw_connextdds_common/include/rmw_connextdds/rmw_waitset_std.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -350,43 +350,44 @@ class RMW_Connext_StatusCondition : public RMW_Connext_Condition
has_status(const rmw_event_type_t event_type) = 0;

void
notify_new_event()
notify_new_event(rmw_event_type_t event_type)
{
std::unique_lock<std::mutex> lock_mutex(new_event_mutex_);
if (new_event_cb_) {
new_event_cb_(user_data_, 1);
if (new_event_cb_[event_type]) {
new_event_cb_[event_type](user_data_[event_type], 1);
} else {
unread_events_count_++;
unread_events_count_[event_type]++;
}
}

void
set_new_event_callback(
rmw_event_type_t event_type,
rmw_event_callback_t callback,
const void * user_data)
{
std::unique_lock<std::mutex> lock_mutex(new_event_mutex_);

if (callback) {
// Push events arrived before setting the executor's callback
if (unread_events_count_ > 0) {
callback(user_data, unread_events_count_);
unread_events_count_ = 0;
if (unread_events_count_[event_type] > 0) {
callback(user_data, unread_events_count_[event_type]);
unread_events_count_[event_type] = 0;
}
user_data_ = user_data;
new_event_cb_ = callback;
user_data_[event_type] = user_data;
new_event_cb_[event_type] = callback;
} else {
user_data_ = nullptr;
new_event_cb_ = nullptr;
user_data_[event_type] = nullptr;
new_event_cb_[event_type] = nullptr;
}
}

protected:
DDS_StatusCondition * scond;
std::mutex new_event_mutex_;
rmw_event_callback_t new_event_cb_{nullptr};
const void * user_data_{nullptr};
uint64_t unread_events_count_ = 0;
rmw_event_callback_t new_event_cb_[RMW_EVENT_INVALID] = {};
const void * user_data_[RMW_EVENT_INVALID] = {};
uint64_t unread_events_count_[RMW_EVENT_INVALID] = {0};
Comment on lines +388 to +390
Copy link
Collaborator

@asorbini asorbini Mar 30, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The use of an enum value to define the size of an array is an unsual pattern which caught me by surprise.

It makes sense, but it's not commonly used, so I would add a comment to point it out.

Something like: "These arrays must be able to contain a value for each event kind. RMW_EVENT_INVALID is used because it is the last enumeration of rmw_event_type_t".

Copy link
Collaborator

@asorbini asorbini Mar 30, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will address the comment in #76

};

void
Expand Down
12 changes: 8 additions & 4 deletions rmw_connextdds_common/src/common/rmw_impl_waitset_std.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -708,7 +708,7 @@ RMW_Connext_SubscriberStatusCondition::update_status_deadline(
this->status_deadline.total_count_change = this->status_deadline.total_count;
this->status_deadline.total_count_change -= this->status_deadline_last.total_count;

this->notify_new_event();
this->notify_new_event(RMW_EVENT_REQUESTED_DEADLINE_MISSED);
}

void
Expand All @@ -724,7 +724,7 @@ RMW_Connext_SubscriberStatusCondition::update_status_liveliness(
this->status_liveliness.not_alive_count_change -=
this->status_liveliness_last.not_alive_count;

this->notify_new_event();
this->notify_new_event(RMW_EVENT_LIVELINESS_CHANGED);
}

void
Expand All @@ -736,6 +736,7 @@ RMW_Connext_SubscriberStatusCondition::update_status_qos(

this->status_qos.total_count_change = this->status_qos.total_count;
this->status_qos.total_count_change -= this->status_qos_last.total_count;
this->notify_new_event(RMW_EVENT_REQUESTED_QOS_INCOMPATIBLE);
}

void
Expand All @@ -748,6 +749,7 @@ RMW_Connext_SubscriberStatusCondition::update_status_sample_lost(
this->status_sample_lost.total_count_change = this->status_sample_lost.total_count;
this->status_sample_lost.total_count_change -=
this->status_sample_lost_last.total_count;
this->notify_new_event(RMW_EVENT_MESSAGE_LOST);
}

void
Expand Down Expand Up @@ -882,7 +884,7 @@ RMW_Connext_PublisherStatusCondition::update_status_deadline(
this->status_deadline.total_count_change = this->status_deadline.total_count;
this->status_deadline.total_count_change -= this->status_deadline_last.total_count;

this->notify_new_event();
this->notify_new_event(RMW_EVENT_OFFERED_DEADLINE_MISSED);
}

void
Expand All @@ -895,7 +897,7 @@ RMW_Connext_PublisherStatusCondition::update_status_liveliness(
this->status_liveliness.total_count_change = this->status_liveliness.total_count;
this->status_liveliness.total_count_change -= this->status_liveliness_last.total_count;

this->notify_new_event();
this->notify_new_event(RMW_EVENT_LIVELINESS_CHANGED);
}

void
Expand All @@ -907,4 +909,6 @@ RMW_Connext_PublisherStatusCondition::update_status_qos(

this->status_qos.total_count_change = this->status_qos.total_count;
this->status_qos.total_count_change -= this->status_qos_last.total_count;

this->notify_new_event(RMW_EVENT_OFFERED_QOS_INCOMPATIBLE);
}
2 changes: 1 addition & 1 deletion rmw_connextdds_common/src/common/rmw_listener.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ rmw_api_connextdds_event_set_callback(
} else {
condition = RMW_Connext_Event::publisher(event)->condition();
}
condition->set_new_event_callback(callback, user_data);
condition->set_new_event_callback(event->event_type, callback, user_data);
return RMW_RET_OK;
}

Expand Down