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

Add ringbuffer and handler for pull examples #377

Merged
merged 24 commits into from
Apr 8, 2024
Merged

Add ringbuffer and handler for pull examples #377

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
aceca42
Remove ACK message
Mallets Mar 19, 2024
61683e4
Cleanup pull and subscriber
Mallets Mar 19, 2024
6125424
Remove Put/Del from ResponseBody
Mallets Mar 19, 2024
e781e84
Cleanup flags
Mallets Mar 19, 2024
557e665
Comment unusued variables in pull examples
Mallets Mar 19, 2024
5bca588
Merge branch 'protocol_changes' into protocol_cleanup
Mallets Mar 19, 2024
235226f
Fix alignment test
Mallets Mar 19, 2024
be9bf0e
Add ring to collections
Mallets Mar 19, 2024
d2bc7d3
Merge branch 'protocol_changes' into protocol_pull
Mallets Mar 20, 2024
bff0356
Pull examples. Fake z_owned_sample_t
Mallets Mar 20, 2024
ede9911
Channels initial commit
Mallets Mar 21, 2024
defb3ac
Added channel macros
Mallets Mar 21, 2024
3d68a8c
Rename sample_ring to sample_channel_ring
Mallets Mar 21, 2024
2e13913
Add fifo and lifo collections
Mallets Mar 21, 2024
25b0661
Moved utils to handlers. Added FIFO handlers. Add z_sub_channel.
Mallets Mar 22, 2024
0d76f92
Improve pull example
Mallets Mar 22, 2024
b5546d5
Fix return in void functions
Mallets Mar 23, 2024
c736568
Fix prototype
Mallets Mar 23, 2024
2e24b29
Fix newline at end of the file
Mallets Mar 23, 2024
9171695
Fix windows examples target
Mallets Mar 23, 2024
c2d6d27
Remove unexisting windows z_sub_channel from CMAke
Mallets Mar 23, 2024
84f22e6
Remove unexisting unix c99 z_sub_channel from CMAke
Mallets Mar 23, 2024
3559b3e
Rework owned sample (#391)
sashacmc Apr 5, 2024
2c009d0
Merge branch 'protocol_changes' into protocol_pull
Mallets Apr 5, 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 CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -331,6 +331,7 @@ if(UNIX OR MSVC)
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/tests")

add_executable(z_data_struct_test ${PROJECT_SOURCE_DIR}/tests/z_data_struct_test.c)
add_executable(z_collections_test ${PROJECT_SOURCE_DIR}/tests/z_collections_test.c)
add_executable(z_endpoint_test ${PROJECT_SOURCE_DIR}/tests/z_endpoint_test.c)
add_executable(z_iobuf_test ${PROJECT_SOURCE_DIR}/tests/z_iobuf_test.c)
add_executable(z_msgcodec_test ${PROJECT_SOURCE_DIR}/tests/z_msgcodec_test.c)
Expand All @@ -343,6 +344,7 @@ if(UNIX OR MSVC)
add_executable(z_perf_rx ${PROJECT_SOURCE_DIR}/tests/z_perf_rx.c)

target_link_libraries(z_data_struct_test ${Libname})
target_link_libraries(z_collections_test ${Libname})
target_link_libraries(z_endpoint_test ${Libname})
target_link_libraries(z_iobuf_test ${Libname})
target_link_libraries(z_msgcodec_test ${Libname})
Expand Down
49 changes: 27 additions & 22 deletions examples/arduino/z_pull.ino
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,20 +34,22 @@

#define KEYEXPR "demo/example/**"

z_owned_pull_subscriber_t sub;
// @TODO
// z_owned_pull_subscriber_t sub;

void data_handler(const z_sample_t *sample, void *arg) {
z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
std::string val((const char *)sample->payload.start, sample->payload.len);
// @TODO
// void data_handler(const z_sample_t *sample, void *arg) {
// z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
// std::string val((const char *)sample->payload.start, sample->payload.len);

Serial.print(" >> [Subscription listener] Received (");
Serial.print(z_str_loan(&keystr));
Serial.print(", ");
Serial.print(val.c_str());
Serial.println(")");
// Serial.print(" >> [Subscription listener] Received (");
// Serial.print(z_str_loan(&keystr));
// Serial.print(", ");
// Serial.print(val.c_str());
// Serial.println(")");

z_str_drop(z_str_move(&keystr));
}
// z_str_drop(z_str_move(&keystr));
// }

void setup() {
// Initialize Serial for debug
Expand Down Expand Up @@ -91,23 +93,26 @@ void setup() {
Serial.print("Declaring Subscriber on ");
Serial.print(KEYEXPR);
Serial.println(" ...");
z_owned_closure_sample_t callback = z_closure_sample(data_handler, NULL, NULL);
sub = z_declare_pull_subscriber(z_session_loan(&s), z_keyexpr(KEYEXPR), z_closure_sample_move(&callback), NULL);
if (!z_pull_subscriber_check(&sub)) {
Serial.println("Unable to declare subscriber.");
while (1) {
;
}
}
Serial.println("OK");
Serial.println("Zenoh setup finished!");
// @TODO
// z_owned_closure_sample_t callback = z_closure_sample(data_handler, NULL, NULL);
// @TODO
// sub = z_declare_pull_subscriber(z_session_loan(&s), z_keyexpr(KEYEXPR), z_closure_sample_move(&callback), NULL);
// if (!z_pull_subscriber_check(&sub)) {
// Serial.println("Unable to declare subscriber.");
// while (1) {
// ;
// }
// }
// Serial.println("OK");
// Serial.println("Zenoh setup finished!");
Serial.println("Pull Subscriber not supported... exiting");

delay(300);
}

void loop() {
delay(5000);
z_subscriber_pull(z_pull_subscriber_loan(&sub));
// z_subscriber_pull(z_pull_subscriber_loan(&sub));
}
#else
void setup() {
Expand Down
48 changes: 26 additions & 22 deletions examples/espidf/z_pull.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,12 +100,13 @@ void wifi_init_sta(void) {
vEventGroupDelete(s_event_group_handler);
}

void data_handler(const z_sample_t* sample, void* arg) {
z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
printf(" >> [Subscriber handler] Received ('%s': '%.*s')\n", z_loan(keystr), (int)sample->payload.len,
sample->payload.start);
z_drop(z_move(keystr));
}
// @TODO
// void data_handler(const z_sample_t* sample, void* arg) {
// z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
// printf(" >> [Subscriber handler] Received ('%s': '%.*s')\n", z_loan(keystr), (int)sample->payload.len,
// sample->payload.start);
// z_drop(z_move(keystr));
// }

void app_main() {
esp_err_t ret = nvs_flash_init();
Expand Down Expand Up @@ -144,23 +145,26 @@ void app_main() {
zp_start_read_task(z_loan(s), NULL);
zp_start_lease_task(z_loan(s), NULL);

// @TODO
// z_owned_closure_sample_t callback = z_closure(data_handler);
printf("Declaring Subscriber on '%s'...", KEYEXPR);
z_owned_closure_sample_t callback = z_closure(data_handler);
z_owned_pull_subscriber_t sub = z_declare_pull_subscriber(z_loan(s), z_keyexpr(KEYEXPR), z_move(callback), NULL);
if (!z_check(sub)) {
printf("Unable to declare subscriber.\n");
exit(-1);
}
printf("OK!\n");

while (1) {
sleep(5);
printf("Pulling data from '%s'...\n", KEYEXPR);
z_subscriber_pull(z_loan(sub));
}

printf("Closing Zenoh Session...");
z_undeclare_pull_subscriber(z_move(sub));
// @TODO
// z_owned_pull_subscriber_t sub = z_declare_pull_subscriber(z_loan(s), z_keyexpr(KEYEXPR), z_move(callback), NULL);
// if (!z_check(sub)) {
// printf("Unable to declare subscriber.\n");
// exit(-1);
// }
// printf("OK!\n");

// while (1) {
// sleep(5);
// printf("Pulling data from '%s'...\n", KEYEXPR);
// z_subscriber_pull(z_loan(sub));
// }

// printf("Closing Zenoh Session...");
// z_undeclare_pull_subscriber(z_move(sub));
printf("Pull Subscriber not supported... exiting\n");

// Stop the receive and the session lease loop for zenoh-pico
zp_stop_read_task(z_loan(s));
Expand Down
42 changes: 23 additions & 19 deletions examples/freertos_plus_tcp/z_pull.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,14 @@

#define KEYEXPR "demo/example/**"

void data_handler(const z_sample_t *sample, void *ctx) {
(void)(ctx);
z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
printf(">> [Subscriber] Received ('%s': '%.*s')\n", z_loan(keystr), (int)sample->payload.len,
sample->payload.start);
z_drop(z_move(keystr));
}
// @TODO
// void data_handler(const z_sample_t *sample, void *ctx) {
// (void)(ctx);
// z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
// printf(">> [Subscriber] Received ('%s': '%.*s')\n", z_loan(keystr), (int)sample->payload.len,
// sample->payload.start);
// z_drop(z_move(keystr));
// }

void app_main(void) {
z_owned_config_t config = z_config_default();
Expand All @@ -57,21 +58,24 @@
return;
}

z_owned_closure_sample_t callback = z_closure(data_handler);
// @TODO
// z_owned_closure_sample_t callback = z_closure(data_handler);
printf("Declaring Subscriber on '%s'...\n", KEYEXPR);
z_owned_pull_subscriber_t sub = z_declare_pull_subscriber(z_loan(s), z_keyexpr(KEYEXPR), z_move(callback), NULL);
if (!z_check(sub)) {
printf("Unable to declare subscriber.\n");
return;
}
// @TODO
// z_owned_pull_subscriber_t sub = z_declare_pull_subscriber(z_loan(s), z_keyexpr(KEYEXPR), z_move(callback), NULL);
// if (!z_check(sub)) {
// printf("Unable to declare subscriber.\n");
// return;
// }

while (1) {
zp_sleep_s(5);
printf("Pulling data from '%s'...\n", KEYEXPR);
z_subscriber_pull(z_loan(sub));
}
// while (1) {
// zp_sleep_s(5);
// printf("Pulling data from '%s'...\n", KEYEXPR);
// z_subscriber_pull(z_loan(sub));
// }

z_undeclare_pull_subscriber(z_move(sub));
// z_undeclare_pull_subscriber(z_move(sub));
printf("Pull Subscriber not supported... exiting\n");

Check notice

Code scanning / Cppcheck (reported by Codacy)

MISRA 17.7 rule Note

MISRA 17.7 rule

// Stop read and lease tasks for zenoh-pico
zp_stop_read_task(z_loan(s));
Expand Down
46 changes: 25 additions & 21 deletions examples/mbed/z_pull.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,12 +30,13 @@

#define KEYEXPR "demo/example/**"

void data_handler(const z_sample_t *sample, void *arg) {
z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
printf(" >> [Subscriber handler] Received ('%s': '%.*s')\n", z_str_loan(&keystr), (int)sample->payload.len,
sample->payload.start);
z_str_drop(z_str_move(&keystr));
}
// @TODO
// void data_handler(const z_sample_t *sample, void *arg) {
// z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
// printf(" >> [Subscriber handler] Received ('%s': '%.*s')\n", z_str_loan(&keystr), (int)sample->payload.len,
// sample->payload.start);
// z_str_drop(z_str_move(&keystr));
// }

int main(int argc, char **argv) {
randLIB_seed_random();
Expand Down Expand Up @@ -64,24 +65,27 @@ int main(int argc, char **argv) {
zp_start_read_task(z_session_loan(&s), NULL);
zp_start_lease_task(z_session_loan(&s), NULL);

// @TODO
// z_owned_closure_sample_t callback = z_closure_sample(data_handler, NULL, NULL);
printf("Declaring Subscriber on '%s'...", KEYEXPR);
z_owned_closure_sample_t callback = z_closure_sample(data_handler, NULL, NULL);
z_owned_pull_subscriber_t sub =
z_declare_pull_subscriber(z_session_loan(&s), z_keyexpr(KEYEXPR), z_closure_sample_move(&callback), NULL);
if (!z_pull_subscriber_check(&sub)) {
printf("Unable to declare subscriber.\n");
exit(-1);
}
printf("OK!\n");
// @TODO
// z_owned_pull_subscriber_t sub =
// z_declare_pull_subscriber(z_session_loan(&s), z_keyexpr(KEYEXPR), z_closure_sample_move(&callback), NULL);
// if (!z_pull_subscriber_check(&sub)) {
// printf("Unable to declare subscriber.\n");
// exit(-1);
// }
// printf("OK!\n");

while (1) {
zp_sleep_s(5);
printf("Pulling data from '%s'...\n", KEYEXPR);
z_subscriber_pull(z_pull_subscriber_loan(&sub));
}
// while (1) {
// zp_sleep_s(5);
// printf("Pulling data from '%s'...\n", KEYEXPR);
// z_subscriber_pull(z_pull_subscriber_loan(&sub));
// }

printf("Closing Zenoh Session...");
z_undeclare_pull_subscriber(z_pull_subscriber_move(&sub));
// printf("Closing Zenoh Session...");
// z_undeclare_pull_subscriber(z_pull_subscriber_move(&sub));
printf("Pull Subscriber not supported... exiting\n");

// Stop the receive and the session lease loop for zenoh-pico
zp_stop_read_task(z_session_loan(&s));
Expand Down
2 changes: 1 addition & 1 deletion examples/unix/c11/z_get.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@
return -1;
}

printf("Enter any key to pull data or 'q' to quit...\n");
printf("Enter any key to get data or 'q' to quit...\n");

Check notice

Code scanning / Cppcheck (reported by Codacy)

MISRA 17.7 rule Note

MISRA 17.7 rule
char c = '\0';
while (1) {
fflush(stdin);
Expand Down
55 changes: 30 additions & 25 deletions examples/unix/c11/z_pull.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,14 @@
#include <zenoh-pico.h>

#if Z_FEATURE_SUBSCRIPTION == 1
void data_handler(const z_sample_t *sample, void *ctx) {
(void)(ctx);
z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
printf(">> [Subscriber] Received ('%s': '%.*s')\n", z_loan(keystr), (int)sample->payload.len,
sample->payload.start);
z_drop(z_move(keystr));
}
// @TODO
// void data_handler(const z_sample_t *sample, void *ctx) {
// (void)(ctx);
// z_owned_str_t keystr = z_keyexpr_to_string(sample->keyexpr);
// printf(">> [Subscriber] Received ('%s': '%.*s')\n", z_loan(keystr), (int)sample->payload.len,
// sample->payload.start);
// z_drop(z_move(keystr));
// }

int main(int argc, char **argv) {
const char *keyexpr = "demo/example/**";
Expand Down Expand Up @@ -71,27 +72,31 @@
return -1;
}

z_owned_closure_sample_t callback = z_closure(data_handler);
// @TODO
// z_owned_closure_sample_t callback = z_closure(data_handler);
printf("Declaring Subscriber on '%s'...\n", keyexpr);
z_owned_pull_subscriber_t sub = z_declare_pull_subscriber(z_loan(s), z_keyexpr(keyexpr), z_move(callback), NULL);
if (!z_check(sub)) {
printf("Unable to declare subscriber.\n");
return -1;
}

printf("Enter any key to pull data or 'q' to quit...\n");
char c = '\0';
while (1) {
fflush(stdin);
int ret = scanf("%c", &c);
(void)ret; // Remove unused result warning
if (c == 'q') {
break;
}
z_subscriber_pull(z_loan(sub));
}
// @TODO
// z_owned_pull_subscriber_t sub = z_declare_pull_subscriber(z_loan(s), z_keyexpr(keyexpr), z_move(callback), NULL);
// if (!z_check(sub)) {
// printf("Unable to declare subscriber.\n");
// return -1;
// }

// printf("Enter any key to pull data or 'q' to quit...\n");
// char c = '\0';
// while (1) {
// fflush(stdin);
// int ret = scanf("%c", &c);
// (void)ret; // Remove unused result warning
// if (c == 'q') {
// break;
// }
// z_subscriber_pull(z_loan(sub));
// }

z_undeclare_pull_subscriber(z_move(sub));
// z_undeclare_pull_subscriber(z_move(sub));
printf("Pull Subscriber not supported... exiting\n");

Check notice

Code scanning / Cppcheck (reported by Codacy)

MISRA 17.7 rule Note

MISRA 17.7 rule

// Stop read and lease tasks for zenoh-pico
zp_stop_read_task(z_loan(s));
Expand Down
2 changes: 1 addition & 1 deletion examples/unix/c99/z_get.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ int main(int argc, char **argv) {
return -1;
}

printf("Enter any key to pull data or 'q' to quit...\n");
printf("Enter any key to get data or 'q' to quit...\n");
char c = '\0';
while (1) {
fflush(stdin);
Expand Down
Loading
Loading