ceph/src/rgw/rgw_notify.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #pragma once
   5
   6 #include <string>
   7 #include "common/ceph_time.h"
   8 #include "include/common_fwd.h"
   9 #include "rgw_notify_event_type.h"
  10 #include "common/async/yield_context.h"
  11 #include "cls/2pc_queue/cls_2pc_queue_types.h"
  12 #include "rgw_pubsub.h"
  13
  14 // forward declarations
  15 namespace rgw::sal {
  16     class RGWRadosStore;
  17     class RGWObject;
  18 }
  19
  20 class RGWRados;
  21 struct rgw_obj_key;
  22
  23 namespace rgw::notify {
  24
  25 // initialize the notification manager
  26 // notification manager is dequeing the 2-phase-commit queues
  27 // and send the notifications to the endpoints
  28 bool init(CephContext* cct, rgw::sal::RGWRadosStore* store, const DoutPrefixProvider *dpp);
  29
  30 // shutdown the notification manager
  31 void shutdown();
  32
  33 // create persistent delivery queue for a topic (endpoint)
  34 // this operation also add a topic name to the common (to all RGWs) list of all topics
  35 int add_persistent_topic(const std::string& topic_name, optional_yield y);
  36
  37 // remove persistent delivery queue for a topic (endpoint)
  38 // this operation also remove the topic name from the common (to all RGWs) list of all topics
  39 int remove_persistent_topic(const std::string& topic_name, optional_yield y);
  40
  41 // struct holding reservation information
  42 // populated in the publish_reserve call
  43 // then used to commit or abort the reservation
  44 struct reservation_t {
  45   struct topic_t {
  46     topic_t(const std::string& _configurationId, const rgw_pubsub_topic& _cfg, cls_2pc_reservation::id_t _res_id) :
  47         configurationId(_configurationId), cfg(_cfg), res_id(_res_id) {}
  48
  49     const std::string configurationId;
  50     const rgw_pubsub_topic cfg;
  51     // res_id is reset after topic is committed/aborted
  52     cls_2pc_reservation::id_t res_id;
  53   };
  54
  55   const DoutPrefixProvider *dpp;
  56   std::vector<topic_t> topics;
  57   rgw::sal::RGWRadosStore* const store;
  58   const req_state* const s;
  59   size_t size;
  60   rgw::sal::RGWObject* const object;
  61
  62   reservation_t(const DoutPrefixProvider *_dpp, rgw::sal::RGWRadosStore* _store, const req_state* _s, rgw::sal::RGWObject* _object) :
  63       dpp(_dpp), store(_store), s(_s), object(_object) {}
  64
  65   // dtor doing resource leak guarding
  66   // aborting the reservation if not already committed or aborted
  67   ~reservation_t();
  68 };
  69
  70 // create a reservation on the 2-phase-commit queue
  71 int publish_reserve(const DoutPrefixProvider *dpp,
  72         EventType event_type,
  73         reservation_t& reservation,
  74         const RGWObjTags* req_tags);
  75
  76 // commit the reservation to the queue
  77 int publish_commit(rgw::sal::RGWObject* obj,
  78         uint64_t size,
  79         const ceph::real_time& mtime,
  80         const std::string& etag,
  81         EventType event_type,
  82         reservation_t& reservation,
  83         const DoutPrefixProvider *dpp);
  84
  85 // cancel the reservation
  86 int publish_abort(const DoutPrefixProvider *dpp, reservation_t& reservation);
  87
  88 }
  89