feat(server): expiry notifications

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

Author: dranikpg

src/core/bloom.cc

@@ -8,6 +8,7 @@
 #include <absl/numeric/bits.h>
 #include <xxhash.h>
 
+#include <algorithm>
 #include <cmath>
 
 #include "base/logging.h"

src/core/string_or_view.h

@@ -4,6 +4,8 @@
 
 #pragma once
 
+#include <absl/types/span.h>
+
 #include <string>
 #include <string_view>
 #include <variant>

src/facade/dragonfly_connection.cc

@@ -275,22 +275,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) {
@@ -361,7 +345,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) +
@@ -449,8 +433,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

@@ -71,15 +71,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

@@ -161,7 +161,7 @@ struct ErrorReply {
   }
   explicit ErrorReply(const char* msg,
                       std::string_view kind = {})  // to resolve ambiguity of constructors above
-      : message{std::string_view{msg}}, kind{kind} {
+      : message{std::string_view(msg)}, kind{kind} {
   }
   explicit ErrorReply(OpStatus status) : message{}, kind{}, status{status} {
   }

src/facade/reply_builder.cc

@@ -496,7 +496,7 @@ void RedisReplyBuilder::SendMGetResponse(MGetResponse resp) {
 
 void RedisReplyBuilder::SendSimpleStrArr(StrSpan arr) {
   string res = absl::StrCat("*", arr.Size(), kCRLF);
-  for (std::string_view str : arr)
+  for (string_view str : arr)
     StrAppend(&res, "+", str, kCRLF);
 
   SendRaw(res);

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"

src/facade/reply_capture.cc

@@ -18,7 +18,7 @@ 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) {
@@ -170,7 +170,7 @@ struct CaptureVisitor {
   }
 
   void operator()(CapturingReplyBuilder::Error err) {
-    rb->SendError(err.first, err.second);
+    rb->SendError(std::move(err));
   }
 
   void operator()(OpStatus status) {
@@ -226,7 +226,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->ToSv(), 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

@@ -285,7 +285,7 @@ GenericError AclFamily::LoadToRegistryFromFile(std::string_view full_path,
     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())};
+      return {std::string{error.ToSv()}};
     }
     requests.push_back(std::move(std::get<User::UpdateRequest>(req)));
   }