feat(server): expiry notifications

Signed-off-by: Vladislav Oleshko <[email protected]>

Author: dranikpg

src/core/string_or_view.h

@@ -4,6 +4,8 @@
 
 #pragma once
 
+#include <absl/types/span.h>
+
 #include <string>
 #include <string_view>
 #include <variant>
@@ -69,4 +71,68 @@ class StringOrView {
   std::variant<std::string_view, std::string> val_;
 };
 
+// Whenever you need to pass const string[] or string_view[]
+struct StringOrViewSpan {
+  StringOrViewSpan() = default;
+  StringOrViewSpan(const StringOrViewSpan&) = default;
+  StringOrViewSpan(StringOrViewSpan&&) = default;
+
+  template <typename T, typename = typename std::enable_if_t<
+                            !std::is_base_of_v<StringOrViewSpan, std::decay_t<T>>>>
+  StringOrViewSpan(T&& span) : span_{std::forward<T>(span)} {
+  }
+
+  struct Iterator {
+    Iterator(const StringOrViewSpan& span, size_t idx) : idx(idx), span(span) {
+    }
+
+    using iterator_category = std::forward_iterator_tag;
+    using value_type = std::string_view;
+
+    Iterator& operator++() {
+      idx++;
+      return *this;
+    }
+
+    std::string_view operator*() const {
+      return span[idx];
+    }
+
+    bool operator==(const Iterator& other) const {
+      return &other.span == &span && other.idx == idx;
+    }
+
+    bool operator!=(const Iterator& other) const {
+      return !operator==(other);
+    }
+
+   private:
+    size_t idx = 0;
+    const StringOrViewSpan& span;
+  };
+
+  std::string_view operator[](size_t i) const {
+    return std::visit([i](const auto& span) -> std::string_view { return span[i]; }, span_);
+  }
+
+  size_t Size() const {
+    return std::visit([](const auto& span) { return span.size(); }, span_);
+  }
+
+  Iterator begin() const {
+    return {*this, 0};
+  }
+
+  Iterator end() const {
+    return {*this, Size()};
+  }
+
+  static std::string OwnedOp(std::string_view str) {
+    return std::string{str};
+  }
+
+ private:
+  std::variant<absl::Span<const std::string_view>, absl::Span<const std::string>> span_;
+};
+
 }  // namespace dfly

src/facade/dragonfly_connection.cc

@@ -272,22 +272,6 @@ struct Connection::Shutdown {
   }
 };
 
-Connection::PubMessage::PubMessage(string pattern, shared_ptr<char[]> buf, size_t channel_len,
-                                   size_t message_len)
-    : pattern{std::move(pattern)},
-      buf{std::move(buf)},
-      channel_len{channel_len},
-      message_len{message_len} {
-}
-
-string_view Connection::PubMessage::Channel() const {
-  return {buf.get(), channel_len};
-}
-
-string_view Connection::PubMessage::Message() const {
-  return {buf.get() + channel_len, message_len};
-}
-
 void Connection::PipelineMessage::SetArgs(const RespVec& args) {
   auto* next = storage.data();
   for (size_t i = 0; i < args.size(); ++i) {
@@ -358,7 +342,7 @@ size_t Connection::PipelineMessage::StorageCapacity() const {
 size_t Connection::MessageHandle::UsedMemory() const {
   struct MessageSize {
     size_t operator()(const PubMessagePtr& msg) {
-      return sizeof(PubMessage) + (msg->channel_len + msg->message_len);
+      return sizeof(PubMessage) + (msg->channel.size() + msg->message.size());
     }
     size_t operator()(const PipelineMessagePtr& msg) {
       return sizeof(PipelineMessage) + msg->args.capacity() * sizeof(MutableSlice) +
@@ -447,8 +431,8 @@ void Connection::DispatchOperations::operator()(const PubMessage& pub_msg) {
     arr[i++] = "pmessage";
     arr[i++] = pub_msg.pattern;
   }
-  arr[i++] = pub_msg.Channel();
-  arr[i++] = pub_msg.Message();
+  arr[i++] = pub_msg.channel;
+  arr[i++] = pub_msg.message;
   rbuilder->SendStringArr(absl::Span<string_view>{arr.data(), i},
                           RedisReplyBuilder::CollectionType::PUSH);
 }

src/facade/dragonfly_connection.h

@@ -67,15 +67,9 @@ class Connection : public util::Connection {
 
   // PubSub message, either incoming message for active subscription or reply for new subscription.
   struct PubMessage {
-    std::string pattern{};            // non-empty for pattern subscriber
-    std::shared_ptr<char[]> buf;      // stores channel name and message
-    size_t channel_len, message_len;  // lengths in buf
-
-    std::string_view Channel() const;
-    std::string_view Message() const;
-
-    PubMessage(std::string pattern, std::shared_ptr<char[]> buf, size_t channel_len,
-               size_t message_len);
+    std::string pattern{};              // non-empty for pattern subscriber
+    std::shared_ptr<char[]> buf;        // stores channel name and message
+    std::string_view channel, message;  // channel and message parts from buf
   };
 
   // Pipeline message, accumulated Redis command to be executed.

src/facade/facade_types.h

@@ -12,6 +12,7 @@
 #include <string_view>
 #include <variant>
 
+#include "core/string_or_view.h"
 #include "facade/op_status.h"
 
 namespace facade {
@@ -107,22 +108,19 @@ struct FacadeStats {
 
 struct ErrorReply {
   explicit ErrorReply(std::string&& msg, std::string_view kind = {})
-      : message{std::move(msg)}, kind{kind} {
+      : message{dfly::StringOrView::FromString(std::move(msg))}, kind{kind} {
   }
-  explicit ErrorReply(std::string_view msg, std::string_view kind = {}) : message{msg}, kind{kind} {
+  explicit ErrorReply(std::string_view msg, std::string_view kind = {})
+      : message{dfly::StringOrView::FromView(msg)}, kind{kind} {
   }
   explicit ErrorReply(const char* msg,
                       std::string_view kind = {})  // to resolve ambiguity of constructors above
-      : message{std::string_view{msg}}, kind{kind} {
+      : message{dfly::StringOrView::FromView(msg)}, kind{kind} {
   }
   explicit ErrorReply(OpStatus status) : message{}, kind{}, status{status} {
   }
 
-  std::string_view ToSv() const {
-    return std::visit([](auto& str) { return std::string_view(str); }, message);
-  }
-
-  std::variant<std::string, std::string_view> message;
+  dfly::StringOrView message;
   std::string_view kind;
   std::optional<OpStatus> status{std::nullopt};
 };

src/facade/reply_builder.cc

@@ -139,9 +139,7 @@ bool SinkReplyBuilder::HasReplied() const {
 void SinkReplyBuilder::SendError(ErrorReply error) {
   if (error.status)
     return SendError(*error.status);
-
-  string_view message_sv = visit([](auto&& str) -> string_view { return str; }, error.message);
-  SendError(message_sv, error.kind);
+  SendError(error.message.view(), error.kind);
 }
 
 void SinkReplyBuilder::SendError(OpStatus status) {
@@ -264,14 +262,6 @@ void MCReplyBuilder::SendNotFound() {
   SendSimpleString("NOT_FOUND");
 }
 
-size_t RedisReplyBuilder::WrappedStrSpan::Size() const {
-  return visit([](auto arr) { return arr.size(); }, (const StrSpan&)*this);
-}
-
-string_view RedisReplyBuilder::WrappedStrSpan::operator[](size_t i) const {
-  return visit([i](auto arr) { return string_view{arr[i]}; }, (const StrSpan&)*this);
-}
-
 char* RedisReplyBuilder::FormatDouble(double val, char* dest, unsigned dest_len) {
   StringBuilder sb(dest, dest_len);
   CHECK(dfly_conv.ToShortest(val, &sb));
@@ -504,12 +494,10 @@ void RedisReplyBuilder::SendMGetResponse(MGetResponse resp) {
 }
 
 void RedisReplyBuilder::SendSimpleStrArr(StrSpan arr) {
-  WrappedStrSpan warr{arr};
-
-  string res = absl::StrCat("*", warr.Size(), kCRLF);
+  string res = absl::StrCat("*", arr.Size(), kCRLF);
 
-  for (unsigned i = 0; i < warr.Size(); i++)
-    StrAppend(&res, "+", warr[i], kCRLF);
+  for (string_view str : arr)
+    StrAppend(&res, "+", str, kCRLF);
 
   SendRaw(res);
 }
@@ -523,16 +511,13 @@ void RedisReplyBuilder::SendEmptyArray() {
 }
 
 void RedisReplyBuilder::SendStringArr(StrSpan arr, CollectionType type) {
-  WrappedStrSpan warr{arr};
-
-  if (type == ARRAY && warr.Size() == 0) {
+  if (type == ARRAY && arr.Size() == 0) {
     SendRaw("*0\r\n");
     return;
   }
 
-  auto cb = [&](size_t i) { return warr[i]; };
-
-  SendStringArrInternal(warr.Size(), std::move(cb), type);
+  auto cb = [&](size_t i) { return arr[i]; };
+  SendStringArrInternal(arr.Size(), std::move(cb), type);
 }
 
 void RedisReplyBuilder::StartArray(unsigned len) {

src/facade/reply_builder.h

@@ -8,6 +8,7 @@
 #include <optional>
 #include <string_view>
 
+#include "core/string_or_view.h"
 #include "facade/facade_types.h"
 #include "facade/op_status.h"
 #include "io/io.h"
@@ -203,7 +204,7 @@ class RedisReplyBuilder : public SinkReplyBuilder {
 
   enum VerbatimFormat { TXT, MARKDOWN };
 
-  using StrSpan = std::variant<absl::Span<const std::string>, absl::Span<const std::string_view>>;
+  using StrSpan = dfly::StringOrViewSpan;
 
   RedisReplyBuilder(::io::Sink* stream);
 
@@ -242,12 +243,6 @@ class RedisReplyBuilder : public SinkReplyBuilder {
 
   static char* FormatDouble(double val, char* dest, unsigned dest_len);
 
- protected:
-  struct WrappedStrSpan : public StrSpan {
-    size_t Size() const;
-    std::string_view operator[](size_t index) const;
-  };
-
  private:
   void SendStringArrInternal(size_t size, absl::FunctionRef<std::string_view(unsigned)> producer,
                              CollectionType type);

src/facade/reply_capture.cc

@@ -18,15 +18,13 @@ using namespace std;
 
 void CapturingReplyBuilder::SendError(std::string_view str, std::string_view type) {
   SKIP_LESS(ReplyMode::ONLY_ERR);
-  Capture(Error{str, type});
+  Capture(facade::ErrorReply(string{str}, type));
 }
 
 void CapturingReplyBuilder::SendError(ErrorReply error) {
   SKIP_LESS(ReplyMode::ONLY_ERR);
-
-  string message =
-      visit([](auto&& str) -> string { return string{std::move(str)}; }, error.message);
-  Capture(Error{std::move(message), error.kind});
+  error.message.MakeOwned();
+  Capture(std::move(error));
 }
 
 void CapturingReplyBuilder::SendMGetResponse(MGetResponse resp) {
@@ -53,11 +51,8 @@ void CapturingReplyBuilder::SendSimpleStrArr(StrSpan arr) {
   SKIP_LESS(ReplyMode::FULL);
   DCHECK_EQ(current_.index(), 0u);
 
-  WrappedStrSpan warr{arr};
-  vector<string> sarr(warr.Size());
-  for (unsigned i = 0; i < warr.Size(); i++)
-    sarr[i] = warr[i];
-
+  vector<string> sarr(arr.Size());
+  transform(arr.begin(), arr.end(), sarr.begin(), StrSpan::OwnedOp);
   Capture(StrArrPayload{true, ARRAY, std::move(sarr)});
 }
 
@@ -66,11 +61,8 @@ void CapturingReplyBuilder::SendStringArr(StrSpan arr, CollectionType type) {
   DCHECK_EQ(current_.index(), 0u);
 
   // TODO: 1. Allocate all strings at once 2. Allow movable types
-  WrappedStrSpan warr{arr};
-  vector<string> sarr(warr.Size());
-  for (unsigned i = 0; i < warr.Size(); i++)
-    sarr[i] = warr[i];
-
+  vector<string> sarr(arr.Size());
+  transform(arr.begin(), arr.end(), sarr.begin(), StrSpan::OwnedOp);
   Capture(StrArrPayload{false, type, std::move(sarr)});
 }
 
@@ -183,7 +175,7 @@ struct CaptureVisitor {
   }
 
   void operator()(CapturingReplyBuilder::Error err) {
-    rb->SendError(err.first, err.second);
+    rb->SendError(std::move(err));
   }
 
   void operator()(OpStatus status) {
@@ -239,7 +231,7 @@ void CapturingReplyBuilder::SetReplyMode(ReplyMode mode) {
 
 optional<CapturingReplyBuilder::ErrorRef> CapturingReplyBuilder::GetError(const Payload& pl) {
   if (auto* err = get_if<Error>(&pl); err != nullptr) {
-    return ErrorRef{err->first, err->second};
+    return ErrorRef{err->message.view(), err->kind};
   }
   return nullopt;
 }

src/facade/reply_capture.h

@@ -48,8 +48,8 @@ class CapturingReplyBuilder : public RedisReplyBuilder {
   void StartCollection(unsigned len, CollectionType type) override;
 
  public:
-  using Error = std::pair<std::string, std::string>;  // SendError (msg, type)
-  using Null = std::nullptr_t;                        // SendNull or SendNullArray
+  using Error = facade::ErrorReply;
+  using Null = std::nullptr_t;  // SendNull or SendNullArray
 
   struct StrArrPayload {
     bool simple;

src/server/acl/acl_family.cc

@@ -282,8 +282,8 @@ GenericError AclFamily::LoadToRegistryFromFile(std::string_view full_path,
     auto req = ParseAclSetUser<std::vector<std::string_view>&>(cmds, *cmd_registry_, true);
     if (std::holds_alternative<ErrorReply>(req)) {
       auto error = std::move(std::get<ErrorReply>(req));
-      LOG(WARNING) << "Error while parsing aclfile: " << error.ToSv();
-      return {std::string(error.ToSv())};
+      LOG(WARNING) << "Error while parsing aclfile: " << error.message.view();
+      return {std::string{error.message.view()}};
     }
     requests.push_back(std::move(std::get<User::UpdateRequest>(req)));
   }