rgw/pubsub: fix comments from PR #27091

Signed-off-by: Yuval Lifshitz <yuvalif@yahoo.com>

diff --git a/src/rgw/rgw_main.cc b/src/rgw/rgw_main.cc
index 63ef569bc11e25549d86fef9a42f6526943f1fda..2d4c8d7468977e00034c18879cddec1cdc579c81 100644 (file)
--- a/src/rgw/rgw_main.cc
+++ b/src/rgw/rgw_main.cc
@@ -41,11 +41,6 @@
 #if defined(WITH_RADOSGW_BEAST_FRONTEND)
 #include "rgw_asio_frontend.h"
 #endif /* WITH_RADOSGW_BEAST_FRONTEND */
-
-#ifdef WITH_RADOSGW_AMQP_ENDPOINT
-#include "rgw_amqp.h"
-#endif /* WITH_RADOSGW_AMQP_ENDPOINT */
-
 #include "rgw_dmclock_scheduler_ctx.h"
 
 #include "services/svc_zone.h"
@@ -306,10 +301,6 @@ int main(int argc, const char **argv)
   FCGX_Init();
 #endif
 
-#ifdef WITH_RADOSGW_AMQP_ENDPOINT
-  rgw::amqp::init(cct.get());
-#endif
-
   RGWRados *store =
     RGWStoreManager::get_storage(g_ceph_context,
                                 g_conf()->rgw_enable_gc_threads,
@@ -602,10 +593,6 @@ int main(int argc, const char **argv)
   rgw_http_client_cleanup();
   rgw::curl::cleanup_curl();
 
-#ifdef WITH_RADOSGW_AMQP_ENDPOINT
-  rgw::amqp::shutdown();
-#endif
-
   rgw_perf_stop(g_ceph_context);
 
   dout(1) << "final shutdown" << dendl;

diff --git a/src/rgw/rgw_pubsub.h b/src/rgw/rgw_pubsub.h
index 1e33303f12bbc6eecb80d9fd4628dc2621873b2d..56c737ce63de0fbdd6d6a628d0bf15926b7f6665 100644 (file)
--- a/src/rgw/rgw_pubsub.h
+++ b/src/rgw/rgw_pubsub.h
@@ -224,7 +224,6 @@ struct rgw_pubsub_event {
 WRITE_CLASS_ENCODER(rgw_pubsub_event)
 
 struct rgw_pubsub_sub_dest {
-  rgw_pubsub_sub_dest() : bucket_name(""), oid_prefix(""), push_endpoint(""), push_endpoint_args("") {}
   std::string bucket_name;
   std::string oid_prefix;
   std::string push_endpoint;

diff --git a/src/rgw/rgw_sync_module_pubsub.cc b/src/rgw/rgw_sync_module_pubsub.cc
index 5153967e203ffa1c2c6d032e2fed938c858e0b48..098892c391f839779649c9412a10fd3af7f6fea3 100644 (file)
--- a/src/rgw/rgw_sync_module_pubsub.cc
+++ b/src/rgw/rgw_sync_module_pubsub.cc
@@ -15,6 +15,9 @@
 #include "rgw_pubsub.h"
 #include "rgw_pubsub_push.h"
 #include "rgw_perf_counters.h"
+#ifdef WITH_RADOSGW_AMQP_ENDPOINT
+#include "rgw_amqp.h"
+#endif
 
 #include <boost/algorithm/hex.hpp>
 #include <boost/asio/yield.hpp>
@@ -1532,6 +1535,17 @@ RGWPSSyncModuleInstance::RGWPSSyncModuleInstance(CephContext *cct, const JSONFor
   } else {
     effective_conf.decode_json(&p);
   }
+#ifdef WITH_RADOSGW_AMQP_ENDPOINT
+  if (!rgw::amqp::init(cct)) {
+    ldout(cct, 1) << "ERROR: failed to initialize AMQP server in pubsub sync module" << dendl;
+  }
+#endif
+}
+
+RGWPSSyncModuleInstance::~RGWPSSyncModuleInstance() {
+#ifdef WITH_RADOSGW_AMQP_ENDPOINT
+  rgw::amqp::shutdown();
+#endif
 }
 
 RGWDataSyncModule *RGWPSSyncModuleInstance::get_data_handler()

diff --git a/src/rgw/rgw_sync_module_pubsub.h b/src/rgw/rgw_sync_module_pubsub.h
index 37a055e76bffa60fc96cb176ad76a1ce69c1e24f..6b4b78153e817399bbf8a09022c9885d11468943 100644 (file)
--- a/src/rgw/rgw_sync_module_pubsub.h
+++ b/src/rgw/rgw_sync_module_pubsub.h
@@ -23,6 +23,7 @@ class RGWPSSyncModuleInstance : public RGWSyncModuleInstance {
   JSONFormattable effective_conf;
 public:
   RGWPSSyncModuleInstance(CephContext *cct, const JSONFormattable& config);
+  ~RGWPSSyncModuleInstance();
   RGWDataSyncModule *get_data_handler() override;
   RGWRESTMgr *get_rest_filter(int dialect, RGWRESTMgr *orig) override;
   bool supports_user_writes() override {

diff --git a/src/rgw/rgw_sync_module_pubsub_rest.cc b/src/rgw/rgw_sync_module_pubsub_rest.cc
index ce00dcf60dcaef1b9c9cde7e8494e01410bacf03..0d781bd4eb482db375b9d82cf7c4e8a2b91c934a 100644 (file)
--- a/src/rgw/rgw_sync_module_pubsub_rest.cc
+++ b/src/rgw/rgw_sync_module_pubsub_rest.cc
@@ -1027,8 +1027,8 @@ private:
     }
     if (sub_conf.s3_id.empty()) {
       if (must_delete) {
-        ldout(s->cct, 1) << "notification does not have an ID, ret=" << op_ret << dendl;
         op_ret = -ENOENT;
+        ldout(s->cct, 1) << "notification does not have an ID, ret=" << op_ret << dendl;
       }
       return;
     }
@@ -1227,8 +1227,8 @@ void RGWPSListNotifs_ObjStore_S3::execute() {
       return;
     }
     if (sub_conf.s3_id.empty()) {
-      ldout(s->cct, 1) << "notification does not have an ID, ret=" << op_ret << dendl;
       op_ret = -ENOENT;
+      ldout(s->cct, 1) << "notification does not have an ID, ret=" << op_ret << dendl;
       return;
     }
     rgw_pubsub_bucket_topics bucket_topics;
@@ -1239,8 +1239,8 @@ void RGWPSListNotifs_ObjStore_S3::execute() {
     }
     const auto topic_it = bucket_topics.topics.find(sub_conf.topic);
     if (topic_it == bucket_topics.topics.end()) {
-      ldout(s->cct, 1) << "notification does not have topic information, ret=" << op_ret << dendl;
       op_ret = -ENOENT;
+      ldout(s->cct, 1) << "notification does not have topic information, ret=" << op_ret << dendl;
       return;
     }
     add_notification_to_list(sub_conf, topic_it->second.events, topic_it->second.topic.arn);