Add Advanced Pub/Sub feature

CMakeLists.txt

@@ -241,6 +241,7 @@ set(Z_TRANSPORT_LEASE 10000 CACHE STRING "Link lease duration in milliseconds to
 
 set(Z_FEATURE_UNSTABLE_API 0 CACHE STRING "Toggle unstable Zenoh-C API")
 set(Z_FEATURE_PUBLICATION 1 CACHE STRING "Toggle publication feature")
+set(Z_FEATURE_ADVANCED_PUBLICATION 1 CACHE STRING "Toggle advanced publication feature")
 set(Z_FEATURE_SUBSCRIPTION 1 CACHE STRING "Toggle subscription feature")
 set(Z_FEATURE_QUERY 1 CACHE STRING "Toggle query feature")
 set(Z_FEATURE_QUERYABLE 1 CACHE STRING "Toggle queryable feature")
@@ -291,10 +292,16 @@ if(Z_FEATURE_MATCHING AND NOT Z_FEATURE_UNSTABLE_API)
   set(Z_FEATURE_MATCHING 0 CACHE STRING "Toggle matching feature" FORCE)
 endif()
 
+if(Z_FEATURE_ADVANCED_PUBLICATION AND NOT Z_FEATURE_UNSTABLE_API AND NOT Z_FEATURE_PUBLICATION AND NOT Z_FEATURE_LIVELINESS)
+  message(WARNING "Z_FEATURE_ADVANCED_PUBLICATION can only be enabled when Z_FEATURE_UNSTABLE_API, Z_FEATURE_PUBLICATION and Z_FEATURE_LIVELINESS is also enabled. Disabling Z_FEATURE_ADVANCED_PUBLICATION.")
+  set(Z_FEATURE_ADVANCED_PUBLICATION 0 CACHE STRING "Toggle advanced publication feature" FORCE)
+endif()
+
 message(STATUS "Building with feature config:\n\
 * UNSTABLE_API: ${Z_FEATURE_UNSTABLE_API}\n\
 * MULTI-THREAD: ${Z_FEATURE_MULTI_THREAD}\n\
 * PUBLICATION: ${Z_FEATURE_PUBLICATION}\n\
+* ADVANCED PUBLICATION: ${Z_FEATURE_ADVANCED_PUBLICATION}\n\
 * SUBSCRIPTION: ${Z_FEATURE_SUBSCRIPTION}\n\
 * QUERY: ${Z_FEATURE_QUERY}\n\
 * QUERYABLE: ${Z_FEATURE_QUERYABLE}\n\
@@ -549,6 +556,7 @@ if(UNIX OR MSVC)
     add_executable(z_api_encoding_test ${PROJECT_SOURCE_DIR}/tests/z_api_encoding_test.c)
     add_executable(z_refcount_test ${PROJECT_SOURCE_DIR}/tests/z_refcount_test.c)
     add_executable(z_lru_cache_test ${PROJECT_SOURCE_DIR}/tests/z_lru_cache_test.c)
+    add_executable(z_utils_test ${PROJECT_SOURCE_DIR}/tests/z_utils_test.c)
     add_executable(z_test_unicast ${PROJECT_SOURCE_DIR}/tests/z_test_unicast.c)
 
     target_link_libraries(z_data_struct_test zenohpico::lib)
@@ -569,6 +577,7 @@ if(UNIX OR MSVC)
     target_link_libraries(z_api_encoding_test zenohpico::lib)
     target_link_libraries(z_refcount_test zenohpico::lib)
     target_link_libraries(z_lru_cache_test zenohpico::lib)
+    target_link_libraries(z_utils_test zenohpico::lib)
     target_link_libraries(z_test_unicast zenohpico::lib)
 
     configure_file(${PROJECT_SOURCE_DIR}/tests/modularity.py ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/modularity.py COPYONLY)
@@ -595,6 +604,7 @@ if(UNIX OR MSVC)
     add_test(z_api_encoding_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_api_encoding_test)
     add_test(z_refcount_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_refcount_test)
     add_test(z_lru_cache_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_lru_cache_test)
+    add_test(z_utils_test ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/z_utils_test)
   endif()
 
   if(BUILD_MULTICAST)

GNUmakefile

@@ -59,6 +59,7 @@ ZENOH_LOG?=""
 Z_FEATURE_UNSTABLE_API?=0
 Z_FEATURE_MULTI_THREAD?=1
 Z_FEATURE_PUBLICATION?=1
+Z_FEATURE_ADVANCED_PUBLICATION?=0
 Z_FEATURE_SUBSCRIPTION?=1
 Z_FEATURE_QUERY?=1
 Z_FEATURE_QUERYABLE?=1
@@ -88,7 +89,7 @@ CROSSIMG_PREFIX=zenoh-pico_
 
 CMAKE_OPT=-DZENOH_DEBUG=$(ZENOH_DEBUG) -DZENOH_LOG=$(ZENOH_LOG) -DZENOH_LOG_PRINT=$(ZENOH_LOG_PRINT) -DBUILD_EXAMPLES=$(BUILD_EXAMPLES) -DCMAKE_BUILD_TYPE=$(BUILD_TYPE) -DBUILD_TESTING=$(BUILD_TESTING) -DBUILD_MULTICAST=$(BUILD_MULTICAST)\
  -DZ_FEATURE_MULTI_THREAD=$(Z_FEATURE_MULTI_THREAD) -DZ_FEATURE_INTEREST=$(Z_FEATURE_INTEREST) -DZ_FEATURE_UNSTABLE_API=$(Z_FEATURE_UNSTABLE_API)\
- -DZ_FEATURE_PUBLICATION=$(Z_FEATURE_PUBLICATION) -DZ_FEATURE_SUBSCRIPTION=$(Z_FEATURE_SUBSCRIPTION) -DZ_FEATURE_QUERY=$(Z_FEATURE_QUERY) -DZ_FEATURE_QUERYABLE=$(Z_FEATURE_QUERYABLE)\
+ -DZ_FEATURE_PUBLICATION=$(Z_FEATURE_PUBLICATION) -DZ_FEATURE_ADVANCED_PUBLICATION=$(Z_FEATURE_ADVANCED_PUBLICATION) -DZ_FEATURE_SUBSCRIPTION=$(Z_FEATURE_SUBSCRIPTION) -DZ_FEATURE_QUERY=$(Z_FEATURE_QUERY) -DZ_FEATURE_QUERYABLE=$(Z_FEATURE_QUERYABLE)\
  -DZ_FEATURE_RAWETH_TRANSPORT=$(Z_FEATURE_RAWETH_TRANSPORT) -DZ_FEATURE_LOCAL_SUBSCRIBER=$(Z_FEATURE_LOCAL_SUBSCRIBER) -DFRAG_MAX_SIZE=$(FRAG_MAX_SIZE) -DBATCH_UNICAST_SIZE=$(BATCH_UNICAST_SIZE)\
  -DBATCH_MULTICAST_SIZE=$(BATCH_MULTICAST_SIZE) -DZ_FEATURE_UNICAST_PEER=$(Z_FEATURE_UNICAST_PEER) -DASAN=$(ASAN) -DBUILD_INTEGRATION=$(BUILD_INTEGRATION) -DBUILD_TOOLS=$(BUILD_TOOLS) -DBUILD_SHARED_LIBS=$(BUILD_SHARED_LIBS) -H.
 

PackageConfig.cmake.in

@@ -3,6 +3,7 @@ include(CMakeFindDependencyMacro)
 set(ZENOHPICO_FEATURE_UNSTABLE_API @Z_FEATURE_UNSTABLE_API@)
 set(ZENOHPICO_FEATURE_MULTI_THREAD @Z_FEATURE_MULTI_THREAD@)
 set(ZENOHPICO_FEATURE_PUBLICATION @Z_FEATURE_PUBLICATION@)
+set(ZENOHPICO_FEATURE_ADVANCED_PUBLICATION @Z_FEATURE_ADVANCED_PUBLICATION@)
 set(ZENOHPICO_FEATURE_SUBSCRIPTION @Z_FEATURE_SUBSCRIPTION@)
 set(ZENOHPICO_FEATURE_QUERY @Z_FEATURE_QUERY@)
 set(ZENOHPICO_FEATURE_QUERYABLE @Z_FEATURE_QUERYABLE@)

include/zenoh-pico/collections/ring.h

@@ -32,6 +32,17 @@ typedef struct {
     size_t _w_idx;
 } _z_ring_t;
 
+/**
+ * An iterator of a ring buffer.
+ */
+typedef struct {
+    void *_val;
+
+    const _z_ring_t *_ring;
+    size_t _r_idx;
+    size_t _w_idx;
+} _z_ring_iterator_t;
+
 z_result_t _z_ring_init(_z_ring_t *ring, size_t capacity);
 _z_ring_t _z_ring_make(size_t capacity);
 
@@ -50,8 +61,13 @@ _z_ring_t *_z_ring_clone(const _z_ring_t *xs, z_element_clone_f d_f);
 void _z_ring_clear(_z_ring_t *v, z_element_free_f f);
 void _z_ring_free(_z_ring_t **xs, z_element_free_f f_f);
 
+_z_ring_iterator_t _z_ring_iterator_make(const _z_ring_t *ring);
+bool _z_ring_iterator_next(_z_ring_iterator_t *iter);
+void *_z_ring_iterator_value(const _z_ring_iterator_t *iter);
+
 #define _Z_RING_DEFINE(name, type)                                                                                     \
     typedef _z_ring_t name##_ring_t;                                                                                   \
+    typedef _z_ring_iterator_t name##_ring_iterator_t;                                                                 \
     static inline z_result_t name##_ring_init(name##_ring_t *ring, size_t capacity) {                                  \
         return _z_ring_init(ring, capacity);                                                                           \
     }                                                                                                                  \
@@ -67,7 +83,14 @@ void _z_ring_free(_z_ring_t **xs, z_element_free_f f_f);
     }                                                                                                                  \
     static inline type *name##_ring_pull(name##_ring_t *r) { return (type *)_z_ring_pull(r); }                         \
     static inline void name##_ring_clear(name##_ring_t *r) { _z_ring_clear(r, name##_elem_free); }                     \
-    static inline void name##_ring_free(name##_ring_t **r) { _z_ring_free(r, name##_elem_free); }
+    static inline void name##_ring_free(name##_ring_t **r) { _z_ring_free(r, name##_elem_free); }                      \
+    static inline name##_ring_iterator_t name##_ring_iterator_make(const name##_ring_t *ring) {                        \
+        return _z_ring_iterator_make(ring);                                                                            \
+    }                                                                                                                  \
+    static inline bool name##_ring_iterator_next(name##_ring_iterator_t *iter) { return _z_ring_iterator_next(iter); } \
+    static inline type *name##_ring_iterator_value(const name##_ring_iterator_t *iter) {                               \
+        return (type *)_z_ring_iterator_value(iter);                                                                   \
+    }
 
 #ifdef __cplusplus
 }

include/zenoh-pico/config.h.in

@@ -29,6 +29,7 @@
 #cmakedefine Z_FEATURE_UNSTABLE_API
 #define Z_FEATURE_MULTI_THREAD @Z_FEATURE_MULTI_THREAD@
 #define Z_FEATURE_PUBLICATION @Z_FEATURE_PUBLICATION@
+#define Z_FEATURE_ADVANCED_PUBLICATION @Z_FEATURE_ADVANCED_PUBLICATION@
 #define Z_FEATURE_SUBSCRIPTION @Z_FEATURE_SUBSCRIPTION@
 #define Z_FEATURE_QUERY @Z_FEATURE_QUERY@
 #define Z_FEATURE_QUERYABLE @Z_FEATURE_QUERYABLE@

include/zenoh-pico/net/sample.h

@@ -45,6 +45,7 @@ typedef struct _z_sample_t {
     _z_source_info_t source_info;
 } _z_sample_t;
 void _z_sample_clear(_z_sample_t *sample);
+size_t _z_sample_size(const _z_sample_t *s);
 
 // Warning: None of the sub-types require a non-0 initialization. Add a init function if it changes.
 static inline _z_sample_t _z_sample_null(void) { return (_z_sample_t){0}; }

include/zenoh-pico/utils/query_params.h

@@ -0,0 +1,43 @@
+//
+// Copyright (c) 2025 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 ZENOH_PICO_UTILS_QUERY_PARAMS_H
+#define ZENOH_PICO_UTILS_QUERY_PARAMS_H
+
+#include "zenoh-pico/utils/string.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+extern const char *_Z_QUERY_PARAMS_KEY_TIME;
+
+typedef struct {
+    _z_str_se_t key;
+    _z_str_se_t value;
+} _z_query_param_t;
+
+/**
+ * Extracts the next query parameter from a `_z_str_se_t` string.
+ *
+ * Returns a `_z_query_param_t` with positions of the next key/value in the string if present.
+ * After invocation `str` will point to the remainder of the string.
+ */
+_z_query_param_t _z_query_params_next(_z_str_se_t *str);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* ZENOH_PICO_UTILS_QUERY_PARAMS_H */

include/zenoh-pico/utils/string.h

@@ -33,6 +33,11 @@ typedef struct {
     char const *delimiter;
 } _z_splitstr_t;
 
+/**
+ * Creates a `_z_str_se_t` from a null-terminated C string.
+ */
+_z_str_se_t _z_bstrnew(const char *start);
+
 /**
  * The reverse equivalent of libc's `strstr`.
  *

include/zenoh-pico/utils/time_range.h

@@ -0,0 +1,47 @@
+//
+// Copyright (c) 2025 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 ZENOH_PICO_UTILS_TIME_RANGE_H
+#define ZENOH_PICO_UTILS_TIME_RANGE_H
+
+#include "zenoh-pico/system/platform.h"
+#include "zenoh-pico/utils/string.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct {
+    enum { _Z_TIME_BOUND_INCLUSIVE, _Z_TIME_BOUND_EXCLUSIVE, _Z_TIME_BOUND_UNBOUNDED } bound;
+    double now_offset;
+} _z_time_bound_t;
+
+typedef struct {
+    _z_time_bound_t start;
+    _z_time_bound_t end;
+} _z_time_range_t;
+
+/**
+ * Parse a time range from a string.
+ *
+ * Returns true if the string contained a valid time range, false otherwise.
+ * If valid range will contain the result.
+ */
+bool _z_time_range_from_str(const char *str, size_t len, _z_time_range_t *range);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* ZENOH_PICO_UTILS_TIME_RANGE_H */

src/collections/ring.c

@@ -113,3 +113,24 @@ void _z_ring_free(_z_ring_t **r, z_element_free_f free_f) {
         *r = NULL;
     }
 }
+
+_z_ring_iterator_t _z_ring_iterator_make(const _z_ring_t *ring) {
+    _z_ring_iterator_t iter = {0};
+
+    iter._ring = ring;
+    iter._r_idx = ring->_r_idx;
+    iter._w_idx = ring->_w_idx;
+
+    return iter;
+}
+
+bool _z_ring_iterator_next(_z_ring_iterator_t *iter) {
+    if (iter->_r_idx != iter->_w_idx) {
+        iter->_val = iter->_ring->_val[iter->_r_idx];
+        iter->_r_idx = (iter->_r_idx + (size_t)1) % iter->_ring->_capacity;
+        return true;
+    }
+    return false;
+}
+
+void *_z_ring_iterator_value(const _z_ring_iterator_t *iter) { return iter->_val; }

src/net/sample.c

@@ -40,6 +40,11 @@ void _z_sample_clear(_z_sample_t *sample) {
     _z_bytes_drop(&sample->attachment);
 }
 
+size_t _z_sample_size(const _z_sample_t *s) {
+    (void)(s);
+    return sizeof(_z_sample_t);
+}
+
 void _z_sample_free(_z_sample_t **sample) {
     _z_sample_t *ptr = *sample;
     if (ptr != NULL) {

src/utils/query_params.c

@@ -0,0 +1,49 @@
+//
+// Copyright (c) 2025 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 "zenoh-pico/utils/query_params.h"
+
+#include "zenoh-pico/utils/pointers.h"
+
+const char *_Z_QUERY_PARAMS_KEY_TIME = "_time";
+
+static const char *_Z_QUERY_PARAMS_LIST_SEPARATOR = ";";
+static const char *_Z_QUERY_PARAMS_FIELD_SEPARATOR = "=";
+
+_z_query_param_t _z_query_params_next(_z_str_se_t *str) {
+    _z_query_param_t result = {0};
+
+    _z_splitstr_t params = {.s = *str, .delimiter = _Z_QUERY_PARAMS_LIST_SEPARATOR};
+    _z_str_se_t param = _z_splitstr_next(&params);
+    str->start = params.s.start;
+    str->end = params.s.end;
+
+    if (param.start != NULL) {
+        _z_splitstr_t kvpair = {.s = param, .delimiter = _Z_QUERY_PARAMS_FIELD_SEPARATOR};
+        _z_str_se_t key = _z_splitstr_next(&kvpair);
+
+        // Set key if length > 0
+        if (_z_ptr_char_diff(key.end, key.start) > 0) {
+            result.key.start = key.start;
+            result.key.end = key.end;
+
+            // Set value if length > 0
+            if (_z_ptr_char_diff(kvpair.s.end, kvpair.s.start) > 0) {
+                result.value.start = kvpair.s.start;
+                result.value.end = kvpair.s.end;
+            }
+        }
+    }
+    return result;
+}