Skip to content
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

Local put will trigger local subscriptions #316

Merged
merged 3 commits into from
Jan 10, 2024
Merged

Local put will trigger local subscriptions #316

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
896b7dd
feat: trigger local subscriptions when performing a pub/put
jean-roland Jan 10, 2024
f365b8c
feat: add dummy function when feature not present
jean-roland Jan 10, 2024
b4e2a2e
fix: missing end of file new line
jean-roland Jan 10, 2024
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
2 changes: 2 additions & 0 deletions include/zenoh-pico/session/subscription.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ _z_subscription_sptr_list_t *_z_get_subscriptions_by_key(_z_session_t *zn, uint8
const _z_keyexpr_t *keyexpr);

_z_subscription_sptr_t *_z_register_subscription(_z_session_t *zn, uint8_t is_local, _z_subscription_t *sub);
void _z_trigger_local_subscriptions(_z_session_t *zn, const _z_keyexpr_t keyexpr, const uint8_t *payload,
_z_zint_t payload_len);
int8_t _z_trigger_subscriptions(_z_session_t *zn, const _z_keyexpr_t keyexpr, const _z_bytes_t payload,
const _z_encoding_t encoding, const _z_zint_t kind, const _z_timestamp_t timestamp);
void _z_unregister_subscription(_z_session_t *zn, uint8_t is_local, _z_subscription_sptr_t *sub);
Expand Down
7 changes: 7 additions & 0 deletions src/api/api.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include "zenoh-pico/protocol/keyexpr.h"
#include "zenoh-pico/session/queryable.h"
#include "zenoh-pico/session/resource.h"
#include "zenoh-pico/session/subscription.h"
#include "zenoh-pico/session/utils.h"
#include "zenoh-pico/system/platform.h"
#include "zenoh-pico/transport/multicast.h"
Expand Down Expand Up @@ -612,6 +613,9 @@ int8_t z_put(z_session_t zs, z_keyexpr_t keyexpr, const uint8_t *payload, z_zint
ret = _z_write(zs._val, keyexpr, (const uint8_t *)payload, payload_len, opt.encoding, Z_SAMPLE_KIND_PUT,
opt.congestion_control, opt.priority);

// Trigger local subscriptions
_z_trigger_local_subscriptions(zs._val, keyexpr, payload, payload_len);

return ret;
}

Expand Down Expand Up @@ -685,6 +689,9 @@ int8_t z_publisher_put(const z_publisher_t pub, const uint8_t *payload, size_t l
ret = _z_write(pub._val->_zn, pub._val->_key, payload, len, opt.encoding, Z_SAMPLE_KIND_PUT,
pub._val->_congestion_control, pub._val->_priority);

// Trigger local subscriptions
_z_trigger_local_subscriptions(pub._val->_zn, pub._val->_key, payload, len);

return ret;
}

Expand Down
1 change: 0 additions & 1 deletion src/net/primitives.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,6 @@ _z_subscriber_t *_z_declare_subscriber(_z_session_t *zn, _z_keyexpr_t keyexpr, _
&keyexpr, s._id, sub_info.reliability == Z_RELIABILITY_RELIABLE, sub_info.mode == Z_SUBMODE_PULL);
_z_network_message_t n_msg = _z_n_msg_make_declare(declaration);
if (_z_send_n_msg(zn, &n_msg, Z_RELIABILITY_RELIABLE, Z_CONGESTION_CONTROL_BLOCK) != _Z_RES_OK) {
// ret = _Z_ERR_TRANSPORT_TX_FAILED;
_z_unregister_subscription(zn, _Z_RESOURCE_IS_LOCAL, sp_s);
_z_subscriber_free(&ret);
}
Expand Down
20 changes: 19 additions & 1 deletion src/session/subscription.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,14 @@
return ret;
}

void _z_trigger_local_subscriptions(_z_session_t *zn, const _z_keyexpr_t keyexpr, const uint8_t *payload,

Check warning

Code scanning / Cppcheck (reported by Codacy)

misra violation 804 with no text in the supplied rule-texts-file Warning

misra violation 804 with no text in the supplied rule-texts-file
_z_zint_t payload_len) {
_z_encoding_t encoding = {.prefix = Z_ENCODING_PREFIX_DEFAULT, .suffix = _z_bytes_wrap(NULL, 0)};
int8_t ret = _z_trigger_subscriptions(zn, keyexpr, _z_bytes_wrap(payload, payload_len), encoding, Z_SAMPLE_KIND_PUT,
_z_timestamp_null());
(void)ret;
}

int8_t _z_trigger_subscriptions(_z_session_t *zn, const _z_keyexpr_t keyexpr, const _z_bytes_t payload,
const _z_encoding_t encoding, const _z_zint_t kind, const _z_timestamp_t timestamp) {
int8_t ret = _Z_RES_OK;
Expand Down Expand Up @@ -227,4 +235,14 @@
_z_mutex_unlock(&zn->_mutex_inner);
#endif // Z_FEATURE_MULTI_THREAD == 1
}
#endif
#else // Z_FEATURE_SUBSCRIPTION == 0

void _z_trigger_local_subscriptions(_z_session_t *zn, const _z_keyexpr_t keyexpr, const uint8_t *payload,

Check warning

Code scanning / Cppcheck (reported by Codacy)

misra violation 804 with no text in the supplied rule-texts-file Warning

misra violation 804 with no text in the supplied rule-texts-file
_z_zint_t payload_len) {
_ZP_UNUSED(zn);
_ZP_UNUSED(keyexpr);
_ZP_UNUSED(payload);
_ZP_UNUSED(payload_len);
}

#endif // Z_FEATURE_SUBSCRIPTION == 1
Loading