add subtree filtering to notifications subscription

Change-Id: Ie547b5f478cc8e3b09ea2f324b62854576787e1b

Author: peckato1

CMakeLists.txt

@@ -82,7 +82,7 @@ if(BUILD_TESTING)
         --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/yang/ietf-interfaces@2018-02-20.yang
         --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/yang/ietf-ip@2018-02-22.yang
         --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/yang/ietf-network-instance@2019-01-21.yang
-        --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/yang/ietf-subscribed-notifications@2019-09-09.yang -e replay
+        --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/yang/ietf-subscribed-notifications@2019-09-09.yang -e replay -e subtree
         --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/yang/ietf-yang-push@2019-09-09.yang -e on-change
         )
 

include/sysrepo-cpp/Session.hpp

@@ -137,17 +137,17 @@ class Session {
             const std::optional<FDHandling>& callbacks = std::nullopt);
 
     [[nodiscard]] DynamicSubscription yangPushPeriodic(
-        const std::optional<std::string>& xpathFilter,
+        const std::optional<std::variant<std::string, libyang::DataNodeAny>>& filter,
         std::chrono::milliseconds periodTime,
         const std::optional<NotificationTimeStamp>& anchorTime = std::nullopt,
         const std::optional<NotificationTimeStamp>& stopTime = std::nullopt);
     [[nodiscard]] DynamicSubscription yangPushOnChange(
-        const std::optional<std::string>& xpathFilter,
+        const std::optional<std::variant<std::string, libyang::DataNodeAny>>& filter,
         const std::optional<std::chrono::milliseconds>& dampeningPeriod = std::nullopt,
         SyncOnStart syncOnStart = SyncOnStart::No,
         const std::optional<NotificationTimeStamp>& stopTime = std::nullopt);
     [[nodiscard]] DynamicSubscription subscribeNotifications(
-        const std::optional<std::string>& xpathFilter,
+        const std::optional<std::variant<std::string, libyang::DataNodeAny>>& filter,
         const std::optional<std::string>& stream = std::nullopt,
         const std::optional<NotificationTimeStamp>& stopTime = std::nullopt,
         const std::optional<NotificationTimeStamp>& startTime = std::nullopt);

src/Session.cpp

@@ -42,6 +42,37 @@ libyang::DataNode wrapSrData(std::shared_ptr<sr_session_ctx_s> sess, sr_data_t*
         sr_release_data(data);
     }));
 }
+
+std::optional<std::string> constructXPathFilter(const std::optional<std::variant<std::string, libyang::DataNodeAny>>& filter)
+{
+    if (!filter) {
+        return std::nullopt;
+    }
+
+    if (std::holds_alternative<std::string>(*filter)) {
+        return std::get<std::string>(*filter);
+    }
+
+    auto node = std::get<libyang::DataNodeAny>(*filter);
+    auto value = node.releaseValue();
+
+    if (!value) {
+        return "/"; // select nothing, RFC 6241, 6.4.2
+    }
+
+    if (std::holds_alternative<libyang::DataNode>(*value)) {
+        char* str;
+
+        auto filterTree = std::get<libyang::DataNode>(*value);
+        auto res = srsn_filter_subtree2xpath(libyang::getRawNode(filterTree), nullptr, &str);
+        std::unique_ptr<char, decltype([](auto* p) constexpr { std::free(p); })> strDeleter(str); // pass ownership of c-string to the deleter
+
+        throwIfError(res, "Unable to convert subtree filter to xpath");
+        return str;
+    }
+
+    throw Error("Subtree filter anydata node must contain (possibly empty) libyang tree");
+}
 }
 
 /**
@@ -526,17 +557,17 @@ Subscription Session::onNotification(
 /**
  * Subscribe for receiving notifications according to 'ietf-yang-push' YANG periodic subscriptions.
  *
- * Wraps `srsn_yang_push_periodic`.
+ * Wraps `srsn_subscribe` and `srsn_filter_subtree2xpath` for subtree filters.
  *
- * @param xpathFilter Optional XPath that filters received notification.
+ * @param filter Optional filter for received notification, xpath filter for string type, subtree filter for libyang::DataNodeAny
  * @param periodTime Notification period.
  * @param anchorTime Optional anchor time for the period. Anchor time acts as a reference point for the period.
  * @param stopTime Optional stop time ending the notification subscription.
  *
  * @return A YangPushSubscription handle.
  */
 DynamicSubscription Session::yangPushPeriodic(
-    const std::optional<std::string>& xpathFilter,
+    const std::optional<std::variant<std::string, libyang::DataNodeAny>>& filter,
     std::chrono::milliseconds periodTime,
     const std::optional<NotificationTimeStamp>& anchorTime,
     const std::optional<NotificationTimeStamp>& stopTime)
@@ -545,6 +576,8 @@ DynamicSubscription Session::yangPushPeriodic(
     uint32_t subId;
     auto stopSpec = stopTime ? std::optional{toTimespec(*stopTime)} : std::nullopt;
     auto anchorSpec = anchorTime ? std::optional{toTimespec(*anchorTime)} : std::nullopt;
+    auto xpathFilter = constructXPathFilter(filter);
+
     auto res = srsn_yang_push_periodic(m_sess.get(),
                                        toDatastore(activeDatastore()),
                                        xpathFilter ? xpathFilter->c_str() : nullptr,
@@ -561,24 +594,26 @@ DynamicSubscription Session::yangPushPeriodic(
 /**
  * Subscribe for receiving notifications according to 'ietf-yang-push' YANG on-change subscriptions.
  *
- * Wraps `srsn_yang_push_on_change`.
+ * Wraps `srsn_subscribe` and `srsn_filter_subtree2xpath` for subtree filters.
  *
- * @param xpathFilter Optional XPath that filters received notification.
+ * @param filter Optional filter for received notification, xpath filter for string type, subtree filter for libyang::DataNodeAny
  * @param dampeningPeriod Optional dampening period.
  * @param syncOnStart Whether to start with a notification of the current state.
  * @param stopTime Optional stop time ending the notification subscription.
  *
  * @return A YangPushSubscription handle.
  */
 DynamicSubscription Session::yangPushOnChange(
-    const std::optional<std::string>& xpathFilter,
+    const std::optional<std::variant<std::string, libyang::DataNodeAny>>& filter,
     const std::optional<std::chrono::milliseconds>& dampeningPeriod,
     SyncOnStart syncOnStart,
     const std::optional<NotificationTimeStamp>& stopTime)
 {
     int fd;
     uint32_t subId;
     auto stopSpec = stopTime ? std::optional{toTimespec(*stopTime)} : std::nullopt;
+    auto xpathFilter = constructXPathFilter(filter);
+
     auto res = srsn_yang_push_on_change(m_sess.get(),
                                         toDatastore(activeDatastore()),
                                         xpathFilter ? xpathFilter->c_str() : nullptr,
@@ -596,19 +631,19 @@ DynamicSubscription Session::yangPushOnChange(
 }
 
 /**
- * Subscribe for receiving notifications according to 'ietf-subscribed-notifications'
+ * Subscribe for receiving notifications according to 'ietf-subscribed-notifications'.
  *
- * Wraps `srsn_subscribe.
+ * Wraps `srsn_subscribe` and `srsn_filter_subtree2xpath` for subtree filters.
  *
- * @param xpathFilter Optional XPath that filters received notification.
+ * @param filter Optional filter for received notification, xpath filter for string type, subtree filter for libyang::DataNodeAny
  * @param stream Optional stream to subscribe to.
  * @param stopTime Optional stop time ending the subscription.
  * @param startTime Optional start time of the subscription, used for replaying stored notifications.
  *
  * @return A YangPushSubscription handle.
  */
 DynamicSubscription Session::subscribeNotifications(
-    const std::optional<std::string>& xpathFilter,
+    const std::optional<std::variant<std::string, libyang::DataNodeAny>>& filter,
     const std::optional<std::string>& stream,
     const std::optional<NotificationTimeStamp>& stopTime,
     const std::optional<NotificationTimeStamp>& startTime)
@@ -618,10 +653,11 @@ DynamicSubscription Session::subscribeNotifications(
     auto stopSpec = stopTime ? std::optional{toTimespec(*stopTime)} : std::nullopt;
     auto startSpec = startTime ? std::optional{toTimespec(*startTime)} : std::nullopt;
     struct timespec replayStartSpec;
+    auto xpathFilter = constructXPathFilter(filter);
 
     auto res = srsn_subscribe(m_sess.get(),
                               stream ? stream->c_str() : nullptr,
-                              xpathFilter ? xpathFilter->c_str() : nullptr,
+                              xpathFilter ? xpathFilter->data() : nullptr,
                               stopSpec ? &stopSpec.value() : nullptr,
                               startSpec ? &startSpec.value() : nullptr,
                               false,

tests/subscriptions-dynamic.cpp

@@ -28,6 +28,9 @@
 #define REQUIRE_NOTIFICATION(SUBSCRIPTION, NOTIFICATION) \
     TROMPELOEIL_REQUIRE_CALL(rec, recordNotification(NOTIFICATION)).IN_SEQUENCE(seq);
 
+#define REQUIRE_NAMED_NOTIFICATION(SUBSCRIPTION, NOTIFICATION) \
+    expectations.emplace_back(TROMPELOEIL_NAMED_REQUIRE_CALL(rec, recordNotification(NOTIFICATION)).IN_SEQUENCE(seq));
+
 #define READ_NOTIFICATION(SUBSCRIPTION)                                \
     REQUIRE(pipeStatus((SUBSCRIPTION).fd()) == PipeStatus::DataReady); \
     (SUBSCRIPTION).processEvent(cbNotif);
@@ -276,6 +279,89 @@ TEST_CASE("Dynamic subscriptions")
             const auto excMessage = "Couldn't terminate yang-push subscription with id " + std::to_string(sub->subscriptionId()) + ": SR_ERR_NOT_FOUND";
             REQUIRE_THROWS_WITH_AS(sub->terminate(), excMessage.c_str(), sysrepo::ErrorWithCode);
         }
+
+        DOCTEST_SUBCASE("Filtering")
+        {
+            std::optional<sysrepo::DynamicSubscription> sub;
+            std::vector<std::unique_ptr<trompeloeil::expectation>> expectations;
+
+            DOCTEST_SUBCASE("xpath filter")
+            {
+                sub = sess.subscribeNotifications("/test_module:ping");
+
+                REQUIRE_NAMED_NOTIFICATION(sub, notifications[0]);
+            }
+
+            DOCTEST_SUBCASE("subtree filter")
+            {
+                libyang::CreatedNodes createdNodes;
+
+                DOCTEST_SUBCASE("filter a node")
+                {
+                    DOCTEST_SUBCASE("XML")
+                    {
+                        createdNodes = sess.getContext().newPath2(
+                            "/ietf-subscribed-notifications:establish-subscription/stream-subtree-filter",
+                            libyang::XML{"<ping xmlns='urn:ietf:params:xml:ns:yang:test_module' />"});
+                    }
+
+                    DOCTEST_SUBCASE("JSON")
+                    {
+                        createdNodes = sess.getContext().newPath2(
+                            "/ietf-subscribed-notifications:establish-subscription/stream-subtree-filter",
+                            libyang::JSON{R"({"test_module:ping": {}})"});
+                    }
+
+                    REQUIRE_NAMED_NOTIFICATION(sub, notifications[0]);
+                }
+
+                DOCTEST_SUBCASE("filter more top level nodes")
+                {
+                    DOCTEST_SUBCASE("XML")
+                    {
+                        createdNodes = sess.getContext().newPath2(
+                            "/ietf-subscribed-notifications:establish-subscription/stream-subtree-filter",
+                            libyang::XML{"<ping xmlns='urn:ietf:params:xml:ns:yang:test_module' />"
+                                         "<silent-ping xmlns='urn:ietf:params:xml:ns:yang:test_module' />"});
+                    }
+
+                    DOCTEST_SUBCASE("JSON")
+                    {
+                        createdNodes = sess.getContext().newPath2(
+                            "/ietf-subscribed-notifications:establish-subscription/stream-subtree-filter",
+                            libyang::JSON{R"({
+                                "test_module:ping": {},
+                                "test_module:silent-ping": {}
+                            })"});
+                    }
+
+                    REQUIRE_NAMED_NOTIFICATION(sub, notifications[0]);
+                    REQUIRE_NAMED_NOTIFICATION(sub, notifications[1]);
+                }
+
+                DOCTEST_SUBCASE("empty filter selects nothing")
+                {
+                    createdNodes = sess.getContext().newPath2(
+                        "/ietf-subscribed-notifications:establish-subscription/stream-subtree-filter",
+                        std::nullopt);
+                }
+
+                sub = sess.subscribeNotifications(createdNodes.createdNode->asAny());
+            }
+
+            CLIENT_SEND_NOTIFICATION(notifications[0]);
+            CLIENT_SEND_NOTIFICATION(notifications[1]);
+
+            // read as many notifications as we expect
+            for (size_t i = 0; i < expectations.size(); ++i) {
+                READ_NOTIFICATION_BLOCKING(*sub);
+            }
+
+            sub->terminate();
+
+            // ensure no more notifications were sent
+            REQUIRE_PIPE_HANGUP(*sub);
+        }
     }
 
     DOCTEST_SUBCASE("YANG Push on change")
@@ -285,9 +371,73 @@ TEST_CASE("Dynamic subscriptions")
          * between writing to sysrepo and reading the notifications.
          */
 
-        auto sub = sess.yangPushOnChange(std::nullopt, std::nullopt, sysrepo::SyncOnStart::Yes);
+        DOCTEST_SUBCASE("Filters")
+        {
+            std::optional<sysrepo::DynamicSubscription> sub;
+
+            DOCTEST_SUBCASE("XPath filter")
+            {
+                sub = sess.yangPushOnChange("/test_module:leafInt32 | /test_module:popelnice/content/trash[name='asd']");
+            }
+
+            DOCTEST_SUBCASE("Subtree filter")
+            {
+                auto createdNodes = sess.getContext().newPath2(
+                    "/ietf-subscribed-notifications:establish-subscription/ietf-yang-push:datastore-subtree-filter",
+                    libyang::XML{"<leafInt32 xmlns='http://example.com/' />"
+                                 "<popelnice xmlns='http://example.com/'><content><trash><name>asd</name></trash></content></popelnice>"});
+                sub = sess.yangPushOnChange(createdNodes.createdNode->asAny());
+            }
+
+            client.setItem("/test_module:leafInt32", "42");
+            client.setItem("/test_module:popelnice/s", "asd");
+            client.setItem("/test_module:popelnice/content/trash[name='asd']", std::nullopt);
+            client.applyChanges();
 
-        REQUIRE_YANG_PUSH_UPDATE(sub, R"({
+            client.deleteItem("/test_module:popelnice/s");
+            client.applyChanges();
+
+            REQUIRE_YANG_PUSH_UPDATE(*sub, R"({
+  "ietf-yang-push:push-change-update": {
+    "datastore-changes": {
+      "yang-patch": {
+        "patch-id": "patch-1",
+        "edit": [
+          {
+            "edit-id": "edit-1",
+            "operation": "create",
+            "target": "/test_module:leafInt32",
+            "value": {
+              "test_module:leafInt32": 42
+            }
+          },
+          {
+            "edit-id": "edit-2",
+            "operation": "create",
+            "target": "/test_module:popelnice/content/trash[name='asd']",
+            "value": {
+              "test_module:trash": {
+                "name": "asd"
+              }
+            }
+          }
+        ]
+      }
+    }
+  }
+}
+)");
+            READ_YANG_PUSH_UPDATE(*sub);
+
+            sub->terminate();
+            REQUIRE_PIPE_HANGUP(*sub);
+        }
+
+        DOCTEST_SUBCASE("Sync on start")
+        {
+            auto sub = sess.yangPushOnChange(std::nullopt, std::nullopt, sysrepo::SyncOnStart::Yes);
+
+            REQUIRE_YANG_PUSH_UPDATE(sub, R"({
   "ietf-yang-push:push-update": {
     "datastore-contents": {
       "test_module:values": [
@@ -298,14 +448,14 @@ TEST_CASE("Dynamic subscriptions")
   }
 }
 )");
-        READ_YANG_PUSH_UPDATE(sub);
+            READ_YANG_PUSH_UPDATE(sub);
 
-        client.setItem("/test_module:leafInt32", "123");
-        client.setItem("/test_module:values[.='5']", std::nullopt);
-        client.deleteItem("/test_module:values[.='3']");
-        client.applyChanges();
+            client.setItem("/test_module:leafInt32", "123");
+            client.setItem("/test_module:values[.='5']", std::nullopt);
+            client.deleteItem("/test_module:values[.='3']");
+            client.applyChanges();
 
-        REQUIRE_YANG_PUSH_UPDATE(sub, R"({
+            REQUIRE_YANG_PUSH_UPDATE(sub, R"({
   "ietf-yang-push:push-change-update": {
     "datastore-changes": {
       "yang-patch": {
@@ -342,10 +492,11 @@ TEST_CASE("Dynamic subscriptions")
   }
 }
 )");
-        READ_YANG_PUSH_UPDATE(sub);
+            READ_YANG_PUSH_UPDATE(sub);
 
-        sub.terminate();
-        REQUIRE_PIPE_HANGUP(sub);
+            sub.terminate();
+            REQUIRE_PIPE_HANGUP(sub);
+        }
     }
 
     DOCTEST_SUBCASE("YANG Push periodic")