wrap dynamic subscription functions

This patch wraps the dynamic subscription functions. We add a new
handle for the subscription that contains the file descriptor where
clients are supposed to send the notifications.

We tried to reuse the current Subscription class for dynamic
subscriptions as well and not let sysrepo create its own internal
background threads. However, subscribing to yang push periodic timer
creates a new thread in sysrepo anyway, so completely decoupling from
the C API would be much work.

This patch does not yet wrap the functions for modifying subscriptions.

The new DynamicSubscription class implements pimpl idiom. The class is
movable. When the subscription is moved, the destructor is called on
the old object. However, the destructor is supposed to close the file
descriptor. We need to make sure that the file descriptor is not closed
twice.
I was thinking about using std::unique_ptr to manage the file descriptor
or by setting the fd to something "invalid" after the move. I found the
former solution more elegant.

Maybe we should refactor the Subscription class to use pimpl idiom as
well.

Depends-on: https://gerrit.cesnet.cz/c/CzechLight/libyang-cpp/+/8149
Change-Id: I1cee5e730fcc762b2e0a58d3d5c40bf358504397

Author: peckato1

CMakeLists.txt

@@ -76,9 +76,19 @@ if(BUILD_TESTING)
     set(fixture-test-module
         --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/test_module.yang
         )
+    set(fixture-dynamic-subscriptions
+        --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/test_module.yang
+        --install ${CMAKE_CURRENT_SOURCE_DIR}/tests/yang/iana-if-type@2014-05-08.yang
+        --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-yang-push@2019-09-09.yang -e on-change
+        )
 
     sysrepo_cpp_test(NAME session FIXTURE fixture-test-module)
     sysrepo_cpp_test(NAME subscriptions FIXTURE fixture-test-module LIBRARIES Threads::Threads)
+    sysrepo_cpp_test(NAME subscriptions-dynamic FIXTURE fixture-dynamic-subscriptions LIBRARIES PkgConfig::SYSREPO)
     sysrepo_cpp_test(NAME unsafe FIXTURE fixture-test-module LIBRARIES PkgConfig::SYSREPO)
 endif()
 

include/sysrepo-cpp/Session.hpp

@@ -136,6 +136,22 @@ class Session {
             ExceptionHandler handler = nullptr,
             const std::optional<FDHandling>& callbacks = std::nullopt);
 
+    [[nodiscard]] DynamicSubscription yangPushPeriodic(
+        const std::optional<std::string>& xpathFilter,
+        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::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::string>& stream = std::nullopt,
+        const std::optional<NotificationTimeStamp>& stopTime = std::nullopt,
+        const std::optional<NotificationTimeStamp>& startTime = std::nullopt);
+
     ChangeCollection getChanges(const std::string& xpath = "//.");
     void setErrorMessage(const std::string& msg);
     void setNetconfError(const NetconfErrorInfo& info);

include/sysrepo-cpp/Subscription.hpp

@@ -189,6 +189,13 @@ using RpcActionCb = std::function<ErrorCode(Session session, uint32_t subscripti
  */
 using NotifCb = std::function<void(Session session, uint32_t subscriptionId, const NotificationType type, const std::optional<libyang::DataNode> notificationTree, const NotificationTimeStamp timestamp)>;
 
+/**
+ * A callback for YANG push notification subscriptions.
+ * @param notification The notification tree.
+ * @param timestamp Time when the notification was generated.
+ */
+using YangPushNotifCb = std::function<void(const std::optional<libyang::DataNode> notificationTree, const NotificationTimeStamp timestamp)>;
+
 /**
  * Exception handler type for handling exceptions thrown in user callbacks.
  */
@@ -268,4 +275,46 @@ class Subscription {
 
     bool m_didNacmInit;
 };
+
+enum class SyncOnStart : bool {
+    No,
+    Yes,
+};
+
+/**
+ * @brief Manages lifetime of YANG push subscriptions.
+ *
+ * Users are supposed to create instances of this class via Session::yangPushPeriodic, Session::yangPushOnChange or Session::subscribeNotifications.
+ * Whenever notified about a change (by polling the file descriptor obtained by fd() function),
+ * there is at least one event waiting to be processed by a call to YangPushSubscription::processEvent.
+ *
+ * Internally, the sysrepo C library creates some background thread(s). These are used either for managing internal,
+ * sysrepo-level module subscriptions, or for scheduling of periodic timers. These threads are fully encapsulated by
+ * the C code, and there is no control over them from this C++ wrapper. The public interface of this class is a file
+ * descriptor that the caller is expected to poll for readability/closing (and the subscription ID). Once the FD is
+ * readable, invoke this class' processEvents(). There is no automatic event loop which would take care of this
+ * functionality, and users are expected to integrate this FD into their own event handling.
+ */
+class DynamicSubscription {
+public:
+    DynamicSubscription(const DynamicSubscription&) = delete;
+    DynamicSubscription& operator=(const DynamicSubscription&) = delete;
+    DynamicSubscription(DynamicSubscription&&) noexcept;
+    DynamicSubscription& operator=(DynamicSubscription&&) noexcept;
+    ~DynamicSubscription();
+
+    int fd() const;
+    uint64_t subscriptionId() const;
+    std::optional<NotificationTimeStamp> replayStartTime() const;
+    void processEvent(YangPushNotifCb cb) const;
+    void terminate(const std::optional<std::string>& reason = std::nullopt);
+
+private:
+    DynamicSubscription(std::shared_ptr<sr_session_ctx_s> sess, int fd, uint64_t subId, const std::optional<NotificationTimeStamp>& replayStartTime = std::nullopt);
+
+    struct Data;
+    std::unique_ptr<Data> m_data;
+
+    friend class Session;
+};
 }

src/Session.cpp

@@ -11,6 +11,7 @@ extern "C" {
 #include <sysrepo.h>
 #include <sysrepo/netconf_acm.h>
 #include <sysrepo/error_format.h>
+#include <sysrepo/subscribed_notifications.h>
 }
 #include <libyang-cpp/Context.hpp>
 #include <sysrepo-cpp/Connection.hpp>
@@ -522,6 +523,123 @@ Subscription Session::onNotification(
     return sub;
 }
 
+/**
+ * Subscribe for receiving notifications according to 'ietf-yang-push' YANG periodic subscriptions.
+ *
+ * Wraps `srsn_yang_push_periodic`.
+ *
+ * @param xpathFilter Optional XPath that filters received notification.
+ * @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,
+    std::chrono::milliseconds periodTime,
+    const std::optional<NotificationTimeStamp>& anchorTime,
+    const std::optional<NotificationTimeStamp>& stopTime)
+{
+    int fd;
+    uint32_t subId;
+    auto stopSpec = stopTime ? std::optional{toTimespec(*stopTime)} : std::nullopt;
+    auto anchorSpec = anchorTime ? std::optional{toTimespec(*anchorTime)} : std::nullopt;
+    auto res = srsn_yang_push_periodic(m_sess.get(),
+                                       toDatastore(activeDatastore()),
+                                       xpathFilter ? xpathFilter->c_str() : nullptr,
+                                       periodTime.count(),
+                                       anchorSpec ? &anchorSpec.value() : nullptr,
+                                       stopSpec ? &stopSpec.value() : nullptr,
+                                       &fd,
+                                       &subId);
+    throwIfError(res, "Couldn't create yang-push periodic subscription", m_sess.get());
+
+    return {m_sess, fd, subId};
+}
+
+/**
+ * Subscribe for receiving notifications according to 'ietf-yang-push' YANG on-change subscriptions.
+ *
+ * Wraps `srsn_yang_push_on_change`.
+ *
+ * @param xpathFilter Optional XPath that filters received notification.
+ * @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::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 res = srsn_yang_push_on_change(m_sess.get(),
+                                        toDatastore(activeDatastore()),
+                                        xpathFilter ? xpathFilter->c_str() : nullptr,
+                                        dampeningPeriod ? dampeningPeriod->count() : 0,
+                                        syncOnStart == SyncOnStart::Yes,
+                                        nullptr,
+                                        stopSpec ? &stopSpec.value() : nullptr,
+                                        0,
+                                        nullptr,
+                                        &fd,
+                                        &subId);
+    throwIfError(res, "Couldn't create yang-push on-change subscription", m_sess.get());
+
+    return {m_sess, fd, subId};
+}
+
+/**
+ * Subscribe for receiving notifications according to 'ietf-subscribed-notifications'
+ *
+ * Wraps `srsn_subscribe.
+ *
+ * @param xpathFilter Optional XPath that filters received notification.
+ * @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::string>& stream,
+    const std::optional<NotificationTimeStamp>& stopTime,
+    const std::optional<NotificationTimeStamp>& startTime)
+{
+    int fd;
+    uint32_t subId;
+    auto stopSpec = stopTime ? std::optional{toTimespec(*stopTime)} : std::nullopt;
+    auto startSpec = startTime ? std::optional{toTimespec(*startTime)} : std::nullopt;
+    struct timespec replayStartSpec;
+
+    auto res = srsn_subscribe(m_sess.get(),
+                              stream ? stream->c_str() : nullptr,
+                              xpathFilter ? xpathFilter->c_str() : nullptr,
+                              stopSpec ? &stopSpec.value() : nullptr,
+                              startSpec ? &startSpec.value() : nullptr,
+                              false,
+                              nullptr,
+                              &replayStartSpec,
+                              &fd,
+                              &subId);
+
+    throwIfError(res, "Couldn't create notification subscription", m_sess.get());
+
+    std::optional<NotificationTimeStamp> replayStart;
+    if (replayStartSpec.tv_sec != 0) {
+        replayStart = toTimePoint(replayStartSpec);
+    }
+
+    return {m_sess, fd, subId, replayStart};
+}
+
 /**
  * Returns a collection of changes based on an `xpath`. Use "//." to get a full change subtree.
  *

src/Subscription.cpp

@@ -11,6 +11,7 @@
 extern "C" {
 #include <sysrepo.h>
 #include <sysrepo/netconf_acm.h>
+#include <sysrepo/subscribed_notifications.h>
 }
 #include "utils/enum.hpp"
 #include "utils/exception.hpp"
@@ -410,4 +411,104 @@ bool ChangeIterator::operator==(const ChangeIterator& other) const
         // And then either both contain nothing or contain the same thing.
         (!this->m_current.has_value() || this->m_current->node == other.m_current->node);
 }
+
+
+struct DynamicSubscription::Data {
+    std::shared_ptr<sr_session_ctx_s> sess;
+    int fd;
+    uint64_t subId;
+    std::optional<NotificationTimeStamp> m_replayStartTime;
+    bool m_terminated;
+
+    Data(std::shared_ptr<sr_session_ctx_s> sess, int fd, uint64_t subId, const std::optional<NotificationTimeStamp>& replayStartTime, bool terminated);
+    ~Data();
+    void terminate(const std::optional<std::string>& reason = std::nullopt);
+};
+
+DynamicSubscription::DynamicSubscription(std::shared_ptr<sr_session_ctx_s> sess, int fd, uint64_t subId, const std::optional<NotificationTimeStamp>& replayStartTime)
+    : m_data(std::make_unique<Data>(std::move(sess), fd, subId, replayStartTime, false))
+{
+}
+
+DynamicSubscription::DynamicSubscription(DynamicSubscription&&) noexcept = default;
+DynamicSubscription& DynamicSubscription::operator=(DynamicSubscription&&) noexcept = default;
+DynamicSubscription::~DynamicSubscription() = default;
+
+/** @brief Returns the file descriptor associated with this subscription. */
+int DynamicSubscription::fd() const
+{
+    return m_data->fd;
+}
+
+/** @brief Returns the subscription ID associated with this subscription. */
+uint64_t DynamicSubscription::subscriptionId() const
+{
+    return m_data->subId;
+}
+
+/** @brief Returns the actual start time of replayed notification subscription, if available. */
+std::optional<NotificationTimeStamp> DynamicSubscription::replayStartTime() const
+{
+    return m_data->m_replayStartTime;
+}
+
+/** @brief Terminates the subscription.
+ *
+ * Wraps `srsn_terminate`.
+ */
+void DynamicSubscription::terminate(const std::optional<std::string>& reason)
+{
+    m_data->terminate(reason);
+}
+
+/** @brief Processes a single event associated with this subscription.
+ *
+ * Invoke only when the file descriptor associated with this subscription is ready for reading.
+ * Otherwise, the function blocks unless the FD is set to non-blocking.
+ *
+ * Wraps `srsn_read_notif`.
+ */
+void DynamicSubscription::processEvent(YangPushNotifCb cb) const
+{
+    struct timespec timestamp;
+    struct lyd_node* tree;
+    auto ctx = std::unique_ptr<const ly_ctx, std::function<void(const ly_ctx*)>>(sr_session_acquire_context(m_data->sess.get()), [&](const ly_ctx*) { sr_session_release_context(m_data->sess.get()); });
+
+    auto err = srsn_read_notif(fd(), ctx.get(), &timestamp, &tree);
+    throwIfError(err, "Couldn't read yang-push notification");
+
+    const auto wrappedNotification = tree ? std::optional{libyang::wrapRawNode(tree)} : std::nullopt;
+
+    // If we see subscription-terminated notification, the subscription is terminated by sysrepo (e.g. because of stop-time)
+    // This is not the correct notification as per RFC to use when reaching stop-time, see https://github.com/sysrepo/sysrepo/issues/3525.
+    if (wrappedNotification && wrappedNotification->path() == "/ietf-subscribed-notifications:subscription-terminated") {
+        m_data->m_terminated = true;
+    }
+
+    cb(wrappedNotification, toTimePoint(timestamp));
+}
+
+DynamicSubscription::Data::Data(std::shared_ptr<sr_session_ctx_s> sess, int fd, uint64_t subId, const std::optional<NotificationTimeStamp>& replayStartTime, bool terminated)
+    : sess(std::move(sess))
+    , fd(fd)
+    , subId(subId)
+    , m_replayStartTime(replayStartTime)
+    , m_terminated(terminated)
+{
+}
+
+DynamicSubscription::Data::~Data()
+{
+    if (!m_terminated) {
+        terminate();
+    }
+    close(fd);
+}
+
+void DynamicSubscription::Data::terminate(const std::optional<std::string>& reason)
+{
+    auto err = srsn_terminate(subId, reason ? reason->c_str() : nullptr);
+    throwIfError(err, "Couldn't terminate yang-push subscription with id " + std::to_string(subId));
+    m_terminated = true;
+}
 }