rgw/sal: add/remove_persistent_topic() interface

Signed-off-by: Casey Bodley <cbodley@redhat.com>
(cherry picked from commit 51ab34d8283dfd08d639860d98f055621baeed44)

diff --git a/src/rgw/driver/rados/rgw_sal_rados.cc b/src/rgw/driver/rados/rgw_sal_rados.cc
index 64444e4f8b729acf00b174695ccbf136628018bf..f1d7b18be4f2712ea552355bf6ec8714987bbda1 100644 (file)
--- a/src/rgw/driver/rados/rgw_sal_rados.cc
+++ b/src/rgw/driver/rados/rgw_sal_rados.cc
@@ -1726,6 +1726,22 @@ int RadosStore::list_account_topics(const DoutPrefixProvider* dpp,
                                 listing.topics, listing.next_marker);
 }
 
+int RadosStore::add_persistent_topic(const DoutPrefixProvider* dpp,
+                                     optional_yield y,
+                                     const std::string& topic_queue)
+{
+  return rgw::notify::add_persistent_topic(
+      dpp, getRados()->get_notif_pool_ctx(), topic_queue, y);
+}
+
+int RadosStore::remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                        optional_yield y,
+                                        const std::string& topic_queue)
+{
+  return rgw::notify::remove_persistent_topic(
+      dpp, getRados()->get_notif_pool_ctx(), topic_queue, y);
+}
+
 int RadosStore::remove_bucket_mapping_from_topics(
     const rgw_pubsub_bucket_topics& bucket_topics,
     const std::string& bucket_key,

diff --git a/src/rgw/driver/rados/rgw_sal_rados.h b/src/rgw/driver/rados/rgw_sal_rados.h
index 78bd849717f605434aab8fac7257292d6a084d5d..9f5d021987ad3979b61b36bccdb71312f9096d9f 100644 (file)
--- a/src/rgw/driver/rados/rgw_sal_rados.h
+++ b/src/rgw/driver/rados/rgw_sal_rados.h
@@ -319,6 +319,12 @@ class RadosStore : public StoreDriver {
                             std::string_view marker,
                             uint32_t max_items,
                             TopicList& listing) override;
+    int add_persistent_topic(const DoutPrefixProvider* dpp,
+                             optional_yield y,
+                             const std::string& topic_queue) override;
+    int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                optional_yield y,
+                                const std::string& topic_queue) override;
     int update_bucket_topic_mapping(const rgw_pubsub_topic& topic,
                                     const std::string& bucket_key,
                                     bool add_mapping,

diff --git a/src/rgw/rgw_sal.h b/src/rgw/rgw_sal.h
index e3a892b6992ceba12c336345763f17c7564ca568..cd29e9c7428345455f5866790f6f1782a00d437a 100644 (file)
--- a/src/rgw/rgw_sal.h
+++ b/src/rgw/rgw_sal.h
@@ -542,6 +542,16 @@ class Driver {
                                     std::string_view marker,
                                     uint32_t max_items,
                                     TopicList& listing) = 0;
+
+    // TODO: backends should manage persistent topic queues internally on
+    // write_topic_v2()/remove_topic_v2()
+    virtual int add_persistent_topic(const DoutPrefixProvider* dpp,
+                                     optional_yield y,
+                                     const std::string& topic_queue) = 0;
+    virtual int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                        optional_yield y,
+                                        const std::string& topic_queue) = 0;
+
     /** Update the bucket-topic mapping in the store, if |add_mapping|=true then
      * adding the |bucket_key| |topic| mapping to store, else delete the
      * |bucket_key| |topic| mapping from the store.  The |bucket_key| is

diff --git a/src/rgw/rgw_sal_dbstore.cc b/src/rgw/rgw_sal_dbstore.cc
index 2ce6304646eac590a4b876a73022ca1855d69652..874935f161974aab522db6fccb283145b6e6623c 100644 (file)
--- a/src/rgw/rgw_sal_dbstore.cc
+++ b/src/rgw/rgw_sal_dbstore.cc
@@ -1912,6 +1912,20 @@ namespace rgw::sal {
     return -ENOTSUP;
   }
 
+  int DBStore::add_persistent_topic(const DoutPrefixProvider* dpp,
+                                    optional_yield y,
+                                    const std::string& topic_queue)
+  {
+    return -ENOTSUP;
+  }
+
+  int DBStore::remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                       optional_yield y,
+                                       const std::string& topic_queue)
+  {
+    return -ENOTSUP;
+  }
+
   RGWLC* DBStore::get_rgwlc(void) {
     return lc;
   }

diff --git a/src/rgw/rgw_sal_dbstore.h b/src/rgw/rgw_sal_dbstore.h
index 4770713e762a6fef597887b680a23057bb35c861..eaa5fda5060cc16a8abfc135b037d5dd2355d8ed 100644 (file)
--- a/src/rgw/rgw_sal_dbstore.h
+++ b/src/rgw/rgw_sal_dbstore.h
@@ -887,6 +887,13 @@ public:
                           uint32_t max_items,
                           TopicList& listing) override;
 
+      int add_persistent_topic(const DoutPrefixProvider* dpp,
+                               optional_yield y,
+                               const std::string& topic_queue) override;
+      int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                  optional_yield y,
+                                  const std::string& topic_queue) override;
+
       virtual RGWLC* get_rgwlc(void) override;
       virtual RGWCoroutinesManagerRegistry* get_cr_registry() override { return NULL; }
       virtual int log_usage(const DoutPrefixProvider *dpp, std::map<rgw_user_bucket, RGWUsageBatch>& usage_info, optional_yield y) override;

diff --git a/src/rgw/rgw_sal_filter.cc b/src/rgw/rgw_sal_filter.cc
index 66a2466040ba56f1c51893fa52f291a9fbbb3e3a..d6d51b204612b7cdc8da317eb74d29f0f7c62ac9 100644 (file)
--- a/src/rgw/rgw_sal_filter.cc
+++ b/src/rgw/rgw_sal_filter.cc
@@ -463,6 +463,20 @@ std::unique_ptr<Notification> FilterDriver::get_notification(
   return std::make_unique<FilterNotification>(std::move(n));
 }
 
+int FilterDriver::add_persistent_topic(const DoutPrefixProvider* dpp,
+                                       optional_yield y,
+                                       const std::string& topic_queue)
+{
+  return next->add_persistent_topic(dpp, y, topic_queue);
+}
+
+int FilterDriver::remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                          optional_yield y,
+                                          const std::string& topic_queue)
+{
+  return next->remove_persistent_topic(dpp, y, topic_queue);
+}
+
 RGWLC* FilterDriver::get_rgwlc()
 {
   return next->get_rgwlc();

diff --git a/src/rgw/rgw_sal_filter.h b/src/rgw/rgw_sal_filter.h
index 95d00960cbd2d4df9ea6af9defc86f8bd6a80502..eaa8af2cb437dc2bf3a5fd009177b3e5001d0819 100644 (file)
--- a/src/rgw/rgw_sal_filter.h
+++ b/src/rgw/rgw_sal_filter.h
@@ -353,6 +353,12 @@ public:
     return next->list_account_topics(dpp, y, account_id, marker,
                                      max_items, listing);
   }
+  int add_persistent_topic(const DoutPrefixProvider* dpp,
+                           optional_yield y,
+                           const std::string& topic_queue) override;
+  int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                              optional_yield y,
+                              const std::string& topic_queue) override;
   int update_bucket_topic_mapping(const rgw_pubsub_topic& topic,
                                   const std::string& bucket_key,
                                   bool add_mapping,