Merge pull request #283 from DenisBiryukov91/querying-subscriber

Querying subscriber

docs/Doxyfile

@@ -165,7 +165,7 @@ ALWAYS_DETAILED_SEC    = NO
 # operators of the base classes will not be shown.
 # The default value is: NO.
 
-INLINE_INHERITED_MEMB  = NO
+INLINE_INHERITED_MEMB  = YES
 
 # If the FULL_PATH_NAMES tag is set to YES, doxygen will prepend the full path
 # before files name in the file list and in the header files. If set to NO the

docs/ext.rst

@@ -17,5 +17,9 @@ Extensions
 Extra functionality, which is not a part of core Zenoh API.
 
 .. doxygenclass:: zenoh::ext::PublicationCache
+   :members:
+   :membergroups: Constructors Operators Methods
+
+.. doxygenclass:: zenoh::ext::QueryingSubscriber
    :members:
    :membergroups: Constructors Operators Methods

docs/session.rst

@@ -24,4 +24,3 @@ Session
 .. doxygenclass:: zenoh::Session
    :members:
    :membergroups: Constructors Operators Methods Fields
-

examples/CMakeLists.txt

@@ -60,6 +60,9 @@ function(add_examples glob mode lib)
 			if(${file} MATCHES ".*liveliness.*$")
 				continue()
 			endif()
+			if(${file} MATCHES ".*query_sub.*$")
+				continue()
+			endif()
 			if(${file} MATCHES ".*pub_cache.*$")
 				continue()
 			endif()

examples/zenohc/z_pub_cache.cxx

@@ -1,5 +1,5 @@
 //
-// Copyright (c) 2022 ZettaScale Technology
+// 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
@@ -46,6 +46,7 @@ int _main(int argc, char **argv) {
     std::cout << "Declaring Publication cache on '" << keyexpr << "'..." << std::endl;
     Session::PublicationCacheOptions opts;
     opts.history = std::atoi(history);
+    opts.queryable_complete = true;
     if (!std::string(prefix).empty()) {
         opts.queryable_prefix = KeyExpr(prefix);
     }

examples/zenohc/z_query_sub.cxx

@@ -0,0 +1,80 @@
+//
+// 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]>
+//
+#include <stdio.h>
+#include <string.h>
+
+#include <chrono>
+#include <iostream>
+#include <limits>
+#include <sstream>
+#include <thread>
+
+#include "../getargs.h"
+#include "zenoh.hxx"
+
+using namespace zenoh;
+using namespace std::chrono_literals;
+
+const char *default_keyexpr = "demo/example/**";
+const char *default_query = "";
+
+const char *kind_to_str(SampleKind kind) {
+    switch (kind) {
+        case SampleKind::Z_SAMPLE_KIND_PUT:
+            return "PUT";
+        case SampleKind::Z_SAMPLE_KIND_DELETE:
+            return "DELETE";
+        default:
+            return "UNKNOWN";
+    }
+}
+
+int _main(int argc, char **argv) {
+    const char *keyexpr = default_keyexpr;
+    const char *query = default_query;
+    Config config = parse_args(argc, argv, {}, {{"key_expression", &keyexpr}}, {{"-q", {"query", &query}}});
+
+    std::cout << "Opening session..." << std::endl;
+    auto session = Session::open(std::move(config));
+
+    std::cout << "Declaring Querying Subscriber on '" << keyexpr << "' with initial query on '" << query << "'"
+              << std::endl;
+    Session::QueryingSubscriberOptions opts;
+
+    if (!std::string(query).empty()) {
+        opts.query_keyexpr = KeyExpr(query);
+        opts.query_accept_replies = ReplyKeyExpr::ZC_REPLY_KEYEXPR_ANY;
+    }
+    auto querying_subscriber = session.declare_querying_subscriber(keyexpr, channels::FifoChannel(16), std::move(opts));
+
+    std::cout << "Press CTRL-C to quit..." << std::endl;
+    for (auto res = querying_subscriber.handler().recv(); std::holds_alternative<Sample>(res);
+         res = querying_subscriber.handler().recv()) {
+        const auto &sample = std::get<Sample>(res);
+        std::cout << ">> [Subscriber] Received " << kind_to_str(sample.get_kind()) << " ('"
+                  << sample.get_keyexpr().as_string_view() << "': '" << sample.get_payload().as_string() << "')"
+                  << std::endl;
+    }
+
+    return 0;
+}
+
+int main(int argc, char **argv) {
+    try {
+        init_log_from_env_or("error");
+        _main(argc, argv);
+    } catch (ZException e) {
+        std::cout << "Received an error :" << e.what() << "\n";
+    }
+}

include/zenoh/api.hxx

@@ -40,4 +40,5 @@
 #include "api/ext/serialization.hxx"
 #if defined(ZENOHCXX_ZENOHC) && defined(Z_FEATURE_UNSTABLE_API)
 #include "api/ext/publication_cache.hxx"
+#include "api/ext/querying_subscriber.hxx"
 #endif

include/zenoh/api/enums.hxx

@@ -110,6 +110,14 @@ inline std::string_view whatami_as_str(WhatAmI whatami) {
 /// - **ZC_LOCALITY_SESSION_LOCAL**: Only from local sessions.
 /// - **ZC_LOCALITY_SESSION_REMOTE**: Only from remote sessions.
 typedef ::zc_locality_t Locality;
+
+/// @warning This API has been marked as unstable: it works as advertised, but it may be changed in a future release.
+/// @brief Key expressions types to which Queryable should reply to.
+///
+/// Values:
+/// - **ZC_REPLY_KEYEXPR_ANY**:  Replies to any key expression queries.
+/// - **ZC_REPLY_KEYEXPR_MATCHING_QUERY**: Replies only to queries with intersecting key expressions.
+typedef ::zc_reply_keyexpr_t ReplyKeyExpr;
 #endif
 
 }  // namespace zenoh