This is an automated email from the ASF dual-hosted git repository.
git-hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 1418e23da feat(keyspace): emit set and del notifications (#3541)
1418e23da is described below
commit 1418e23da766cbe9d72b03bebc52d0152f1cf0bd
Author: Aether <[email protected]>
AuthorDate: Fri Aug 28 15:52:42 2026 +0800
feat(keyspace): emit set and del notifications (#3541)
Implement initial keyspace notifications for set and del events,
compatible with Redis notify-keyspace-events.
This PR adds the initial notification emitters and configuration
support:
- Support `K`, `E`, `g`, `$`, and `A` flags, where `A` currently expands
to the implemented event classes `g$`.
- Emit `set` notifications only when the shared `Set` path actually
applies a write, including conditional `SET` variants.
- Emit `del` notifications only for keys that are actually deleted
through the shared delete path.
- Deduplicate repeated keys in a single delete operation, so the same
key is deleted and notified once.
- Queue notifications inside `MULTI/EXEC` and publish them after a
successful commit.
- Map notification DB names correctly:
- default namespace uses DB `0`
- `redis-databases` namespaces map back to Redis DB indexes
This PR only adds the initial set and del event paths on primary nodes,
keeping the initial patch small and reviewable. Commands sharing the
same underlying mutation APIs may also produce these events.
Replica-side notifications and additional event types, if supported, can
be added in follow-up PRs.
Unsupported notification classes are rejected for now until their
emitters are implemented.
Ref Proposal: #3533
Tracking Issue: #2915
Assisted by Codex/GPT-5.5.
---------
Co-authored-by: hulk <[email protected]>
Co-authored-by: Aleks Lozovyuk <[email protected]>
---
kvrocks.conf | 17 ++
src/commands/cmd_txn.cc | 5 +
src/common/keyspace_events.cc | 64 +++++
src/common/keyspace_events.h | 62 +++++
src/config/config.cc | 13 +
src/config/config.h | 6 +
src/server/redis_connection.cc | 42 +++
src/server/redis_connection.h | 8 +
src/server/server.cc | 12 +
src/server/server.h | 3 +
src/storage/redis_db.cc | 19 +-
src/storage/storage.h | 39 ++-
src/types/redis_string.cc | 6 +-
tests/cppunit/keyspace_events_test.cc | 151 +++++++++++
tests/gocase/unit/keyspace/keyspace_test.go | 3 +
.../unit/keyspacenotify/keyspacenotify_test.go | 291 +++++++++++++++++++++
16 files changed, 738 insertions(+), 3 deletions(-)
diff --git a/kvrocks.conf b/kvrocks.conf
index d13a13cd1..0337b6ab1 100644
--- a/kvrocks.conf
+++ b/kvrocks.conf
@@ -610,6 +610,23 @@ lua-strict-key-accessing no
#
# tls-replication yes
+############################# KEYSPACE NOTIFICATIONS ##########################
+
+# Keyspace notifications publish key changes to SUBSCRIBE and PSUBSCRIBE
clients.
+# Supported flags:
+# K keyspace channels
+# E keyevent channels
+# g generic events, currently del
+# $ string events, currently set
+# A same as g$, without K or E
+#
+# Default namespace uses db 0. Redis database namespaces use db indexes.
+# Other namespaces use their original names.
+# Notifications are emitted only when at least one channel flag (K or E) and
one event class are enabled.
+#
+# Default: "" disabled
+notify-keyspace-events ""
+
################################## SLOW LOG ###################################
# The Kvrocks Slow Log is a mechanism to log queries that exceeded a specified
diff --git a/src/commands/cmd_txn.cc b/src/commands/cmd_txn.cc
index 16d25a58e..0a25d7e72 100644
--- a/src/commands/cmd_txn.cc
+++ b/src/commands/cmd_txn.cc
@@ -90,6 +90,11 @@ class CommandExec : public Commander {
s = storage->CommitTxn();
}
+ // Publish queued notifications after a successful commit.
+ if (s.IsOK()) {
+ conn->FlushKeyspaceEvents();
+ }
+
conn->ResetMultiExec();
reset_multiexec.Disable();
diff --git a/src/common/keyspace_events.cc b/src/common/keyspace_events.cc
new file mode 100644
index 000000000..9f9be31c4
--- /dev/null
+++ b/src/common/keyspace_events.cc
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "keyspace_events.h"
+
+#include <cstring>
+
+#include "config/config.h"
+#include "fmt/format.h"
+
+StatusOr<std::pair<KeyspaceEventChannel, KeyspaceEventType>>
ParseNotifyKeyspaceEventsFlags(std::string_view input) {
+ int channel_flags = 0;
+ int type_flags = 0;
+ for (const char c : input) {
+ switch (c) {
+ case 'K':
+ channel_flags |= kNotifyKeyspace;
+ break;
+ case 'E':
+ channel_flags |= kNotifyKeyevent;
+ break;
+ case 'A':
+ type_flags |= kNotifyAll;
+ break;
+ case 'g':
+ type_flags |= kNotifyGeneric;
+ break;
+ case '$':
+ type_flags |= kNotifyString;
+ break;
+ default:
+ return {Status::NotOK, fmt::format("unsupported notify-keyspace-events
flag: '{}'", c)};
+ }
+ }
+
+ return std::pair{static_cast<KeyspaceEventChannel>(channel_flags),
static_cast<KeyspaceEventType>(type_flags)};
+}
+
+std::string FormatKeyspaceNotificationScope(const std::string &ns, int
redis_databases) {
+ if (ns == kDefaultNamespace) {
+ return "0";
+ }
+ if (redis_databases > 0 && ns.rfind(kDatabaseNamespacePrefix, 0) == 0) {
+ return ns.substr(strlen(kDatabaseNamespacePrefix));
+ }
+ return ns;
+}
diff --git a/src/common/keyspace_events.h b/src/common/keyspace_events.h
new file mode 100644
index 000000000..5b4744e70
--- /dev/null
+++ b/src/common/keyspace_events.h
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#pragma once
+
+#include <string>
+#include <string_view>
+#include <utility>
+
+#include "status.h"
+
+enum KeyspaceEventChannel {
+ kNotifyNoChannel = 0,
+ kNotifyKeyspace = 1 << 0, // K, keyspace channels
+ kNotifyKeyevent = 1 << 1, // E, keyevent channels
+};
+
+// Event type flags for notify-keyspace-events, separate from RedisType.
+enum KeyspaceEventType {
+ kNotifyNoType = 0,
+ kNotifyGeneric = 1 << 0, // g, emits del
+ kNotifyString = 1 << 1, // $, emits set
+ // A, supported data classes without K or E.
+ kNotifyAll = kNotifyGeneric | kNotifyString,
+};
+
+struct KeyspaceEvent {
+ KeyspaceEvent(KeyspaceEventType type_flag, std::string_view event,
KeyspaceEventChannel channel_flags,
+ std::string_view ns, std::string_view key)
+ : type_flag(type_flag), channel_flags(channel_flags), event(event),
ns(ns), key(key) {}
+
+ KeyspaceEventType type_flag;
+ KeyspaceEventChannel channel_flags;
+ std::string event;
+ std::string ns;
+ std::string key;
+};
+
+// Parses notify-keyspace-events flags into channel flags followed by event
type flags.
+StatusOr<std::pair<KeyspaceEventChannel, KeyspaceEventType>>
ParseNotifyKeyspaceEventsFlags(std::string_view input);
+
+// Formats the namespace or database scope used in keyspace notification
channel names.
+// Default namespace maps to 0; database namespaces map back to db indexes
when redis-databases is enabled.
+// Other namespaces use their original names.
+std::string FormatKeyspaceNotificationScope(const std::string &ns, int
redis_databases);
diff --git a/src/config/config.cc b/src/config/config.cc
index 0de6a2c6f..11a69a025 100644
--- a/src/config/config.cc
+++ b/src/config/config.cc
@@ -35,6 +35,7 @@
#include <utility>
#include <vector>
+#include "common/keyspace_events.h"
#include "common/string_util.h"
#include "config_type.h"
#include "config_util.h"
@@ -192,6 +193,7 @@ Config::Config() {
{"compact-cron", false, new StringField(&compact_cron_str_, "")},
{"bgsave-cron", false, new StringField(&bgsave_cron_str_, "")},
{"dbsize-scan-cron", false, new StringField(&dbsize_scan_cron_str_, "")},
+ {"notify-keyspace-events", false, new
StringField(¬ify_keyspace_events_str_, "")},
{"replica-announce-ip", false, new StringField(&replica_announce_ip,
"")},
{"replica-announce-port", false, new UInt32Field(&replica_announce_port,
0, 0, PORT_LIMIT)},
{"compaction-checker-range", false, new
StringField(&compaction_checker_range_str_, "")},
@@ -379,6 +381,10 @@ void Config::initFieldValidator() {
}
return Status::OK();
}},
+ {"notify-keyspace-events",
+ []([[maybe_unused]] const std::string &k, const std::string &v) ->
Status {
+ return ParseNotifyKeyspaceEventsFlags(v).ToStatus();
+ }},
{"compact-cron",
[this]([[maybe_unused]] const std::string &k, const std::string &v) ->
Status {
std::vector<std::string> args = util::Split(v, " \t");
@@ -556,6 +562,13 @@ void Config::initFieldCallback() {
srv->AdjustWorkerThreads();
return Status::OK();
}},
+ {"notify-keyspace-events",
+ [this]([[maybe_unused]] Server *srv, [[maybe_unused]] const
std::string &k, const std::string &v) -> Status {
+ const auto flags = GET_OR_RET(ParseNotifyKeyspaceEventsFlags(v));
+ notify_keyspace_event_channels = flags.first;
+ notify_keyspace_event_types = flags.second;
+ return Status::OK();
+ }},
{"dir",
[this]([[maybe_unused]] Server *srv, [[maybe_unused]] const
std::string &k,
[[maybe_unused]] const std::string &v) -> Status {
diff --git a/src/config/config.h b/src/config/config.h
index a6dd19f21..d8313bddc 100644
--- a/src/config/config.h
+++ b/src/config/config.h
@@ -32,6 +32,7 @@
#include "config_type.h"
#include "cron.h"
+#include "keyspace_events.h"
#include "spdlog/common.h"
#include "status.h"
#include "storage/redis_metadata.h"
@@ -219,6 +220,10 @@ struct Config {
// Enable transactional mode in engine::Context
bool txn_context_enabled = false;
+ // Parsed notify-keyspace-events flags.
+ KeyspaceEventChannel notify_keyspace_event_channels = kNotifyNoChannel;
+ KeyspaceEventType notify_keyspace_event_types = kNotifyNoType;
+
bool skip_block_cache_deallocation_on_close = false;
bool lua_strict_key_accessing = false;
@@ -315,6 +320,7 @@ struct Config {
std::string compaction_checker_cron_str_;
std::string profiling_sample_commands_str_;
std::string client_output_buffer_limit_str_;
+ std::string notify_keyspace_events_str_;
std::map<std::string, std::unique_ptr<ConfigField>> fields_;
std::vector<std::string> rename_command_;
std::string histogram_bucket_boundaries_str_;
diff --git a/src/server/redis_connection.cc b/src/server/redis_connection.cc
index 65061c017..f212966d8 100644
--- a/src/server/redis_connection.cc
+++ b/src/server/redis_connection.cc
@@ -23,6 +23,7 @@
#include <mutex>
#include <nonstd/span.hpp>
+#include <optional>
#include <shared_mutex>
#include "commands/commander.h"
@@ -634,6 +635,7 @@ void Connection::ExecuteCommands(std::deque<CommandTokens>
*to_process_cmds) {
}
SetLastCmd(cmd_name);
+ std::vector<KeyspaceEvent> keyspace_events;
{
std::optional<MultiLockGuard> guard;
if (cmd_flags & kCmdWrite) {
@@ -652,6 +654,9 @@ void Connection::ExecuteCommands(std::deque<CommandTokens>
*to_process_cmds) {
guard.emplace(srv_->storage->GetLockManager(), lock_keys);
}
engine::Context ctx(srv_->storage);
+ if (cmd_flags & kCmdWrite) {
+
ctx.EnableKeyspaceEventCollection(config->notify_keyspace_event_channels,
config->notify_keyspace_event_types);
+ }
std::vector<GlobalIndexer::RecordResult> index_records;
if (!srv_->index_mgr.index_map.empty() && IsCmdForIndexing(cmd_flags,
attributes->category) &&
@@ -679,6 +684,13 @@ void Connection::ExecuteCommands(std::deque<CommandTokens>
*to_process_cmds) {
WARN("[connection] index updating failed for key: {}", record.key);
}
}
+ if (ctx.HasKeyspaceEvents()) {
+ keyspace_events = ctx.TakeKeyspaceEvents();
+ }
+ }
+ // Nested Lua and function commands reuse their outer context. Publish
only after index updates and key unlocking.
+ if (!keyspace_events.empty()) {
+ queueOrPublishKeyspaceEvents(std::move(keyspace_events));
}
if (!(cmd_flags & redis::kCmdSkipMonitor)) {
@@ -709,10 +721,40 @@ void
Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) {
}
}
+void Connection::queueOrPublishKeyspaceEvents(std::vector<KeyspaceEvent>
&&events) {
+ if (events.empty()) return;
+
+ if (in_exec_) {
+ // Queue transaction events until commit.
+ for (auto &event : events) {
+ pending_keyspace_events_.emplace_back(std::move(event));
+ }
+ return;
+ }
+
+ for (const auto &event : events) {
+ srv_->NotifyKeyspaceEvent(event);
+ }
+}
+
+void Connection::FlushKeyspaceEvents() {
+ for (const auto &e : pending_keyspace_events_) {
+ srv_->NotifyKeyspaceEvent(e);
+ }
+ pending_keyspace_events_.clear();
+}
+
void Connection::ResetMultiExec() {
in_exec_ = false;
multi_error_ = false;
multi_cmds_.clear();
+ // Drop events from failed or aborted transactions.
+ pending_keyspace_events_.clear();
+ // Retain capacity for typical transactions, but request releasing unusually
large buffers.
+ constexpr std::size_t kMaxRetainedKeyspaceEvents = 1024;
+ if (pending_keyspace_events_.capacity() > kMaxRetainedKeyspaceEvents) {
+ pending_keyspace_events_.shrink_to_fit();
+ }
DisableFlag(Connection::kMultiExec);
}
diff --git a/src/server/redis_connection.h b/src/server/redis_connection.h
index c4bbd630c..3678bfa08 100644
--- a/src/server/redis_connection.h
+++ b/src/server/redis_connection.h
@@ -30,6 +30,7 @@
#include <vector>
#include "commands/commander.h"
+#include "common/keyspace_events.h"
#include "event_util.h"
#include "redis_request.h"
#include "server/redis_reply.h"
@@ -208,6 +209,8 @@ class Connection : public EvbufCallbackBase<Connection> {
void ResetMultiExec();
std::deque<redis::CommandTokens> *GetMultiExecCommands() { return
&multi_cmds_; }
+ void FlushKeyspaceEvents();
+
std::function<void(int)> close_cb = nullptr;
std::set<std::string> watched_keys;
@@ -218,6 +221,9 @@ class Connection : public EvbufCallbackBase<Connection> {
ReplyMode GetReplyMode() const { return reply_mode_; }
private:
+ // Queues events while EXEC is running; publishes them otherwise.
+ void queueOrPublishKeyspaceEvents(std::vector<KeyspaceEvent> &&events);
+
uint64_t id_ = 0;
std::atomic<int> flags_ = 0;
std::string ns_;
@@ -248,6 +254,8 @@ class Connection : public EvbufCallbackBase<Connection> {
bool multi_error_ = false;
std::atomic<bool> is_running_ = false;
std::deque<redis::CommandTokens> multi_cmds_;
+
+ std::vector<KeyspaceEvent> pending_keyspace_events_;
bool in_script_ = false;
bool importing_ = false;
diff --git a/src/server/server.cc b/src/server/server.cc
index 48f6177d8..cbb7bd120 100644
--- a/src/server/server.cc
+++ b/src/server/server.cc
@@ -41,6 +41,7 @@
#include "commands/command_parser.h"
#include "commands/commander.h"
+#include "common/keyspace_events.h"
#include "common/string_util.h"
#include "config/config.h"
#include "fmt/format.h"
@@ -478,6 +479,17 @@ int Server::PublishMessage(const std::string &channel,
const std::string &msg) {
return cnt;
}
+void Server::NotifyKeyspaceEvent(const KeyspaceEvent &event) {
+ const std::string scope = FormatKeyspaceNotificationScope(event.ns,
GetConfig()->redis_databases);
+ // Publish keyspace before keyevent for each key.
+ if (event.channel_flags & kNotifyKeyspace) {
+ PublishMessage("__keyspace@" + scope + "__:" + event.key, event.event);
+ }
+ if (event.channel_flags & kNotifyKeyevent) {
+ PublishMessage("__keyevent@" + scope + "__:" + event.event, event.key);
+ }
+}
+
void Server::SubscribeChannel(const std::string &channel, redis::Connection
*conn) {
std::lock_guard<std::mutex> guard(pubsub_channels_mu_);
diff --git a/src/server/server.h b/src/server/server.h
index 4214cecc5..4ceb38fc4 100644
--- a/src/server/server.h
+++ b/src/server/server.h
@@ -218,6 +218,9 @@ class Server {
int GetFetchFileThreadNum() const { return fetch_file_threads_num_; }
int PublishMessage(const std::string &channel, const std::string &msg);
+
+ // Publishes a keyspace event through the channels selected when it was
collected.
+ void NotifyKeyspaceEvent(const KeyspaceEvent &event);
void SubscribeChannel(const std::string &channel, redis::Connection *conn);
void UnsubscribeChannel(const std::string &channel, redis::Connection *conn);
void GetChannelsByPattern(const std::string &pattern,
std::vector<std::string> *channels);
diff --git a/src/storage/redis_db.cc b/src/storage/redis_db.cc
index 99b3d96a3..118468cd9 100644
--- a/src/storage/redis_db.cc
+++ b/src/storage/redis_db.cc
@@ -21,6 +21,8 @@
#include "redis_db.h"
#include <ctime>
+#include <string_view>
+#include <unordered_set>
#include <utility>
#include "cluster/redis_slot.h"
@@ -186,6 +188,13 @@ rocksdb::Status Database::MDel(engine::Context &ctx, const
std::vector<Slice> &k
storage_->MultiGet(ctx, ctx.DefaultMultiGetOptions(), metadata_cf_handle_,
slice_keys.size(), slice_keys.data(),
pin_values.data(), statuses.data());
+ const bool collect_del_events = ctx.IsKeyspaceEventEnabled(kNotifyGeneric);
+ std::vector<size_t> deleted_key_indexes;
+ if (collect_del_events) deleted_key_indexes.reserve(keys.size());
+
+ std::unordered_set<std::string_view> deleted_ns_keys;
+ const bool deduplicate_keys = keys.size() > 1;
+ if (deduplicate_keys) deleted_ns_keys.reserve(keys.size());
for (size_t i = 0; i < slice_keys.size(); i++) {
if (!statuses[i].ok() && !statuses[i].IsNotFound()) return statuses[i];
if (statuses[i].IsNotFound()) continue;
@@ -196,15 +205,23 @@ rocksdb::Status Database::MDel(engine::Context &ctx,
const std::vector<Slice> &k
auto s = metadata.Decode(rocksdb::Slice(pin_values[i].data(),
pin_values[i].size()));
if (!s.ok()) continue;
if (metadata.Expired()) continue;
+ if (deduplicate_keys && !deleted_ns_keys.emplace(ns_keys[i]).second)
continue;
s = batch->Delete(metadata_cf_handle_, ns_keys[i]);
if (!s.ok()) return s;
*deleted_cnt += 1;
+ if (collect_del_events) deleted_key_indexes.emplace_back(i);
}
if (*deleted_cnt == 0) return rocksdb::Status::OK();
- return storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+ s = storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+ if (!s.ok()) return s;
+
+ for (const auto index : deleted_key_indexes) {
+ ctx.AddKeyspaceEventIfEnabled(kNotifyGeneric, "del", namespace_,
keys[index].ToStringView());
+ }
+ return rocksdb::Status::OK();
}
rocksdb::Status Database::Exists(engine::Context &ctx, const
std::vector<Slice> &keys, uint32_t *ret) {
diff --git a/src/storage/storage.h b/src/storage/storage.h
index 6778a12b1..cd4917364 100644
--- a/src/storage/storage.h
+++ b/src/storage/storage.h
@@ -35,9 +35,11 @@
#include <memory>
#include <shared_mutex>
#include <string>
+#include <string_view>
#include <utility>
#include <vector>
+#include "common/keyspace_events.h"
#include "common/port.h"
#include "config/config.h"
#include "lock_manager.h"
@@ -470,13 +472,22 @@ struct Context {
storage = ctx.storage;
snapshot_ = ctx.snapshot_;
batch = std::move(ctx.batch);
+ keyspace_event_channel_flags_ = ctx.keyspace_event_channel_flags_;
+ keyspace_event_type_flags_ = ctx.keyspace_event_type_flags_;
+ keyspace_events_ = std::move(ctx.keyspace_events_);
ctx.storage = nullptr;
ctx.snapshot_ = nullptr;
}
return *this;
}
- Context(Context &&ctx) noexcept : storage(ctx.storage),
batch(std::move(ctx.batch)), snapshot_(ctx.snapshot_) {
+ Context(Context &&ctx) noexcept
+ : storage(ctx.storage),
+ batch(std::move(ctx.batch)),
+ snapshot_(ctx.snapshot_),
+ keyspace_event_channel_flags_(ctx.keyspace_event_channel_flags_),
+ keyspace_event_type_flags_(ctx.keyspace_event_type_flags_),
+ keyspace_events_(std::move(ctx.keyspace_events_)) {
ctx.storage = nullptr;
ctx.snapshot_ = nullptr;
}
@@ -492,6 +503,29 @@ struct Context {
return snapshot_;
}
+ void EnableKeyspaceEventCollection(KeyspaceEventChannel channel_flags,
KeyspaceEventType type_flags) {
+ keyspace_event_channel_flags_ = channel_flags;
+ keyspace_event_type_flags_ = type_flags;
+ }
+
+ bool IsKeyspaceEventEnabled(KeyspaceEventType type_flag) const {
+ return keyspace_event_channel_flags_ != kNotifyNoChannel &&
(keyspace_event_type_flags_ & type_flag) != 0;
+ }
+
+ void AddKeyspaceEventIfEnabled(KeyspaceEventType type_flag, std::string_view
event, std::string_view ns,
+ std::string_view key) {
+ if (!IsKeyspaceEventEnabled(type_flag)) return;
+ keyspace_events_.emplace_back(type_flag, event,
keyspace_event_channel_flags_, ns, key);
+ }
+
+ bool HasKeyspaceEvents() const { return !keyspace_events_.empty(); }
+
+ std::vector<KeyspaceEvent> TakeKeyspaceEvents() {
+ std::vector<KeyspaceEvent> events;
+ events.swap(keyspace_events_);
+ return events;
+ }
+
private:
/// It is only used by NonTransactionContext
explicit Context(engine::Storage *storage, bool txn_mode) :
storage(storage), txn_context_enabled(txn_mode) {}
@@ -501,6 +535,9 @@ struct Context {
/// Normally it will be fixed to the latest Snapshot when the Context is
constructed.
/// If is_txn_mode is false, the snapshot is nullptr.
const rocksdb::Snapshot *snapshot_ = nullptr;
+ KeyspaceEventChannel keyspace_event_channel_flags_ = kNotifyNoChannel;
+ KeyspaceEventType keyspace_event_type_flags_ = kNotifyNoType;
+ std::vector<KeyspaceEvent> keyspace_events_;
};
} // namespace engine
diff --git a/src/types/redis_string.cc b/src/types/redis_string.cc
index 8563275bf..b385d66ce 100644
--- a/src/types/redis_string.cc
+++ b/src/types/redis_string.cc
@@ -345,7 +345,11 @@ rocksdb::Status String::Set(engine::Context &ctx, const
std::string &user_key, c
metadata.expire = expire;
metadata.Encode(&new_raw_value);
new_raw_value.append(value);
- return updateRawValue(ctx, ns_key, new_raw_value);
+ auto s = updateRawValue(ctx, ns_key, new_raw_value);
+ if (!s.ok()) return s;
+
+ ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", namespace_, user_key);
+ return rocksdb::Status::OK();
}
rocksdb::Status String::SetEX(engine::Context &ctx, const std::string
&user_key, const std::string &value,
diff --git a/tests/cppunit/keyspace_events_test.cc
b/tests/cppunit/keyspace_events_test.cc
new file mode 100644
index 000000000..049a8068c
--- /dev/null
+++ b/tests/cppunit/keyspace_events_test.cc
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "common/keyspace_events.h"
+
+#include <gtest/gtest.h>
+
+#include <utility>
+
+#include "config/config.h"
+#include "storage/storage.h"
+
+TEST(KeyspaceEvents, ContextFiltersAndCapturesEvent) {
+ auto ctx = engine::Context::NoTransactionContext(nullptr);
+ EXPECT_FALSE(ctx.HasKeyspaceEvents());
+
+ ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", "tenant", "disabled");
+ EXPECT_FALSE(ctx.HasKeyspaceEvents());
+
+ ctx.EnableKeyspaceEventCollection(kNotifyNoChannel, kNotifyString);
+ EXPECT_FALSE(ctx.IsKeyspaceEventEnabled(kNotifyString));
+
+ ctx.EnableKeyspaceEventCollection(kNotifyKeyspace, kNotifyString);
+ EXPECT_TRUE(ctx.IsKeyspaceEventEnabled(kNotifyString));
+ EXPECT_FALSE(ctx.IsKeyspaceEventEnabled(kNotifyGeneric));
+
+ ctx.AddKeyspaceEventIfEnabled(kNotifyGeneric, "del", "tenant", "ignored");
+ EXPECT_FALSE(ctx.HasKeyspaceEvents());
+ ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", "tenant", "key");
+
+ auto events = ctx.TakeKeyspaceEvents();
+ ASSERT_EQ(events.size(), 1);
+ EXPECT_EQ(events[0].channel_flags, kNotifyKeyspace);
+ EXPECT_EQ(events[0].event, "set");
+ EXPECT_EQ(events[0].ns, "tenant");
+ EXPECT_EQ(events[0].key, "key");
+ EXPECT_FALSE(ctx.HasKeyspaceEvents());
+ EXPECT_TRUE(ctx.TakeKeyspaceEvents().empty());
+}
+
+TEST(KeyspaceEvents, ContextCapturesNamespacePerEvent) {
+ auto ctx = engine::Context::NoTransactionContext(nullptr);
+ ctx.EnableKeyspaceEventCollection(kNotifyKeyspace, kNotifyString);
+ ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", "tenant-1", "first");
+ ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", "tenant-2", "second");
+
+ auto events = ctx.TakeKeyspaceEvents();
+ ASSERT_EQ(events.size(), 2);
+ EXPECT_EQ(events[0].ns, "tenant-1");
+ EXPECT_EQ(events[1].ns, "tenant-2");
+}
+
+TEST(KeyspaceEvents, ContextMovePreservesEventOrder) {
+ const auto channel_flags = static_cast<KeyspaceEventChannel>(kNotifyKeyspace
| kNotifyKeyevent);
+ auto ctx = engine::Context::NoTransactionContext(nullptr);
+ ctx.EnableKeyspaceEventCollection(channel_flags, kNotifyAll);
+ ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", "tenant", "first");
+ ctx.AddKeyspaceEventIfEnabled(kNotifyGeneric, "del", "tenant", "second");
+
+ auto moved_ctx = std::move(ctx);
+ auto assigned_ctx = engine::Context::NoTransactionContext(nullptr);
+ assigned_ctx = std::move(moved_ctx);
+
+ auto events = assigned_ctx.TakeKeyspaceEvents();
+ ASSERT_EQ(events.size(), 2);
+ EXPECT_EQ(events[0].channel_flags, channel_flags);
+ EXPECT_EQ(events[0].event, "set");
+ EXPECT_EQ(events[0].ns, "tenant");
+ EXPECT_EQ(events[0].key, "first");
+ EXPECT_EQ(events[1].channel_flags, channel_flags);
+ EXPECT_EQ(events[1].event, "del");
+ EXPECT_EQ(events[1].ns, "tenant");
+ EXPECT_EQ(events[1].key, "second");
+ EXPECT_FALSE(assigned_ctx.HasKeyspaceEvents());
+}
+
+TEST(KeyspaceEvents, ParseFlags) {
+ // Empty disables notifications.
+ auto empty_flags = ParseNotifyKeyspaceEventsFlags("");
+ ASSERT_TRUE(empty_flags.IsOK());
+ EXPECT_EQ(empty_flags->first, kNotifyNoChannel);
+ EXPECT_EQ(empty_flags->second, kNotifyNoType);
+
+ // Channel and event type flags are parsed into independent masks.
+ auto channel_flags = ParseNotifyKeyspaceEventsFlags("KE");
+ ASSERT_TRUE(channel_flags.IsOK());
+ EXPECT_EQ(channel_flags->first, kNotifyKeyspace | kNotifyKeyevent);
+ EXPECT_EQ(channel_flags->second, kNotifyNoType);
+
+ auto type_flags = ParseNotifyKeyspaceEventsFlags("g$");
+ ASSERT_TRUE(type_flags.IsOK());
+ EXPECT_EQ(type_flags->first, kNotifyNoChannel);
+ EXPECT_EQ(type_flags->second, kNotifyGeneric | kNotifyString);
+
+ // KEA enables both channels and set or del.
+ auto flags = ParseNotifyKeyspaceEventsFlags("KEA");
+ ASSERT_TRUE(flags.IsOK());
+ EXPECT_EQ(flags->first, kNotifyKeyspace | kNotifyKeyevent);
+ EXPECT_EQ(flags->second, kNotifyAll);
+}
+
+TEST(KeyspaceEvents, ParseFlagsAExpansion) {
+ auto flags = ParseNotifyKeyspaceEventsFlags("A");
+ ASSERT_TRUE(flags.IsOK());
+ // A expands to all supported event classes without K or E.
+ EXPECT_EQ(flags->first, kNotifyNoChannel);
+ EXPECT_EQ(flags->second, kNotifyAll);
+}
+
+TEST(KeyspaceEvents, ParseFlagsRejectsUnsupported) {
+ // Unsupported flags are rejected.
+ for (const auto *bad : {"a", "d", "x", "e", "m", "n", "o", "c", "l", "s",
"h", "z", "t", "Kx", "KEl", "?"}) {
+ ASSERT_FALSE(ParseNotifyKeyspaceEventsFlags(bad).IsOK()) << "should
reject: " << bad;
+ }
+}
+
+TEST(KeyspaceEvents, FormatKeyspaceNotificationScope) {
+ // Default namespace maps to db 0.
+ EXPECT_EQ(FormatKeyspaceNotificationScope(kDefaultNamespace, 0), "0");
+ EXPECT_EQ(FormatKeyspaceNotificationScope(kDefaultNamespace, 16), "0");
+
+ // Non-default namespaces use their original names.
+ EXPECT_EQ(FormatKeyspaceNotificationScope("0", 0), "0");
+ EXPECT_EQ(FormatKeyspaceNotificationScope("tenantA", 0), "tenantA");
+ EXPECT_EQ(FormatKeyspaceNotificationScope("a.b-c_d", 0), "a.b-c_d");
+ EXPECT_EQ(FormatKeyspaceNotificationScope("db1", 0), "db1");
+ EXPECT_EQ(FormatKeyspaceNotificationScope("a b", 0), "a b");
+ EXPECT_EQ(FormatKeyspaceNotificationScope("a:b", 0), "a:b");
+ EXPECT_EQ(FormatKeyspaceNotificationScope("100%", 0), "100%");
+
+ // Redis database namespaces map back to numeric database names when
redis-databases is enabled.
+ EXPECT_EQ(FormatKeyspaceNotificationScope("db1", 16), "1");
+ EXPECT_EQ(FormatKeyspaceNotificationScope("db15", 16), "15");
+}
diff --git a/tests/gocase/unit/keyspace/keyspace_test.go
b/tests/gocase/unit/keyspace/keyspace_test.go
index 37b86afd1..d67f1f434 100644
--- a/tests/gocase/unit/keyspace/keyspace_test.go
+++ b/tests/gocase/unit/keyspace/keyspace_test.go
@@ -53,6 +53,9 @@ func TestKeyspace(t *testing.T) {
require.NoError(t, rdb.Set(ctx, "foo3", "c", 0).Err())
require.EqualValues(t, 3, rdb.Del(ctx, "foo1", "foo2",
"foo3").Val())
require.Equal(t, []interface{}{nil, nil, nil}, rdb.MGet(ctx,
"foo1", "foo2", "foo3").Val())
+
+ require.NoError(t, rdb.Set(ctx, "foo-dup", "a", 0).Err())
+ require.EqualValues(t, 1, rdb.Del(ctx, "foo-dup",
"foo-dup").Val())
})
t.Run("KEYS with pattern", func(t *testing.T) {
diff --git a/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go
b/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go
new file mode 100644
index 000000000..16ecda745
--- /dev/null
+++ b/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package keyspacenotify
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/apache/kvrocks/tests/gocase/util"
+ "github.com/redis/go-redis/v9"
+ "github.com/stretchr/testify/require"
+)
+
+// drainSubscribeConfirms waits for subscription confirmations.
+func drainSubscribeConfirms(t *testing.T, ctx context.Context, pubsub
*redis.PubSub, n int) {
+ t.Helper()
+ for range n {
+ msg, err := pubsub.ReceiveTimeout(ctx, 2*time.Second)
+ require.NoError(t, err)
+ require.IsType(t, &redis.Subscription{}, msg)
+ }
+}
+
+// expectMessage checks the next pubsub message.
+func expectMessage(t *testing.T, ctx context.Context, pubsub *redis.PubSub,
channel, payload string) {
+ t.Helper()
+ msg, err := pubsub.ReceiveTimeout(ctx, 2*time.Second)
+ require.NoError(t, err)
+ m, ok := msg.(*redis.Message)
+ require.Truef(t, ok, "expected *redis.Message, got %T", msg)
+ require.Equal(t, channel, m.Channel)
+ require.Equal(t, payload, m.Payload)
+}
+
+// expectNoMessage checks that no message arrives soon.
+func expectNoMessage(t *testing.T, ctx context.Context, pubsub *redis.PubSub) {
+ t.Helper()
+ msg, err := pubsub.ReceiveTimeout(ctx, 300*time.Millisecond)
+ require.Errorf(t, err, "expected no message, got %v", msg)
+}
+
+func TestKeyspaceNotify(t *testing.T) {
+ srv := util.StartServer(t, map[string]string{"notify-keyspace-events":
"KEA"})
+ defer srv.Close()
+
+ ctx := context.Background()
+ rdb := srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ // Subscribe to both channel forms.
+ pubsub := rdb.PSubscribe(ctx, "__keyspace@0__:*", "__keyevent@0__:*")
+ defer func() { require.NoError(t, pubsub.Close()) }()
+ drainSubscribeConfirms(t, ctx, pubsub, 2)
+
+ t.Run("SET publishes set", func(t *testing.T) {
+ require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+ // Keyspace is published before keyevent.
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:foo", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "foo")
+ })
+
+ t.Run("SETEX publishes set from the shared Set API", func(t *testing.T)
{
+ require.NoError(t, rdb.Do(ctx, "SETEX", "setex-key", 60,
"value").Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:setex-key", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "setex-key")
+ })
+
+ t.Run("SET NX on existing key publishes nothing", func(t *testing.T) {
+ require.NoError(t, rdb.Set(ctx, "nxkey", "v1", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:nxkey", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "nxkey")
+
+ // NX fails, so nothing is published.
+ require.NoError(t, rdb.SetNX(ctx, "nxkey", "v2", 0).Err())
+ expectNoMessage(t, ctx, pubsub)
+ })
+
+ t.Run("SET GET with conditions publishes only on write", func(t
*testing.T) {
+ require.NoError(t, rdb.Set(ctx, "getnx", "v1", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:getnx", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "getnx")
+
+ cmd := rdb.Do(ctx, "SET", "getnx", "v2", "GET", "NX")
+ require.NoError(t, cmd.Err())
+ require.Equal(t, "v1", cmd.Val())
+ expectNoMessage(t, ctx, pubsub)
+
+ cmd = rdb.Do(ctx, "SET", "getnx-missing", "v1", "GET", "NX")
+ require.ErrorIs(t, cmd.Err(), redis.Nil)
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:getnx-missing",
"set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set",
"getnx-missing")
+
+ cmd = rdb.Do(ctx, "SET", "getxx-missing", "v1", "GET", "XX")
+ require.ErrorIs(t, cmd.Err(), redis.Nil)
+ expectNoMessage(t, ctx, pubsub)
+
+ require.NoError(t, rdb.Set(ctx, "ifeq", "v1", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:ifeq", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "ifeq")
+
+ cmd = rdb.Do(ctx, "SET", "ifeq", "v2", "GET", "IFEQ", "other")
+ require.NoError(t, cmd.Err())
+ require.Equal(t, "v1", cmd.Val())
+ expectNoMessage(t, ctx, pubsub)
+
+ cmd = rdb.Do(ctx, "SET", "ifeq", "v2", "GET", "IFEQ", "v1")
+ require.NoError(t, cmd.Err())
+ require.Equal(t, "v1", cmd.Val())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:ifeq", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "ifeq")
+ })
+
+ t.Run("DEL publishes one del per deleted key", func(t *testing.T) {
+ require.NoError(t, rdb.Set(ctx, "d1", "x", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:d1", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "d1")
+
+ // Only d1 is deleted.
+ require.EqualValues(t, 1, rdb.Del(ctx, "d1", "d2").Val())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:d1", "del")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "d1")
+ expectNoMessage(t, ctx, pubsub)
+
+ require.NoError(t, rdb.Set(ctx, "ddup", "x", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:ddup", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "ddup")
+
+ require.EqualValues(t, 1, rdb.Del(ctx, "ddup", "ddup").Val())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:ddup", "del")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "ddup")
+ expectNoMessage(t, ctx, pubsub)
+ })
+
+ t.Run("UNLINK publishes del", func(t *testing.T) {
+ require.NoError(t, rdb.Set(ctx, "unlink-key", "x", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:unlink-key",
"set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set",
"unlink-key")
+
+ require.EqualValues(t, 1, rdb.Unlink(ctx, "unlink-key").Val())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:unlink-key",
"del")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:del",
"unlink-key")
+ })
+
+ t.Run("Lua nested commands publish events", func(t *testing.T) {
+ script := `
+ redis.call("SET", KEYS[1], "v")
+ redis.call("DEL", KEYS[1])
+ return 1
+ `
+ require.NoError(t, rdb.Eval(ctx, script,
[]string{"lua-key"}).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:lua-key", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "lua-key")
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:lua-key", "del")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "lua-key")
+ expectNoMessage(t, ctx, pubsub)
+ })
+
+ t.Run("MULTI/EXEC publishes queued events after commit", func(t
*testing.T) {
+ _, err := rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error
{
+ pipe.Set(ctx, "m1", "v", 0)
+ pipe.Del(ctx, "m1")
+ return nil
+ })
+ require.NoError(t, err)
+ // Events publish after commit.
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:m1", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "m1")
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:m1", "del")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "m1")
+ expectNoMessage(t, ctx, pubsub)
+ })
+
+ t.Run("MULTI/EXEC preserves per-command notification config", func(t
*testing.T) {
+ require.NoError(t, rdb.ConfigSet(ctx, "notify-keyspace-events",
"E$").Err())
+ _, err := rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error
{
+ pipe.Set(ctx, "event-before-disable", "v", 0)
+ pipe.ConfigSet(ctx, "notify-keyspace-events", "")
+ pipe.Set(ctx, "event-while-disabled", "v", 0)
+ pipe.ConfigSet(ctx, "notify-keyspace-events", "K$")
+ pipe.Set(ctx, "event-after-enable", "v", 0)
+ return nil
+ })
+ require.NoError(t, err)
+
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set",
"event-before-disable")
+ expectMessage(t, ctx, pubsub,
"__keyspace@0__:event-after-enable", "set")
+ expectNoMessage(t, ctx, pubsub)
+ })
+}
+
+func TestKeyspaceNotifyDisabled(t *testing.T) {
+ srv := util.StartServer(t, map[string]string{})
+ defer srv.Close()
+
+ ctx := context.Background()
+ rdb := srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ pubsub := rdb.PSubscribe(ctx, "__keyspace@0__:*", "__keyevent@0__:*")
+ defer func() { require.NoError(t, pubsub.Close()) }()
+ drainSubscribeConfirms(t, ctx, pubsub, 2)
+
+ require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+ require.NoError(t, rdb.Del(ctx, "foo").Err())
+ expectNoMessage(t, ctx, pubsub)
+
+ // CONFIG SET enables notifications at runtime.
+ require.NoError(t, rdb.ConfigSet(ctx, "notify-keyspace-events",
"KEA").Err())
+ require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@0__:foo", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "foo")
+
+ // CONFIG SET rejects unsupported flags.
+ require.Error(t, rdb.ConfigSet(ctx, "notify-keyspace-events",
"Kx").Err())
+ require.Error(t, rdb.ConfigSet(ctx, "notify-keyspace-events",
"KEl").Err())
+}
+
+func TestKeyspaceNotifyRedisDatabases(t *testing.T) {
+ srv := util.StartServer(t, map[string]string{"notify-keyspace-events":
"KEA", "redis-databases": "16"})
+ defer srv.Close()
+
+ ctx := context.Background()
+ sub := srv.NewClient()
+ defer func() { require.NoError(t, sub.Close()) }()
+ writer := srv.NewClient()
+ defer func() { require.NoError(t, writer.Close()) }()
+
+ pubsub := sub.PSubscribe(ctx, "__keyspace@1__:*", "__keyevent@1__:*")
+ defer func() { require.NoError(t, pubsub.Close()) }()
+ drainSubscribeConfirms(t, ctx, pubsub, 2)
+
+ require.NoError(t, writer.Do(ctx, "SELECT", 1).Err())
+ require.NoError(t, writer.Set(ctx, "db-key", "v", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@1__:db-key", "set")
+ expectMessage(t, ctx, pubsub, "__keyevent@1__:set", "db-key")
+
+ require.EqualValues(t, 1, writer.Del(ctx, "db-key").Val())
+ expectMessage(t, ctx, pubsub, "__keyspace@1__:db-key", "del")
+ expectMessage(t, ctx, pubsub, "__keyevent@1__:del", "db-key")
+ expectNoMessage(t, ctx, pubsub)
+}
+
+func TestKeyspaceNotifyNamespace(t *testing.T) {
+ const (
+ adminPassword = "admin-password"
+ namespace = "tenant:1"
+ namespaceToken = "tenant-token"
+ )
+
+ srv := util.StartServer(t, map[string]string{
+ "notify-keyspace-events": "KEA",
+ "requirepass": adminPassword,
+ })
+ defer srv.Close()
+
+ ctx := context.Background()
+ admin := srv.NewClientWithOption(&redis.Options{Password:
adminPassword})
+ defer func() { require.NoError(t, admin.Close()) }()
+ require.NoError(t, admin.Do(ctx, "NAMESPACE", "ADD", namespace,
namespaceToken).Err())
+
+ sub := srv.NewClientWithOption(&redis.Options{Password: namespaceToken})
+ defer func() { require.NoError(t, sub.Close()) }()
+ writer := srv.NewClientWithOption(&redis.Options{Password:
namespaceToken})
+ defer func() { require.NoError(t, writer.Close()) }()
+
+ pubsub := sub.PSubscribe(ctx, "__keyspace@tenant:1__:*",
"__keyevent@tenant:1__:*")
+ defer func() { require.NoError(t, pubsub.Close()) }()
+ drainSubscribeConfirms(t, ctx, pubsub, 2)
+
+ require.NoError(t, writer.Set(ctx, "namespace-key", "v", 0).Err())
+ expectMessage(t, ctx, pubsub, "__keyspace@tenant:1__:namespace-key",
"set")
+ expectMessage(t, ctx, pubsub, "__keyevent@tenant:1__:set",
"namespace-key")
+}
