Add ringbuffer and handler for pull examples

CMakeLists.txt

@@ -331,6 +331,8 @@ 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_channels_test ${PROJECT_SOURCE_DIR}/tests/z_channels_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)
@@ -343,6 +345,8 @@ 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_channels_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})
@@ -361,6 +365,8 @@ if(UNIX OR MSVC)
 
     enable_testing()
     add_test(z_data_struct_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_data_struct_test)
+    add_test(z_channels_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_channels_test)
+    add_test(z_collections_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_collections_test)
     add_test(z_endpoint_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_endpoint_test)
     add_test(z_iobuf_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_iobuf_test)
     add_test(z_msgcodec_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_msgcodec_test)
@@ -475,4 +481,4 @@ if(PACKAGING)
   include(CPack)
 endif()
 
-endif()
+endif()

examples/CMakeLists.txt

@@ -37,6 +37,7 @@ if(UNIX)
         add_example(z_pub unix/c11/z_pub.c)
         add_example(z_pub_st unix/c11/z_pub_st.c)
         add_example(z_sub unix/c11/z_sub.c)
+        add_example(z_sub_channel unix/c11/z_sub_channel.c)
         add_example(z_sub_st unix/c11/z_sub_st.c)
         add_example(z_pull unix/c11/z_pull.c)
         add_example(z_get unix/c11/z_get.c)

examples/unix/c11/z_pull.c

@@ -19,30 +19,29 @@
 #include <zenoh-pico.h>
 
 #if Z_FEATURE_SUBSCRIPTION == 1
-// @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/**";
     char *locator = NULL;
+    size_t interval = 5000;
+    size_t size = 3;
 
     int opt;
-    while ((opt = getopt(argc, argv, "k:e:")) != -1) {
+    while ((opt = getopt(argc, argv, "k:e:i:s:")) != -1) {
         switch (opt) {
             case 'k':
                 keyexpr = optarg;
                 break;
             case 'e':
                 locator = optarg;
                 break;
+            case 'i':
+                interval = (size_t)atoi(optarg);
+                break;
+            case 's':
+                size = (size_t)atoi(optarg);
+                break;
             case '?':
-                if (optopt == 'k' || optopt == 'e') {
+                if (optopt == 'k' || optopt == 'e' || optopt == 'i' || optopt == 's') {
                     fprintf(stderr, "Option -%c requires an argument.\n", optopt);
                 } else {
                     fprintf(stderr, "Unknown option `-%c'.\n", optopt);
@@ -72,31 +71,30 @@ int main(int argc, char **argv) {
         return -1;
     }
 
-    // @TODO
-    // z_owned_closure_sample_t callback = z_closure(data_handler);
     printf("Declaring Subscriber on '%s'...\n", keyexpr);
+    z_owned_sample_ring_channel_t channel = z_sample_ring_channel_new(size);
+    z_owned_subscriber_t sub = z_declare_subscriber(z_loan(s), z_keyexpr(keyexpr), z_move(channel.send), NULL);
+    if (!z_check(sub)) {
+        printf("Unable to declare subscriber.\n");
+        return -1;
+    }
 
-    // @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));
-    // }
+    printf("Pulling data every %zu ms... Ring size: %zd\n", interval, size);
+    z_owned_sample_t sample = z_sample_null();
+    while (true) {
+        for (z_call(channel.recv, &sample); z_check(sample); z_call(channel.recv, &sample)) {
+            z_owned_str_t keystr = z_keyexpr_to_string(z_loan(sample).keyexpr);
+            printf(">> [Subscriber] Pulled ('%s': '%.*s')\n", z_loan(keystr), (int)z_loan(sample).payload.len,
+                   z_loan(sample).payload.start);
+            z_drop(z_move(keystr));
+            z_drop(z_move(sample));
+        }
+        printf(">> [Subscriber] Nothing to pull... sleep for %zu ms\n", interval);
+        zp_sleep_ms(interval);
+    }
 
-    // z_undeclare_pull_subscriber(z_move(sub));
-    printf("Pull Subscriber not supported... exiting\n");
+    z_undeclare_subscriber(z_move(sub));
+    z_drop(z_move(channel));
 
     // Stop read and lease tasks for zenoh-pico
     zp_stop_read_task(z_loan(s));

examples/unix/c11/z_sub_channel.c

@@ -0,0 +1,100 @@
+//
+// Copyright (c) 2022 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+
+#include <ctype.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <zenoh-pico.h>
+
+#if Z_FEATURE_SUBSCRIPTION == 1
+int main(int argc, char **argv) {
+    const char *keyexpr = "demo/example/**";
+    char *locator = NULL;
+
+    int opt;
+    while ((opt = getopt(argc, argv, "k:e:")) != -1) {
+        switch (opt) {
+            case 'k':
+                keyexpr = optarg;
+                break;
+            case 'e':
+                locator = optarg;
+                break;
+            case '?':
+                if (optopt == 'k' || optopt == 'e') {
+                    fprintf(stderr, "Option -%c requires an argument.\n", optopt);
+                } else {
+                    fprintf(stderr, "Unknown option `-%c'.\n", optopt);
+                }
+                return 1;
+            default:
+                return -1;
+        }
+    }
+
+    z_owned_config_t config = z_config_default();
+    if (locator != NULL) {
+        zp_config_insert(z_loan(config), Z_CONFIG_CONNECT_KEY, z_string_make(locator));
+    }
+
+    printf("Opening session...\n");
+    z_owned_session_t s = z_open(z_move(config));
+    if (!z_check(s)) {
+        printf("Unable to open session!\n");
+        return -1;
+    }
+
+    // Start read and lease tasks for zenoh-pico
+    if (zp_start_read_task(z_loan(s), NULL) < 0 || zp_start_lease_task(z_loan(s), NULL) < 0) {
+        printf("Unable to start read and lease tasks\n");
+        z_close(z_session_move(&s));
+        return -1;
+    }
+
+    printf("Declaring Subscriber on '%s'...\n", keyexpr);
+    z_owned_sample_fifo_channel_t channel = z_sample_fifo_channel_new(3);
+    z_owned_subscriber_t sub = z_declare_subscriber(z_loan(s), z_keyexpr(keyexpr), z_move(channel.send), NULL);
+    if (!z_check(sub)) {
+        printf("Unable to declare subscriber.\n");
+        return -1;
+    }
+
+    z_owned_sample_t sample = z_sample_null();
+    for (z_call(channel.recv, &sample); z_check(sample); z_call(channel.recv, &sample)) {
+        z_owned_str_t keystr = z_keyexpr_to_string(z_loan(sample).keyexpr);
+        printf(">> [Subscriber] Received ('%s': '%.*s')\n", z_loan(keystr), (int)z_loan(sample).payload.len,
+               z_loan(sample).payload.start);
+        z_drop(z_move(keystr));
+        z_drop(z_move(sample));
+        sample = z_sample_null();
+    }
+
+    z_undeclare_subscriber(z_move(sub));
+    z_drop(z_move(channel));
+
+    // Stop read and lease tasks for zenoh-pico
+    zp_stop_read_task(z_loan(s));
+    zp_stop_lease_task(z_loan(s));
+
+    z_close(z_move(s));
+
+    return 0;
+}
+#else
+int main(void) {
+    printf("ERROR: Zenoh pico was compiled without Z_FEATURE_SUBSCRIPTION but this example requires it.\n");
+    return -2;
+}
+#endif

include/zenoh-pico.h

@@ -17,11 +17,12 @@
 
 #define ZENOH_PICO "0.11.0.0"
 #define ZENOH_PICO_MAJOR 0
-#define ZENOH_PICO_MINOR 10
+#define ZENOH_PICO_MINOR 11
 #define ZENOH_PICO_PATCH 0
 #define ZENOH_PICO_TWEAK 0
 
 #include "zenoh-pico/api/constants.h"
+#include "zenoh-pico/api/handlers.h"
 #include "zenoh-pico/api/macros.h"
 #include "zenoh-pico/api/primitives.h"
 #include "zenoh-pico/api/types.h"

include/zenoh-pico/api/handlers.h

@@ -0,0 +1,88 @@
+//
+// Copyright (c) 2024 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+//
+#ifndef INCLUDE_ZENOH_PICO_API_HANDLERS_H
+#define INCLUDE_ZENOH_PICO_API_HANDLERS_H
+
+#include <stdint.h>
+
+#include "zenoh-pico/api/macros.h"
+#include "zenoh-pico/api/types.h"
+#include "zenoh-pico/collections/element.h"
+#include "zenoh-pico/collections/fifo_mt.h"
+#include "zenoh-pico/collections/ring_mt.h"
+#include "zenoh-pico/utils/logging.h"
+
+// -- Samples handler
+void _z_owned_sample_move(z_owned_sample_t *dst, const z_owned_sample_t *src);
+z_owned_sample_t *_z_sample_to_owned_ptr(const _z_sample_t *src);
+
+// -- Channel
+#define _Z_CHANNEL_DEFINE(name, send_closure_name, recv_closure_name, send_type, recv_type, collection_type,      \
+                          collection_new_f, collection_free_f, collection_push_f, collection_pull_f, elem_move_f, \
+                          elem_convert_f, elem_free_f)                                                            \
+    typedef struct {                                                                                              \
+        z_owned_##send_closure_name##_t send;                                                                     \
+        z_owned_##recv_closure_name##_t recv;                                                                     \
+        collection_type *collection;                                                                              \
+    } z_owned_##name##_t;                                                                                         \
+                                                                                                                  \
+    static inline void _z_##name##_elem_free(void **elem) {                                                       \
+        elem_free_f((recv_type *)*elem);                                                                          \
+        *elem = NULL;                                                                                             \
+    }                                                                                                             \
+    static inline void _z_##name##_elem_move(void *dst, const void *src) {                                        \
+        elem_move_f((recv_type *)dst, (const recv_type *)src);                                                    \
+    }                                                                                                             \
+    static inline void _z_##name##_send(const send_type *elem, void *context) {                                   \
+        void *internal_elem = elem_convert_f(elem);                                                               \
+        if (internal_elem == NULL) {                                                                              \
+            return;                                                                                               \
+        }                                                                                                         \
+        int8_t ret = collection_push_f(internal_elem, context, _z_##name##_elem_free);                            \
+        if (ret != _Z_RES_OK) {                                                                                   \
+            _Z_ERROR("%s failed: %i", #collection_push_f, ret);                                                   \
+        }                                                                                                         \
+    }                                                                                                             \
+    static inline void _z_##name##_recv(recv_type *elem, void *context) {                                         \
+        int8_t ret = collection_pull_f(elem, context, _z_##name##_elem_move);                                     \
+        if (ret != _Z_RES_OK) {                                                                                   \
+            _Z_ERROR("%s failed: %i", #collection_pull_f, ret);                                                   \
+        }                                                                                                         \
+    }                                                                                                             \
+                                                                                                                  \
+    static inline z_owned_##name##_t z_##name##_new(size_t capacity) {                                            \
+        z_owned_##name##_t channel;                                                                               \
+        channel.collection = collection_new_f(capacity);                                                          \
+        channel.send = z_##send_closure_name(_z_##name##_send, NULL, channel.collection);                         \
+        channel.recv = z_##recv_closure_name(_z_##name##_recv, NULL, channel.collection);                         \
+        return channel;                                                                                           \
+    }                                                                                                             \
+    static inline z_owned_##name##_t *z_##name##_move(z_owned_##name##_t *val) { return val; }                    \
+    static inline void z_##name##_drop(z_owned_##name##_t *channel) {                                             \
+        collection_free_f(channel->collection, _z_##name##_elem_free);                                            \
+        z_##send_closure_name##_drop(&channel->send);                                                             \
+        z_##recv_closure_name##_drop(&channel->recv);                                                             \
+    }
+
+// z_owned_sample_ring_channel_t
+_Z_CHANNEL_DEFINE(sample_ring_channel, closure_sample, closure_owned_sample, z_sample_t, z_owned_sample_t, _z_ring_mt_t,
+                  _z_ring_mt_new, _z_ring_mt_free, _z_ring_mt_push, _z_ring_mt_pull, _z_owned_sample_move,
+                  _z_sample_to_owned_ptr, z_sample_drop)
+
+// z_owned_sample_fifo_channel_t
+_Z_CHANNEL_DEFINE(sample_fifo_channel, closure_sample, closure_owned_sample, z_sample_t, z_owned_sample_t, _z_fifo_mt_t,
+                  _z_fifo_mt_new, _z_fifo_mt_free, _z_fifo_mt_push, _z_fifo_mt_pull, _z_owned_sample_move,
+                  _z_sample_to_owned_ptr, z_sample_drop)
+
+#endif  // INCLUDE_ZENOH_PICO_API_HANDLERS_H