Rework channels

examples/unix/c11/z_get_channel.c

@@ -96,16 +96,17 @@ int main(int argc, char **argv) {
         z_bytes_serialize_from_string(&payload, value);
         opts.payload = &payload;
     }
-    z_owned_reply_ring_channel_t channel;
-    z_reply_ring_channel_new(&channel, 1);
-    if (z_get(z_loan(s), z_loan(ke), "", z_move(channel.send), &opts) < 0) {
+    z_owned_closure_reply_t closure;
+    z_owned_ring_handler_reply_t handler;
+    z_ring_channel_reply_new(&closure, &handler, 1);
+    if (z_get(z_loan(s), z_loan(ke), "", z_move(closure), &opts) < 0) {
         printf("Unable to send query.\n");
         return -1;
     }
 
     z_owned_reply_t reply;
     z_null(&reply);
-    for (z_call(channel.recv, &reply); z_check(reply); z_call(channel.recv, &reply)) {
+    for (z_recv(z_loan(handler), &reply); z_check(reply); z_recv(z_loan(handler), &reply)) {
         if (z_reply_is_ok(z_loan(reply))) {
             const z_loaned_sample_t *sample = z_reply_ok(z_loan(reply));
             z_owned_string_t keystr;
@@ -121,7 +122,7 @@ int main(int argc, char **argv) {
         }
     }
 
-    z_drop(z_move(channel));
+    z_drop(z_move(handler));
 
     // Stop read and lease tasks for zenoh-pico
     zp_stop_read_task(z_loan_mut(s));

examples/unix/c11/z_pull.c

@@ -73,12 +73,13 @@ int main(int argc, char **argv) {
     }
 
     printf("Declaring Subscriber on '%s'...\n", keyexpr);
-    z_owned_sample_ring_channel_t channel;
-    z_sample_ring_channel_new(&channel, size);
+    z_owned_closure_sample_t closure;
+    z_owned_ring_handler_sample_t handler;
+    z_ring_channel_sample_new(&closure, &handler, size);
     z_owned_subscriber_t sub;
     z_view_keyexpr_t ke;
     z_view_keyexpr_from_string(&ke, keyexpr);
-    if (z_declare_subscriber(&sub, z_loan(s), z_loan(ke), z_move(channel.send), NULL) < 0) {
+    if (z_declare_subscriber(&sub, z_loan(s), z_loan(ke), z_move(closure), NULL) < 0) {
         printf("Unable to declare subscriber.\n");
         return -1;
     }
@@ -87,7 +88,7 @@ int main(int argc, char **argv) {
     z_owned_sample_t sample;
     z_null(&sample);
     while (true) {
-        for (z_call(channel.try_recv, &sample); z_check(sample); z_call(channel.try_recv, &sample)) {
+        for (z_try_recv(z_loan(handler), &sample); z_check(sample); z_try_recv(z_loan(handler), &sample)) {
             z_owned_string_t keystr;
             z_keyexpr_to_string(z_sample_keyexpr(z_loan(sample)), &keystr);
             z_owned_string_t value;
@@ -103,7 +104,7 @@ int main(int argc, char **argv) {
     }
 
     z_undeclare_subscriber(z_move(sub));
-    z_drop(z_move(channel));
+    z_drop(z_move(handler));
 
     // Stop read and lease tasks for zenoh-pico
     zp_stop_read_task(z_loan_mut(s));

examples/unix/c11/z_queryable_channel.c

@@ -88,17 +88,18 @@ int main(int argc, char **argv) {
     }
 
     printf("Creating Queryable on '%s'...\n", keyexpr);
-    z_owned_query_ring_channel_t channel;
-    z_query_ring_channel_new(&channel, 10);
+    z_owned_closure_query_t closure;
+    z_owned_ring_handler_query_t handler;
+    z_ring_channel_query_new(&closure, &handler, 10);
     z_owned_queryable_t qable;
-    if (z_declare_queryable(&qable, z_loan(s), z_loan(ke), z_move(channel.send), NULL) < 0) {
+    if (z_declare_queryable(&qable, z_loan(s), z_loan(ke), z_move(closure), NULL) < 0) {
         printf("Unable to create queryable.\n");
         return -1;
     }
 
     z_owned_query_t query;
     z_null(&query);
-    for (z_call(channel.recv, &query); z_check(query); z_call(channel.recv, &query)) {
+    for (z_recv(z_loan(handler), &query); z_check(query); z_recv(z_loan(handler), &query)) {
         const z_loaned_query_t *q = z_loan(query);
         z_owned_string_t keystr;
         z_keyexpr_to_string(z_query_keyexpr(q), &keystr);
@@ -125,7 +126,7 @@ int main(int argc, char **argv) {
         z_drop(z_move(query));
     }
 
-    z_drop(z_move(channel));
+    z_drop(z_move(handler));
     z_undeclare_queryable(z_move(qable));
 
     // Stop read and lease tasks for zenoh-pico

examples/unix/c11/z_sub_channel.c

@@ -65,19 +65,21 @@ int main(int argc, char **argv) {
     }
 
     printf("Declaring Subscriber on '%s'...\n", keyexpr);
-    z_owned_sample_fifo_channel_t channel;
-    z_sample_fifo_channel_new(&channel, 3);
+    z_owned_closure_sample_t closure;
+    z_owned_fifo_handler_sample_t handler;
+    z_fifo_channel_sample_new(&closure, &handler, 3);
+
     z_owned_subscriber_t sub;
     z_view_keyexpr_t ke;
     z_view_keyexpr_from_string(&ke, keyexpr);
-    if (z_declare_subscriber(&sub, z_loan(s), z_loan(ke), z_move(channel.send), NULL) < 0) {
+    if (z_declare_subscriber(&sub, z_loan(s), z_loan(ke), z_move(closure), NULL) < 0) {
         printf("Unable to declare subscriber.\n");
         return -1;
     }
 
     z_owned_sample_t sample;
     z_null(&sample);
-    for (z_call(channel.recv, &sample); z_check(sample); z_call(channel.recv, &sample)) {
+    for (z_recv(z_loan(handler), &sample); z_check(sample); z_recv(z_loan(handler), &sample)) {
         z_owned_string_t keystr;
         z_keyexpr_to_string(z_sample_keyexpr(z_loan(sample)), &keystr);
         z_owned_string_t value;
@@ -90,7 +92,7 @@ int main(int argc, char **argv) {
     }
 
     z_undeclare_subscriber(z_move(sub));
-    z_drop(z_move(channel));
+    z_drop(z_move(handler));
 
     // Stop read and lease tasks for zenoh-pico
     zp_stop_read_task(z_loan_mut(s));