You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kvrocks.apache.org by hu...@apache.org on 2022/08/17 09:30:58 UTC
[incubator-kvrocks] branch unstable updated: Implement the data structure and partial commands of the Redis stream (#745)
This is an automated email from the ASF dual-hosted git repository.
hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/incubator-kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 73092dc Implement the data structure and partial commands of the Redis stream (#745)
73092dc is described below
commit 73092dc4a5e4f87411133efdcf10b393dd8f9ae9
Author: Yaroslav <to...@gmail.com>
AuthorDate: Wed Aug 17 12:30:53 2022 +0300
Implement the data structure and partial commands of the Redis stream (#745)
The data type Redis stream was introduced in Redis 5.0 as an alternative to Kafka. For more details, see https://redis.io/docs/data-types/streams.
Like the most complex data types, each stream has a metadata key to identify which version is alive and then stores a key-value per entry in a new column family(stream). The reason why we didn't put those entry key-values in the existing column family is that replicas need to replay and notify blocking clients like PubSub.
In case of the command: `XADD s1 1-0 key1 val1 key2 val2`, the ID is 1-0 and the value is `key1 val1 key2 val2`.
In RocksDB entries are represented as key-value pairs where the key is formed as:
`key | version | entry-ID-milliseconds-value | entry-ID-sequence-number-value`
and the value is encoded as:
`key1-length(fixed-32) | key1 | val1-length(fixed-32) | val1 | key2-length(fixed-32) | key2 | val2-length(fixed-32) | val2`.
Available commands:
XADD
XDEL
XINFO STREAM
XLEN
XRANGE
XREAD
XREVRANGE
XTRIM
---
src/redis_cmd.cc | 876 +++++++++++++++
src/redis_db.cc | 5 +-
src/redis_metadata.cc | 75 +-
src/redis_metadata.h | 19 +-
src/redis_stream.cc | 585 ++++++++++
src/redis_stream.h | 67 ++
src/redis_stream_base.cc | 152 +++
src/redis_stream_base.h | 151 +++
src/replication.cc | 11 +
src/replication.h | 1 +
src/server.cc | 72 ++
src/server.h | 20 +
src/storage.cc | 7 +-
src/storage.h | 2 +
tests/cppunit/t_stream_test.cc | 2057 ++++++++++++++++++++++++++++++++++
tests/tcl/tests/support/util.tcl | 16 +
tests/tcl/tests/test_helper.tcl | 1 +
tests/tcl/tests/unit/command.tcl | 4 +-
tests/tcl/tests/unit/type/stream.tcl | 658 +++++++++++
19 files changed, 4768 insertions(+), 11 deletions(-)
diff --git a/src/redis_cmd.cc b/src/redis_cmd.cc
index 6538a98..ba11d20 100644
--- a/src/redis_cmd.cc
+++ b/src/redis_cmd.cc
@@ -45,6 +45,7 @@
#include "redis_pubsub.h"
#include "redis_sortedint.h"
#include "redis_slot.h"
+#include "redis_stream.h"
#include "replication.h"
#include "util.h"
#include "storage.h"
@@ -66,6 +67,11 @@ const char *errWrongNumOfArguments = "wrong number of arguments";
const char *errValueNotInterger = "value is not an integer or out of range";
const char *errAdministorPermissionRequired = "administor permission required to perform the command";
const char *errValueMustBePositive = "value is out of range, must be positive";
+const char *errNoSuchKey = "no such key";
+const char *errUnbalacedStreamList =
+ "Unbalanced XREAD list of streams: for each stream key an ID or '$' must be specified.";
+const char *errTimeoutIsNegative = "timeout is negative";
+const char *errLimitOptionNotAllowed = "syntax error, LIMIT cannot be used without the special ~ option";
class CommandAuth : public Commander {
public:
@@ -4780,6 +4786,867 @@ class CommandScript : public Commander {
std::string subcommand_;
};
+class CommandXAdd : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ bool entry_id_found = false;
+ stream_name_ = args[1];
+
+ for (size_t i = 2; i < args.size(); ) {
+ auto val = Util::ToLower(args[i]);
+
+ if (val == "nomkstream" && !entry_id_found) {
+ nomkstream_ = true;
+ ++i;
+ continue;
+ }
+
+ if (val == "maxlen" && !entry_id_found) {
+ if (i+1 >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ size_t max_len_idx;
+ bool eq_sign_found = false;
+ if (args[i+1] == "=") {
+ max_len_idx = i+2;
+ eq_sign_found = true;
+ } else {
+ max_len_idx = i+1;
+ }
+
+ if (max_len_idx >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ try {
+ max_len_ = std::stoull(args[max_len_idx]);
+ with_max_len_ = true;
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, errValueNotInterger);
+ }
+
+ i += eq_sign_found ? 3 : 2;
+ continue;
+ }
+
+ if (val == "minid" && !entry_id_found) {
+ if (i+1 >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ size_t min_id_idx;
+ bool eq_sign_found = false;
+ if (args[i+1] == "=") {
+ min_id_idx = i+2;
+ eq_sign_found = true;
+ } else {
+ min_id_idx = i+1;
+ }
+
+ if (min_id_idx >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ auto s = ParseStreamEntryID(args[min_id_idx], &min_id_);
+ if (!s.IsOK()) {
+ return Status(Status::RedisParseErr, s.Msg());
+ }
+
+ with_min_id_ = true;
+ i += eq_sign_found ? 3 : 2;
+ continue;
+ }
+
+ if (val == "limit" && !entry_id_found) {
+ return Status(Status::RedisParseErr, errLimitOptionNotAllowed);
+ }
+
+ if (val == "*" && !entry_id_found) {
+ entry_id_found = true;
+ ++i;
+ continue;
+ } else if (!entry_id_found) {
+ auto s = ParseNewStreamEntryID(val, &entry_id_);
+ if (!s.IsOK()) {
+ return Status(Status::RedisParseErr, s.Msg());
+ }
+ entry_id_found = true;
+ with_entry_id_ = true;
+ ++i;
+ continue;
+ }
+
+ if (entry_id_found) {
+ name_value_pairs_.push_back(val);
+ ++i;
+ }
+ }
+
+ if (name_value_pairs_.empty() || name_value_pairs_.size() % 2 != 0) {
+ return Status(Status::RedisParseErr, errWrongNumOfArguments);
+ }
+
+ return Status::OK();
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ Redis::StreamAddOptions options;
+ options.nomkstream = nomkstream_;
+ if (with_max_len_) {
+ options.trim_options.strategy = StreamTrimStrategy::MaxLen;
+ options.trim_options.max_len = max_len_;
+ }
+ if (with_min_id_) {
+ options.trim_options.strategy = StreamTrimStrategy::MinID;
+ options.trim_options.min_id = min_id_;
+ }
+ if (with_entry_id_) {
+ options.with_entry_id = true;
+ options.entry_id = entry_id_;
+ }
+
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+ StreamEntryID entry_id;
+ auto s = stream_db.Add(stream_name_, options, name_value_pairs_, &entry_id);
+ if (!s.ok() && !s.IsNotFound()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ if (s.IsNotFound() && nomkstream_) {
+ *output = Redis::NilString();
+ return Status::OK();
+ }
+
+ *output = Redis::BulkString(entry_id.ToString());
+
+ svr->OnEntryAddedToStream(conn->GetNamespace(), stream_name_, entry_id);
+
+ return Status::OK();
+ }
+
+ private:
+ std::string stream_name_;
+ uint64_t max_len_ = 0;
+ Redis::StreamEntryID min_id_;
+ Redis::NewStreamEntryID entry_id_;
+ std::vector<std::string> name_value_pairs_;
+ bool nomkstream_ = false;
+ bool with_max_len_ = false;
+ bool with_min_id_ = false;
+ bool with_entry_id_ = false;
+};
+
+class CommandXDel : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ for (size_t i = 2; i < args.size(); ++i) {
+ Redis::StreamEntryID id;
+ auto s = ParseStreamEntryID(args[i], &id);
+ if (!s.IsOK()) {
+ return s;
+ }
+ ids_.push_back(id);
+ }
+ return Status::OK();
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+ uint64_t deleted;
+ auto s = stream_db.DeleteEntries(args_[1], ids_, &deleted);
+ if (!s.ok()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ *output = Redis::Integer(deleted);
+
+ return Status::OK();
+ }
+
+ private:
+ std::vector<Redis::StreamEntryID> ids_;
+};
+
+class CommandXLen : public Commander {
+ public:
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+ uint64_t len;
+ auto s = stream_db.Len(args_[1], &len);
+ if (!s.ok()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ *output = Redis::Integer(len);
+
+ return Status::OK();
+ }
+};
+
+class CommandXInfo : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ auto val = Util::ToLower(args[1]);
+ if (val == "stream" && args.size() >= 2) {
+ stream_ = true;
+ if (args.size() > 3 && Util::ToLower(args[3]) == "full") {
+ full_ = true;
+ }
+ if (args.size() > 5 && Util::ToLower(args[4]) == "count") {
+ try {
+ count_ = std::stoull(args[5]);
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, errValueNotInterger);
+ }
+ }
+ }
+ return Status::OK();
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ if (stream_) {
+ return getStreamInfo(svr, conn, output);
+ }
+ return Status::OK();
+ }
+
+ private:
+ uint64_t count_ = 10; // default Redis value
+ bool stream_ = false;
+ bool full_ = false;
+
+ Status getStreamInfo(Server *svr, Connection *conn, std::string *output) {
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+ Redis::StreamInfo info;
+ auto s = stream_db.GetStreamInfo(args_[2], full_, count_, &info);
+ if (!s.ok() && !s.IsNotFound()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ if (s.IsNotFound()) {
+ return Status(Status::RedisExecErr, errNoSuchKey);
+ }
+
+ if (!full_) {
+ output->append(Redis::MultiLen(14));
+ } else {
+ output->append(Redis::MultiLen(12));
+ }
+ output->append(Redis::BulkString("length"));
+ output->append(Redis::Integer(info.size));
+ output->append(Redis::BulkString("last-generated-id"));
+ output->append(Redis::BulkString(info.last_generated_id.ToString()));
+ output->append(Redis::BulkString("max-deleted-entry-id"));
+ output->append(Redis::BulkString(info.max_deleted_entry_id.ToString()));
+ output->append(Redis::BulkString("entries-added"));
+ output->append(Redis::Integer(info.entries_added));
+ output->append(Redis::BulkString("recorded-first-entry-id"));
+ output->append(Redis::BulkString(info.recorded_first_entry_id.ToString()));
+ if (!full_) {
+ output->append(Redis::BulkString("first-entry"));
+ if (info.first_entry) {
+ output->append(Redis::MultiLen(2));
+ output->append(Redis::BulkString(info.first_entry->key));
+ output->append(Redis::MultiBulkString(info.first_entry->values));
+ } else {
+ output->append(Redis::NilString());
+ }
+ output->append(Redis::BulkString("last-entry"));
+ if (info.last_entry) {
+ output->append(Redis::MultiLen(2));
+ output->append(Redis::BulkString(info.last_entry->key));
+ output->append(Redis::MultiBulkString(info.last_entry->values));
+ } else {
+ output->append(Redis::NilString());
+ }
+ } else {
+ output->append(Redis::BulkString("entries"));
+ output->append(Redis::MultiLen(info.entries.size()));
+ for (const auto& e : info.entries) {
+ output->append(Redis::MultiLen(2));
+ output->append(Redis::BulkString(e.key));
+ output->append(Redis::MultiBulkString(e.values));
+ }
+ }
+
+ return Status::OK();
+ }
+};
+
+class CommandXRange : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ stream_name_ = args[1];
+
+ if (args[2] == "-") {
+ start_ = Redis::StreamEntryID::Minimum();
+ } else if (args[2][0] == '(') {
+ exclude_start_ = true;
+ auto s = ParseRangeStart(args[2].substr(1), &start_);
+ if (!s.IsOK()) return s;
+ } else if (args[2] == "+") {
+ start_ = Redis::StreamEntryID::Maximum();
+ } else {
+ auto s = ParseRangeStart(args[2], &start_);
+ if (!s.IsOK()) return s;
+ }
+
+ if (args[3] == "+") {
+ end_ = Redis::StreamEntryID::Maximum();
+ } else if (args[3][0] == '(') {
+ exclude_end_ = true;
+ auto s = ParseRangeEnd(args[3].substr(1), &end_);
+ if (!s.IsOK()) return s;
+ } else if (args[3] == "-") {
+ end_ = Redis::StreamEntryID::Minimum();
+ } else {
+ auto s = ParseRangeEnd(args[3], &end_);
+ if (!s.IsOK()) return s;
+ }
+
+ if (args.size() >= 5 && Util::ToLower(args[4]) == "count") {
+ if (args.size() != 6) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ try {
+ with_count_ = true;
+ count_ = std::stoull(args[5]);
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, errValueNotInterger);
+ }
+ }
+
+ return Status::OK();
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ if (with_count_ && count_ == 0) {
+ *output = Redis::NilString();
+ return Status::OK();
+ }
+
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+
+ Redis::StreamRangeOptions options;
+ options.reverse = false;
+ options.start = start_;
+ options.end = end_;
+ options.with_count = with_count_;
+ options.count = count_;
+ options.exclude_start = exclude_start_;
+ options.exclude_end = exclude_end_;
+
+ std::vector<StreamEntry> result;
+ auto s = stream_db.Range(stream_name_, options, &result);
+ if (!s.ok()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ output->append(Redis::MultiLen(result.size()));
+
+ for (const auto &e : result) {
+ output->append(Redis::MultiLen(2));
+ output->append(Redis::BulkString(e.key));
+ output->append(Redis::MultiBulkString(e.values));
+ }
+
+ return Status::OK();
+ }
+
+ private:
+ std::string stream_name_;
+ StreamEntryID start_;
+ StreamEntryID end_;
+ uint64_t count_ = 0;
+ bool exclude_start_ = false;
+ bool exclude_end_ = false;
+ bool with_count_ = false;
+};
+
+class CommandXRevRange : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ stream_name_ = args[1];
+
+ if (args[2] == "+") {
+ start_ = Redis::StreamEntryID::Maximum();
+ } else if (args[2][0] == '(') {
+ exclude_start_ = true;
+ auto s = ParseRangeEnd(args[2].substr(1), &start_);
+ if (!s.IsOK()) return s;
+ } else if (args[2] == "-") {
+ start_ = Redis::StreamEntryID::Minimum();
+ } else {
+ auto s = ParseRangeEnd(args[2], &start_);
+ if (!s.IsOK()) return s;
+ }
+
+ if (args[3] == "-") {
+ end_ = Redis::StreamEntryID::Minimum();
+ } else if (args[3][0] == '(') {
+ exclude_end_ = true;
+ auto s = ParseRangeStart(args[3].substr(1), &end_);
+ if (!s.IsOK()) return s;
+ } else if (args[3] == "+") {
+ end_ = Redis::StreamEntryID::Maximum();
+ } else {
+ auto s = ParseRangeStart(args[3], &end_);
+ if (!s.IsOK()) return s;
+ }
+
+ if (args.size() >= 5 && Util::ToLower(args[4]) == "count") {
+ if (args.size() != 6) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ try {
+ with_count_ = true;
+ count_ = std::stoull(args[5]);
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, errValueNotInterger);
+ }
+ }
+
+ return Status::OK();
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ if (with_count_ && count_ == 0) {
+ *output = Redis::NilString();
+ return Status::OK();
+ }
+
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+
+ Redis::StreamRangeOptions options;
+ options.reverse = true;
+ options.start = start_;
+ options.end = end_;
+ options.with_count = with_count_;
+ options.count = count_;
+ options.exclude_start = exclude_start_;
+ options.exclude_end = exclude_end_;
+
+ std::vector<StreamEntry> result;
+ auto s = stream_db.Range(stream_name_, options, &result);
+ if (!s.ok()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ output->append(Redis::MultiLen(result.size()));
+
+ for (const auto &e : result) {
+ output->append(Redis::MultiLen(2));
+ output->append(Redis::BulkString(e.key));
+ output->append(Redis::MultiBulkString(e.values));
+ }
+
+ return Status::OK();
+ }
+
+ private:
+ std::string stream_name_;
+ StreamEntryID start_;
+ StreamEntryID end_;
+ uint64_t count_ = 0;
+ bool exclude_start_ = false;
+ bool exclude_end_ = false;
+ bool with_count_ = false;
+};
+
+class CommandXRead : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ size_t streams_word_idx = 0;
+
+ for (size_t i = 1; i < args.size();) {
+ auto arg = Util::ToLower(args[i]);
+
+ if (arg == "streams") {
+ streams_word_idx = i;
+ break;
+ }
+
+ if (arg == "count") {
+ if (i+1 >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ try {
+ with_count_ = true;
+ count_ = static_cast<uint64_t>(std::stoll(args[i+1]));
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, errValueNotInterger);
+ }
+ i += 2;
+ continue;
+ }
+
+ if (arg == "block") {
+ if (i+1 >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ block_ = true;
+ try {
+ auto v = std::stoll(args[i+1]);
+ if (v < 0) {
+ return Status(Status::RedisParseErr, errTimeoutIsNegative);
+ }
+ block_timeout_ = v;
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, errValueNotInterger);
+ }
+ i += 2;
+ continue;
+ }
+
+ ++i;
+ }
+
+ if (streams_word_idx == 0) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ if ((args.size() - streams_word_idx - 1) % 2 != 0) {
+ return Status(Status::RedisParseErr, errUnbalacedStreamList);
+ }
+
+ size_t number_of_streams = (args.size() - streams_word_idx - 1) / 2;
+
+ for (size_t i = streams_word_idx + 1; i <= streams_word_idx + number_of_streams; ++i) {
+ streams_.push_back(args[i]);
+ auto id_str = args[i + number_of_streams];
+ bool get_latest = id_str == "$";
+ latest_marks_.push_back(get_latest);
+ StreamEntryID id;
+ if (!get_latest) {
+ auto s = ParseStreamEntryID(id_str, &id);
+ if (!s.IsOK()) {
+ return s;
+ }
+ }
+ ids_.push_back(id);
+ }
+
+ return Status::OK();
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+
+ std::vector<Redis::StreamReadResult> results;
+
+ for (size_t i = 0; i < streams_.size(); ++i) {
+ if (latest_marks_[i]) {
+ continue;
+ }
+
+ Redis::StreamRangeOptions options;
+ options.reverse = false;
+ options.start = ids_[i];
+ options.end = StreamEntryID{UINT64_MAX, UINT64_MAX};
+ options.with_count = with_count_;
+ options.count = count_;
+ options.exclude_start = true;
+ options.exclude_end = false;
+
+ std::vector<StreamEntry> result;
+ auto s = stream_db.Range(streams_[i], options, &result);
+ if (!s.ok() && !s.IsNotFound()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ if (result.size() > 0) {
+ results.emplace_back(streams_[i], result);
+ }
+ }
+
+ if (block_ && results.empty()) {
+ if (conn->IsInExec()) {
+ *output = Redis::MultiLen(-1);
+ return Status::OK(); // No blocking in multi-exec
+ }
+
+ return blockingRead(svr, conn, &stream_db);
+ }
+
+ if (!block_ && results.empty()) {
+ *output = Redis::MultiLen(-1);
+ return Status::OK();
+ }
+
+ return sendResults(output, results);
+ }
+
+ Status sendResults(std::string *output, const std::vector<StreamReadResult> &results) {
+ output->append(Redis::MultiLen(results.size()));
+
+ for (size_t i = 0; i < results.size(); ++i) {
+ output->append(Redis::MultiLen(2));
+ output->append(Redis::BulkString(results[i].name));
+ output->append(Redis::MultiLen(results[i].entries.size()));
+ for (size_t j = 0; j < results[i].entries.size(); ++j) {
+ output->append(Redis::MultiLen(2));
+ output->append(Redis::BulkString(results[i].entries[j].key));
+ output->append(Redis::MultiBulkString(results[i].entries[j].values));
+ }
+ }
+
+ return Status::OK();
+ }
+
+ Status blockingRead(Server *svr, Connection *conn, Redis::Stream *stream_db) {
+ if (!with_count_) {
+ with_count_ = true;
+ count_ = blocked_default_count_;
+ }
+
+ for (size_t i = 0; i < streams_.size(); ++i) {
+ if (latest_marks_[i]) {
+ StreamEntryID last_generated_id;;
+ auto s = stream_db->GetLastGeneratedID(streams_[i], &last_generated_id);
+ if (!s.ok()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ ids_[i] = last_generated_id;
+ }
+ }
+
+ svr_ = svr;
+ conn_ = conn;
+
+ svr_->BlockOnStreams(streams_, ids_, conn_);
+
+ auto bev = conn->GetBufferEvent();
+ bufferevent_setcb(bev, nullptr, WriteCB, EventCB, this);
+
+ if (block_timeout_ > 0) {
+ timer_ = evtimer_new(bufferevent_get_base(bev), TimerCB, this);
+ timeval tm;
+ if (block_timeout_ > 1000) {
+ tm.tv_sec = block_timeout_ / 1000;
+ tm.tv_usec = (block_timeout_ % 1000) * 1000;
+ } else {
+ tm.tv_sec = 0;
+ tm.tv_usec = block_timeout_ * 1000;
+ }
+
+ evtimer_add(timer_, &tm);
+ }
+
+ return Status::OK();
+ }
+
+ static void WriteCB(bufferevent *bev, void *ctx) {
+ auto command = reinterpret_cast<CommandXRead *>(ctx);
+
+ if (command->timer_ != nullptr) {
+ event_free(command->timer_);
+ command->timer_ = nullptr;
+ }
+
+ command->unblockAll();
+ bufferevent_setcb(bev, Redis::Connection::OnRead, Redis::Connection::OnWrite,
+ Redis::Connection::OnEvent, command->conn_);
+ bufferevent_enable(bev, EV_READ);
+
+ Redis::Stream stream_db(command->svr_->storage_, command->conn_->GetNamespace());
+
+ std::vector<StreamReadResult> results;
+
+ for (size_t i = 0; i < command->streams_.size(); ++i) {
+ Redis::StreamRangeOptions options;
+ options.reverse = false;
+ options.start = command->ids_[i];
+ options.end = StreamEntryID{UINT64_MAX, UINT64_MAX};
+ options.with_count = command->with_count_;
+ options.count = command->count_;
+ options.exclude_start = true;
+ options.exclude_end = false;
+
+ std::vector<StreamEntry> result;
+ auto s = stream_db.Range(command->streams_[i], options, &result);
+ if (!s.ok()) {
+ command->conn_->Reply(Redis::MultiLen(-1));
+ LOG(ERROR) << "ERR executing XRANGE for stream " << command->streams_[i] << " from "
+ << command->ids_[i].ToString() << " to " << options.end.ToString()
+ << " with count " << command->count_ << ": " << s.ToString();
+ }
+
+ if (result.size() > 0) {
+ results.emplace_back(command->streams_[i], result);
+ }
+ }
+
+ if (results.empty()) {
+ command->conn_->Reply(Redis::MultiLen(-1));
+ }
+
+ command->sendReply(results);
+ }
+
+ void sendReply(const std::vector<StreamReadResult> &results) {
+ std::string output;
+
+ output.append(Redis::MultiLen(results.size()));
+
+ for (size_t i = 0; i < results.size(); ++i) {
+ output.append(Redis::MultiLen(2));
+ output.append(Redis::BulkString(results[i].name));
+ output.append(Redis::MultiLen(results[i].entries.size()));
+ for (size_t j = 0; j < results[i].entries.size(); ++j) {
+ output.append(Redis::MultiLen(2));
+ output.append(Redis::BulkString(results[i].entries[j].key));
+ output.append(Redis::MultiBulkString(results[i].entries[j].values));
+ }
+ }
+
+ conn_->Reply(output);
+ }
+
+ static void EventCB(bufferevent *bev, int16_t events, void *ctx) {
+ auto command = static_cast<CommandXRead *>(ctx);
+
+ if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
+ if (command->timer_ != nullptr) {
+ event_free(command->timer_);
+ command->timer_ = nullptr;
+ }
+ command->unblockAll();
+ }
+ Redis::Connection::OnEvent(bev, events, command->conn_);
+ }
+
+ static void TimerCB(int, int16_t events, void *ctx) {
+ auto command = reinterpret_cast<CommandXRead *>(ctx);
+
+ command->conn_->Reply(Redis::NilString());
+
+ event_free(command->timer_);
+ command->timer_ = nullptr;
+
+ command->unblockAll();
+
+ auto bev = command->conn_->GetBufferEvent();
+ bufferevent_setcb(bev, Redis::Connection::OnRead, Redis::Connection::OnWrite,
+ Redis::Connection::OnEvent, command->conn_);
+ bufferevent_enable(bev, EV_READ);
+ }
+
+ private:
+ std::vector<std::string> streams_;
+ std::vector<StreamEntryID> ids_;
+ std::vector<bool> latest_marks_;
+ Server *svr_ = nullptr;
+ Connection *conn_ = nullptr;
+ event *timer_ = nullptr;
+ uint64_t count_ = 0;
+ int64_t block_timeout_ = 0;
+ int blocked_default_count_ = 1000;
+ bool with_count_ = false;
+ bool block_ = false;
+
+ void unblockAll() {
+ svr_->UnblockOnStreams(streams_, conn_);
+ }
+};
+
+class CommandXTrim : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ bool eq_sign_found = false;
+
+ auto trim_strategy = Util::ToLower(args[2]);
+ if (trim_strategy == "maxlen") {
+ strategy_ = StreamTrimStrategy::MaxLen;
+
+ size_t max_len_idx;
+ if (args[3] != "=") {
+ max_len_idx = 3;
+ } else {
+ max_len_idx = 4;
+ eq_sign_found = true;
+ }
+
+ if (max_len_idx >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ try {
+ max_len_ = std::stoull(args[max_len_idx]);
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, errValueNotInterger);
+ }
+ } else if (trim_strategy == "minid") {
+ strategy_ = StreamTrimStrategy::MinID;
+
+ size_t min_id_idx;
+ if (args[3] != "=") {
+ min_id_idx = 3;
+ } else {
+ min_id_idx = 4;
+ eq_sign_found = true;
+ }
+
+ if (min_id_idx >= args.size()) {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ auto s = ParseStreamEntryID(args[min_id_idx], &min_id_);
+ if (!s.IsOK()) {
+ return s;
+ }
+ } else {
+ return Status(Status::RedisParseErr, errInvalidSyntax);
+ }
+
+ bool limit_option_found = false;
+ if (eq_sign_found) {
+ if (args.size() > 6 && Util::ToLower(args[5]) == "limit") {
+ limit_option_found = true;
+ }
+ } else {
+ if (args.size() > 5 && Util::ToLower(args[4]) == "limit") {
+ limit_option_found = true;
+ }
+ }
+
+ if (limit_option_found) {
+ return Status(Status::RedisParseErr, errLimitOptionNotAllowed);
+ }
+
+ return Status::OK();
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ Redis::Stream stream_db(svr->storage_, conn->GetNamespace());
+
+ StreamTrimOptions options;
+ options.strategy = strategy_;
+ options.max_len = max_len_;
+ options.min_id = min_id_;
+
+ uint64_t removed;
+ auto s = stream_db.Trim(args_[1], options, &removed);
+ if (!s.ok()) {
+ return Status(Status::RedisExecErr, s.ToString());
+ }
+
+ *output = Redis::Integer(removed);
+
+ return Status::OK();
+ }
+
+ private:
+ uint64_t max_len_ = 0;
+ StreamEntryID min_id_;
+ StreamTrimStrategy strategy_ = StreamTrimStrategy::None;
+};
+
#define ADD_CMD(name, arity, description , first_key, last_key, key_step, fn) \
{name, arity, description, 0, first_key, last_key, key_step, []() -> std::unique_ptr<Commander> { \
return std::unique_ptr<Commander>(new fn()); \
@@ -4973,6 +5840,15 @@ CommandAttributes redisCommandTable[] = {
ADD_CMD("_fetch_meta", 1, "read-only replication no-multi no-script", 0, 0, 0, CommandFetchMeta),
ADD_CMD("_fetch_file", 2, "read-only replication no-multi no-script", 0, 0, 0, CommandFetchFile),
ADD_CMD("_db_name", 1, "read-only replication no-multi", 0, 0, 0, CommandDBName),
+
+ ADD_CMD("xadd", -5, "write", 1, 1, 1, CommandXAdd),
+ ADD_CMD("xdel", -3, "write", 1, 1, 1, CommandXDel),
+ ADD_CMD("xlen", 2, "read-only", 1, 1, 1, CommandXLen),
+ ADD_CMD("xinfo", -2, "read-only", 0, 0, 0, CommandXInfo),
+ ADD_CMD("xrange", -4, "read-only", 1, 1, 1, CommandXRange),
+ ADD_CMD("xrevrange", -2, "read-only", 0, 0, 0, CommandXRevRange),
+ ADD_CMD("xread", -4, "read-only", 0, 0, 0, CommandXRead),
+ ADD_CMD("xtrim", -4, "write", 1, 1, 1, CommandXTrim),
};
// Command table after rename-command directive
diff --git a/src/redis_db.cc b/src/redis_db.cc
index df7538c..8f86c9a 100644
--- a/src/redis_db.cc
+++ b/src/redis_db.cc
@@ -48,11 +48,12 @@ rocksdb::Status Database::GetMetadata(RedisType type, const Slice &ns_key, Metad
metadata->Decode(old_metadata);
return rocksdb::Status::NotFound(kErrMsgKeyExpired);
}
- if (metadata->Type() != type && (metadata->size > 0 || metadata->Type() == kRedisString)) {
+ if (metadata->Type() != type
+ && (metadata->size > 0 || metadata->Type() == kRedisString || metadata->Type() == kRedisStream)) {
metadata->Decode(old_metadata);
return rocksdb::Status::InvalidArgument(kErrMsgWrongType);
}
- if (metadata->size == 0) {
+ if (metadata->size == 0 && type != kRedisStream) { // stream is allowed to be empty
metadata->Decode(old_metadata);
return rocksdb::Status::NotFound("no elements");
}
diff --git a/src/redis_metadata.cc b/src/redis_metadata.cc
index 9bed78d..f46c049 100644
--- a/src/redis_metadata.cc
+++ b/src/redis_metadata.cc
@@ -35,6 +35,7 @@ static std::atomic<uint64_t> version_counter_ = {0};
const char* kErrMsgWrongType = "WRONGTYPE Operation against a key holding the wrong kind of value";
const char* kErrMsgKeyExpired = "the key was expired";
+const char* kErrMetadataTooShort = "metadata is too short";
InternalKey::InternalKey(Slice input, bool slot_id_encoded) {
slot_id_encoded_ = slot_id_encoded;
@@ -176,13 +177,13 @@ Metadata::Metadata(RedisType type, bool generate_version) {
rocksdb::Status Metadata::Decode(const std::string &bytes) {
// flags(1byte) + expire (4byte)
if (bytes.size() < 5) {
- return rocksdb::Status::InvalidArgument("the metadata was too short");
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
}
Slice input(bytes);
GetFixed8(&input, &flags);
GetFixed32(&input, reinterpret_cast<uint32_t *>(&expire));
if (Type() != kRedisString) {
- if (input.size() < 12) rocksdb::Status::InvalidArgument("the metadata was too short");
+ if (input.size() < 12) rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
GetFixed64(&input, &version);
GetFixed32(&input, &size);
}
@@ -249,12 +250,14 @@ timeval Metadata::Time() const {
}
bool Metadata::Expired() const {
- if (Type() != kRedisString && size == 0) {
+ if (Type() != kRedisString && Type() != kRedisStream && size == 0) {
return true;
}
+
if (expire <= 0) {
return false;
}
+
int64_t now;
rocksdb::Env::Default()->GetCurrentTime(&now);
return expire < now;
@@ -276,14 +279,76 @@ rocksdb::Status ListMetadata::Decode(const std::string &bytes) {
GetFixed8(&input, &flags);
GetFixed32(&input, reinterpret_cast<uint32_t *>(&expire));
if (Type() != kRedisString) {
- if (input.size() < 12) rocksdb::Status::InvalidArgument("the metadata was too short");
+ if (input.size() < 12) rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
GetFixed64(&input, &version);
GetFixed32(&input, &size);
}
if (Type() == kRedisList) {
- if (input.size() < 16) rocksdb::Status::InvalidArgument("the metadata was too short");
+ if (input.size() < 16) rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
GetFixed64(&input, &head);
GetFixed64(&input, &tail);
}
return rocksdb::Status::OK();
}
+
+void StreamMetadata::Encode(std::string *dst) {
+ Metadata::Encode(dst);
+
+ PutFixed64(dst, last_generated_id.ms);
+ PutFixed64(dst, last_generated_id.seq);
+
+ PutFixed64(dst, recorded_first_entry_id.ms);
+ PutFixed64(dst, recorded_first_entry_id.seq);
+
+ PutFixed64(dst, max_deleted_entry_id.ms);
+ PutFixed64(dst, max_deleted_entry_id.seq);
+
+ PutFixed64(dst, first_entry_id.ms);
+ PutFixed64(dst, first_entry_id.seq);
+
+ PutFixed64(dst, last_entry_id.ms);
+ PutFixed64(dst, last_entry_id.seq);
+
+ PutFixed64(dst, entries_added);
+}
+
+rocksdb::Status StreamMetadata::Decode(const std::string &bytes) {
+ // flags(1byte) + expire (4byte)
+ if (bytes.size() < 5) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ Slice input(bytes);
+ GetFixed8(&input, &flags);
+ GetFixed32(&input, reinterpret_cast<uint32_t *>(&expire));
+
+ if (input.size() < 12) {
+ rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ GetFixed64(&input, &version);
+ GetFixed32(&input, &size);
+
+ if (input.size() < 88) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ GetFixed64(&input, &last_generated_id.ms);
+ GetFixed64(&input, &last_generated_id.seq);
+
+ GetFixed64(&input, &recorded_first_entry_id.ms);
+ GetFixed64(&input, &recorded_first_entry_id.seq);
+
+ GetFixed64(&input, &max_deleted_entry_id.ms);
+ GetFixed64(&input, &max_deleted_entry_id.seq);
+
+ GetFixed64(&input, &first_entry_id.ms);
+ GetFixed64(&input, &first_entry_id.seq);
+
+ GetFixed64(&input, &last_entry_id.ms);
+ GetFixed64(&input, &last_entry_id.seq);
+
+ GetFixed64(&input, &entries_added);
+
+ return rocksdb::Status::OK();
+}
diff --git a/src/redis_metadata.h b/src/redis_metadata.h
index 57d6695..f7a6f32 100644
--- a/src/redis_metadata.h
+++ b/src/redis_metadata.h
@@ -27,6 +27,7 @@
#include <rocksdb/status.h>
#include "encoding.h"
+#include "redis_stream_base.h"
enum RedisType {
kRedisNone,
@@ -37,6 +38,7 @@ enum RedisType {
kRedisZSet,
kRedisBitmap,
kRedisSortedint,
+ kRedisStream,
};
enum RedisCommand {
@@ -56,7 +58,7 @@ enum RedisCommand {
const std::vector<std::string> RedisTypeNames = {
"none", "string", "hash",
- "list", "set", "zset", "bitmap", "sortedint"
+ "list", "set", "zset", "bitmap", "sortedint", "stream"
};
extern const char* kErrMsgWrongType;
@@ -156,3 +158,18 @@ class ListMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class StreamMetadata : public Metadata {
+ public:
+ Redis::StreamEntryID last_generated_id;
+ Redis::StreamEntryID recorded_first_entry_id;
+ Redis::StreamEntryID max_deleted_entry_id;
+ Redis::StreamEntryID first_entry_id;
+ Redis::StreamEntryID last_entry_id;
+ uint64_t entries_added = 0;
+
+ explicit StreamMetadata(bool generate_version = true) : Metadata(kRedisStream, generate_version) {}
+ public:
+ void Encode(std::string *dst) override;
+ rocksdb::Status Decode(const std::string &bytes) override;
+};
diff --git a/src/redis_stream.cc b/src/redis_stream.cc
new file mode 100644
index 0000000..6c1dab3
--- /dev/null
+++ b/src/redis_stream.cc
@@ -0,0 +1,585 @@
+/*
+ * 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 "redis_stream.h"
+
+#include <memory>
+
+#include <glog/logging.h>
+#include <rocksdb/status.h>
+
+#include "db_util.h"
+
+
+namespace Redis {
+
+rocksdb::Status Stream::GetMetadata(const Slice &stream_name, StreamMetadata *metadata) {
+ return Database::GetMetadata(kRedisStream, stream_name, metadata);
+}
+
+rocksdb::Status Stream::GetLastGeneratedID(const Slice &stream_name, StreamEntryID *id) {
+ std::string ns_key;
+ AppendNamespacePrefix(stream_name, &ns_key);
+
+ StreamMetadata metadata;
+ rocksdb::Status s = GetMetadata(ns_key, &metadata);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+
+ if (s.IsNotFound()) {
+ id->ms = 0;
+ id->seq = 0;
+ } else {
+ *id = metadata.last_generated_id;
+ }
+
+ return rocksdb::Status::OK();
+}
+
+StreamEntryID Stream::entryIDFromInternalKey(const rocksdb::Slice &key) const {
+ InternalKey ikey(key, storage_->IsSlotIdEncoded());
+ Slice entry_id = ikey.GetSubKey();
+ StreamEntryID id;
+ GetFixed64(&entry_id, &id.ms);
+ GetFixed64(&entry_id, &id.seq);
+ return id;
+}
+
+std::string Stream::internalKeyFromEntryID(const std::string &ns_key, const StreamMetadata &metadata,
+ const StreamEntryID &id) const {
+ std::string sub_key;
+ PutFixed64(&sub_key, id.ms);
+ PutFixed64(&sub_key, id.seq);
+ std::string entry_key;
+ InternalKey(ns_key, sub_key, metadata.version, storage_->IsSlotIdEncoded()).Encode(&entry_key);
+ return entry_key;
+}
+
+rocksdb::Status Stream::Add(const Slice &stream_name, const StreamAddOptions& options,
+ const std::vector<std::string> &args, StreamEntryID *id) {
+ for (auto const &v : args) {
+ if (v.size() > INT32_MAX) {
+ rocksdb::Status::InvalidArgument("argument length is too high");
+ }
+ }
+
+ std::string entry_value = EncodeStreamEntryValue(args);
+
+ std::string ns_key;
+ AppendNamespacePrefix(stream_name, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata;
+ rocksdb::Status s = GetMetadata(ns_key, &metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+
+ if (s.IsNotFound() && options.nomkstream) {
+ return s;
+ }
+
+ bool first_entry = s.IsNotFound();
+
+ StreamEntryID next_entry_id;
+ s = getNextEntryID(metadata, options, first_entry, &next_entry_id);
+ if (!s.ok()) return s;
+
+ rocksdb::WriteBatch batch;
+ WriteBatchLogData log_data(kRedisStream);
+ batch.PutLogData(log_data.Encode());
+
+ bool should_add = true;
+
+ // trim the stream before adding a new entry to provide atomic XADD + XTRIM
+ if (options.trim_options.strategy != StreamTrimStrategy::None) {
+ StreamTrimOptions trim_options = options.trim_options;
+ if (trim_options.strategy == StreamTrimStrategy::MaxLen) {
+ // because one entry will be added, we can trim up to (MAXLEN-1) if MAXLEN was specified
+ trim_options.max_len = options.trim_options.max_len > 0 ? options.trim_options.max_len - 1 : 0;
+ }
+
+ trim(ns_key, trim_options, &metadata, &batch);
+
+ if (trim_options.strategy == StreamTrimStrategy::MinID && next_entry_id < trim_options.min_id) {
+ // there is no sense to add this element because it would be removed, so just modify metadata and return it's ID
+ should_add = false;
+ }
+
+ if (trim_options.strategy == StreamTrimStrategy::MaxLen && options.trim_options.max_len == 0) {
+ // there is no sense to add this element because it would be removed, so just modify metadata and return it's ID
+ should_add = false;
+ }
+ }
+
+ if (should_add) {
+ std::string entry_key = internalKeyFromEntryID(ns_key, metadata, next_entry_id);
+ batch.Put(stream_cf_handle_, entry_key, entry_value);
+
+ metadata.last_generated_id = next_entry_id;
+ metadata.last_entry_id = next_entry_id;
+ metadata.size += 1;
+
+ if (metadata.size == 1) {
+ metadata.first_entry_id = next_entry_id;
+ metadata.recorded_first_entry_id = next_entry_id;
+ }
+ } else {
+ metadata.last_generated_id = next_entry_id;
+ metadata.max_deleted_entry_id = next_entry_id;
+ }
+
+ metadata.entries_added += 1;
+
+ std::string metadataBytes;
+ metadata.Encode(&metadataBytes);
+ batch.Put(metadata_cf_handle_, ns_key, metadataBytes);
+
+ *id = next_entry_id;
+
+ return storage_->Write(rocksdb::WriteOptions(), &batch);
+}
+
+rocksdb::Status Stream::getNextEntryID(const StreamMetadata &metadata, const StreamAddOptions &options,
+ bool first_entry, StreamEntryID *next_entry_id) const {
+ if (options.with_entry_id) {
+ if (options.entry_id.ms == 0 && !options.entry_id.any_seq_number && options.entry_id.seq == 0) {
+ return rocksdb::Status::InvalidArgument("The ID specified in XADD must be greater than 0-0");
+ }
+
+ if (metadata.last_generated_id.ms == UINT64_MAX && metadata.last_generated_id.seq == UINT64_MAX) {
+ return rocksdb::Status::InvalidArgument(
+ "The stream has exhausted the last possible ID, unable to add more items");
+ }
+
+ if (!first_entry) {
+ if (metadata.last_generated_id.ms > options.entry_id.ms) {
+ return rocksdb::Status::InvalidArgument(
+ "The ID specified in XADD is equal or smaller than the target stream top item");
+ }
+
+ if (metadata.last_generated_id.ms == options.entry_id.ms) {
+ if (!options.entry_id.any_seq_number && metadata.last_generated_id.seq >= options.entry_id.seq) {
+ return rocksdb::Status::InvalidArgument(
+ "The ID specified in XADD is equal or smaller than the target stream top item");
+ }
+
+ if (options.entry_id.any_seq_number && metadata.last_generated_id.seq == UINT64_MAX) {
+ return rocksdb::Status::InvalidArgument(
+ "Elements are too large to be stored"); // Redis responds with exactly this message
+ }
+ }
+
+ if (options.entry_id.any_seq_number) {
+ if (options.entry_id.ms == metadata.last_generated_id.ms) {
+ next_entry_id->seq = metadata.last_generated_id.seq + 1;
+ } else {
+ next_entry_id->seq = 0;
+ }
+ } else {
+ next_entry_id->seq = options.entry_id.seq;
+ }
+ } else {
+ if (options.entry_id.any_seq_number) {
+ next_entry_id->seq = options.entry_id.ms != 0 ? 0 : 1;
+ } else {
+ next_entry_id->seq = options.entry_id.seq;
+ }
+ }
+ next_entry_id->ms = options.entry_id.ms;
+ return rocksdb::Status::OK();
+ } else {
+ return GetNextStreamEntryID(metadata.last_generated_id, next_entry_id);
+ }
+}
+
+rocksdb::Status Stream::DeleteEntries(const rocksdb::Slice &stream_name,
+ const std::vector<StreamEntryID> &ids, uint64_t *ret) {
+ *ret = 0;
+
+ std::string ns_key;
+ AppendNamespacePrefix(stream_name, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(ns_key, &metadata);
+ if (!s.ok()) {
+ return s.IsNotFound() ? rocksdb::Status::OK() : s;
+ }
+
+ rocksdb::WriteBatch batch;
+ WriteBatchLogData log_data(kRedisStream);
+ batch.PutLogData(log_data.Encode());
+
+ std::string next_version_prefix_key;
+ InternalKey(ns_key, "", metadata.version + 1, storage_->IsSlotIdEncoded()).Encode(&next_version_prefix_key);
+ std::string prefix_key;
+ InternalKey(ns_key, "", metadata.version, storage_->IsSlotIdEncoded()).Encode(&prefix_key);
+
+ rocksdb::ReadOptions read_options;
+ LatestSnapShot ss(db_);
+ read_options.snapshot = ss.GetSnapShot();
+ rocksdb::Slice upper_bound(next_version_prefix_key);
+ read_options.iterate_upper_bound = &upper_bound;
+ rocksdb::Slice lower_bound(prefix_key);
+ read_options.iterate_lower_bound = &lower_bound;
+ read_options.fill_cache = false;
+
+ auto iter = DBUtil::UniqueIterator(db_, read_options, stream_cf_handle_);
+
+ for (const auto& id : ids) {
+ std::string entry_key = internalKeyFromEntryID(ns_key, metadata, id);
+ std::string value;
+ s = db_->Get(read_options, stream_cf_handle_, entry_key, &value);
+ if (s.ok()) {
+ *ret += 1;
+ batch.Delete(stream_cf_handle_, entry_key);
+
+ if (metadata.max_deleted_entry_id < id) {
+ metadata.max_deleted_entry_id = id;
+ }
+
+ if (*ret == metadata.size) {
+ metadata.first_entry_id.Clear();
+ metadata.last_entry_id.Clear();
+ metadata.recorded_first_entry_id.Clear();
+ break;
+ }
+
+ if (id == metadata.first_entry_id) {
+ iter->Seek(entry_key);
+ iter->Next();
+ if (iter->Valid()) {
+ metadata.first_entry_id = entryIDFromInternalKey(iter->key());
+ metadata.recorded_first_entry_id = metadata.first_entry_id;
+ } else {
+ metadata.first_entry_id.Clear();
+ metadata.recorded_first_entry_id.Clear();
+ }
+ }
+
+ if (id == metadata.last_entry_id) {
+ iter->Seek(entry_key);
+ iter->Prev();
+ if (iter->Valid()) {
+ metadata.last_entry_id = entryIDFromInternalKey(iter->key());
+ } else {
+ metadata.last_entry_id.Clear();
+ }
+ }
+ }
+ }
+
+ if (*ret > 0) {
+ metadata.size -= *ret;
+
+ std::string bytes;
+ metadata.Encode(&bytes);
+ batch.Put(metadata_cf_handle_, ns_key, bytes);
+ }
+
+ return storage_->Write(rocksdb::WriteOptions(), &batch);
+}
+
+rocksdb::Status Stream::Len(const rocksdb::Slice &stream_name, uint64_t *ret) {
+ *ret = 0;
+ std::string ns_key;
+ AppendNamespacePrefix(stream_name, &ns_key);
+
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(ns_key, &metadata);
+ if (!s.ok()) {
+ return s.IsNotFound() ? rocksdb::Status::OK() : s;
+ }
+
+ *ret = metadata.size;
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Stream::range(const std::string &ns_key, const StreamMetadata &metadata,
+ const StreamRangeOptions &options, std::vector<StreamEntry> *entries) const {
+ std::string start_key = internalKeyFromEntryID(ns_key, metadata, options.start);
+ std::string end_key = internalKeyFromEntryID(ns_key, metadata, options.end);
+
+ if (start_key == end_key) {
+ if (options.exclude_start || options.exclude_end) {
+ return rocksdb::Status::OK();
+ }
+
+ std::string value;
+ auto s = db_->Get(rocksdb::ReadOptions(), stream_cf_handle_, start_key, &value);
+ if (!s.ok()) {
+ return s.IsNotFound() ? rocksdb::Status::OK() : s;
+ }
+
+ entries->emplace_back(options.start.ToString(), DecodeRawStreamEntryValue(value));
+ return rocksdb::Status::OK();
+ }
+
+ if ((!options.reverse && options.end < options.start)
+ || (options.reverse && options.start < options.end)) {
+ return rocksdb::Status::OK();
+ }
+
+ std::string next_version_prefix_key;
+ InternalKey(ns_key, "", metadata.version + 1, storage_->IsSlotIdEncoded()).Encode(&next_version_prefix_key);
+ std::string prefix_key;
+ InternalKey(ns_key, "", metadata.version, storage_->IsSlotIdEncoded()).Encode(&prefix_key);
+
+ rocksdb::ReadOptions read_options;
+ LatestSnapShot ss(db_);
+ read_options.snapshot = ss.GetSnapShot();
+ rocksdb::Slice upper_bound(next_version_prefix_key);
+ read_options.iterate_upper_bound = &upper_bound;
+ rocksdb::Slice lower_bound(prefix_key);
+ read_options.iterate_lower_bound = &lower_bound;
+ read_options.fill_cache = false;
+
+ auto iter = DBUtil::UniqueIterator(db_, read_options, stream_cf_handle_);
+ iter->Seek(start_key);
+ if (options.reverse && (!iter->Valid() || iter->key().ToString() != start_key)) {
+ iter->SeekForPrev(start_key);
+ }
+
+ for (;
+ iter->Valid() && (options.reverse ? iter->key().ToString() >= end_key : iter->key().ToString() <= end_key);
+ options.reverse ? iter->Prev() : iter->Next()) {
+ if (options.exclude_start && iter->key().ToString() == start_key) {
+ continue;
+ }
+
+ if (options.exclude_end && iter->key().ToString() == end_key) {
+ break;
+ }
+
+ entries->emplace_back(entryIDFromInternalKey(iter->key()).ToString(),
+ DecodeRawStreamEntryValue(iter->value().ToString()));
+
+ if (options.with_count && entries->size() == options.count) {
+ break;
+ }
+ }
+
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Stream::getEntryRawValue(const std::string &ns_key, const StreamMetadata &metadata,
+ const StreamEntryID &id, std::string *value) const {
+ std::string entry_key = internalKeyFromEntryID(ns_key, metadata, id);
+ return db_->Get(rocksdb::ReadOptions(), stream_cf_handle_, entry_key, value);
+}
+
+rocksdb::Status Stream::GetStreamInfo(const rocksdb::Slice &stream_name, bool full,
+ uint64_t count, StreamInfo *info) {
+ std::string ns_key;
+ AppendNamespacePrefix(stream_name, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(ns_key, &metadata);
+ if (!s.ok()) return s;
+
+ info->size = metadata.size;
+ info->entries_added = metadata.entries_added;
+ info->last_generated_id = metadata.last_generated_id;
+ info->max_deleted_entry_id = metadata.max_deleted_entry_id;
+ info->recorded_first_entry_id = metadata.recorded_first_entry_id;
+
+ if (metadata.size == 0) {
+ return rocksdb::Status::OK();
+ }
+
+ if (full) {
+ uint64_t need_entries = metadata.size;
+ if (count != 0 && count < metadata.size) {
+ need_entries = count;
+ }
+
+ info->entries.reserve(need_entries);
+
+ StreamRangeOptions options;
+ options.start = metadata.first_entry_id;
+ options.end = metadata.last_entry_id;
+ options.with_count = true;
+ options.count = need_entries;
+
+ s = range(ns_key, metadata, options, &info->entries);
+ if (!s.ok()) {
+ return s;
+ }
+ } else {
+ std::string first_value;
+ s = getEntryRawValue(ns_key, metadata, metadata.first_entry_id, &first_value);
+ if (!s.ok()) {
+ return s;
+ }
+ info->first_entry = std::unique_ptr<StreamEntry>(
+ new StreamEntry{metadata.first_entry_id.ToString(), DecodeRawStreamEntryValue(first_value)});
+
+ std::string last_value;
+ s = getEntryRawValue(ns_key, metadata, metadata.last_entry_id, &last_value);
+ if (!s.ok()) {
+ return s;
+ }
+ info->last_entry = std::unique_ptr<StreamEntry>(
+ new StreamEntry{metadata.last_entry_id.ToString(), DecodeRawStreamEntryValue(last_value)});
+ }
+
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Stream::Range(const Slice &stream_name, const StreamRangeOptions &options,
+ std::vector<StreamEntry> *entries) {
+ entries->clear();
+
+ if (options.with_count && options.count == 0) {
+ return rocksdb::Status::OK();
+ }
+
+ if (options.exclude_start && options.start.IsMaximum()) {
+ return rocksdb::Status::InvalidArgument("invalid start ID for the interval");
+ }
+
+ if (options.exclude_end && options.end.IsMinimum()) {
+ return rocksdb::Status::InvalidArgument("invalid end ID for the interval");
+ }
+
+ std::string ns_key;
+ AppendNamespacePrefix(stream_name, &ns_key);
+
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(ns_key, &metadata);
+ if (!s.ok()) {
+ return s.IsNotFound() ? rocksdb::Status::OK() : s;
+ }
+
+ return range(ns_key, metadata, options, entries);
+}
+
+rocksdb::Status Stream::Trim(const rocksdb::Slice &stream_name, const StreamTrimOptions &options, uint64_t *ret) {
+ *ret = 0;
+
+ if (options.strategy == StreamTrimStrategy::None) {
+ return rocksdb::Status::OK();
+ }
+
+ std::string ns_key;
+ AppendNamespacePrefix(stream_name, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(ns_key, &metadata);
+ if (!s.ok()) {
+ return s.IsNotFound() ? rocksdb::Status::OK() : s;
+ }
+
+ rocksdb::WriteBatch batch;
+ WriteBatchLogData log_data(kRedisStream);
+ batch.PutLogData(log_data.Encode());
+
+ *ret = trim(ns_key, options, &metadata, &batch);
+
+ if (*ret > 0) {
+ std::string bytes;
+ metadata.Encode(&bytes);
+ batch.Put(metadata_cf_handle_, ns_key, bytes);
+
+ return storage_->Write(rocksdb::WriteOptions(), &batch);
+ }
+
+ return rocksdb::Status::OK();
+}
+
+uint64_t Stream::trim(const std::string &ns_key, const StreamTrimOptions &options,
+ StreamMetadata *metadata, rocksdb::WriteBatch *batch) {
+ if (metadata->size == 0) {
+ return 0;
+ }
+
+ if (options.strategy == StreamTrimStrategy::MaxLen && metadata->size <= options.max_len) {
+ return 0;
+ }
+
+ if (options.strategy == StreamTrimStrategy::MinID && metadata->first_entry_id >= options.min_id) {
+ return 0;
+ }
+
+ uint64_t ret = 0;
+
+ std::string next_version_prefix_key;
+ InternalKey(ns_key, "", metadata->version + 1, storage_->IsSlotIdEncoded()).Encode(&next_version_prefix_key);
+ std::string prefix_key;
+ InternalKey(ns_key, "", metadata->version, storage_->IsSlotIdEncoded()).Encode(&prefix_key);
+
+ rocksdb::ReadOptions read_options;
+ LatestSnapShot ss(db_);
+ read_options.snapshot = ss.GetSnapShot();
+ rocksdb::Slice upper_bound(next_version_prefix_key);
+ read_options.iterate_upper_bound = &upper_bound;
+ rocksdb::Slice lower_bound(prefix_key);
+ read_options.iterate_lower_bound = &lower_bound;
+ read_options.fill_cache = false;
+
+ auto iter = DBUtil::UniqueIterator(db_, read_options, stream_cf_handle_);
+ std::string start_key = internalKeyFromEntryID(ns_key, *metadata, metadata->first_entry_id);
+ iter->Seek(start_key);
+
+ std::string last_deleted;
+ while (iter->Valid() && metadata->size > 0) {
+ if (options.strategy == StreamTrimStrategy::MaxLen && metadata->size <= options.max_len) {
+ break;
+ }
+
+ if (options.strategy == StreamTrimStrategy::MinID && metadata->first_entry_id >= options.min_id) {
+ break;
+ }
+
+ batch->Delete(stream_cf_handle_, iter->key());
+
+ ret += 1;
+ metadata->size -= 1;
+ last_deleted = iter->key().ToString();
+
+ iter->Next();
+
+ if (iter->Valid()) {
+ metadata->first_entry_id = entryIDFromInternalKey(iter->key());
+ metadata->recorded_first_entry_id = metadata->first_entry_id;
+ } else {
+ metadata->first_entry_id.Clear();
+ metadata->recorded_first_entry_id.Clear();
+ }
+ }
+
+ if (metadata->size == 0) {
+ metadata->first_entry_id.Clear();
+ metadata->last_entry_id.Clear();
+ metadata->recorded_first_entry_id.Clear();
+ }
+
+ if (ret > 0) {
+ metadata->max_deleted_entry_id = entryIDFromInternalKey(last_deleted);
+ }
+
+ return ret;
+}
+
+
+} // namespace Redis
diff --git a/src/redis_stream.h b/src/redis_stream.h
new file mode 100644
index 0000000..a9bd072
--- /dev/null
+++ b/src/redis_stream.h
@@ -0,0 +1,67 @@
+/*
+ * 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 <vector>
+
+#include <rocksdb/status.h>
+
+#include "redis_db.h"
+#include "redis_metadata.h"
+
+using rocksdb::Slice;
+
+namespace Redis {
+
+class Stream : public SubKeyScanner {
+ public:
+ explicit Stream(Engine::Storage *storage, const std::string &ns) :
+ SubKeyScanner(storage, ns),
+ stream_cf_handle_(storage->GetCFHandle("stream")) {}
+ rocksdb::Status Add(const Slice &stream_name, const StreamAddOptions &options,
+ const std::vector<std::string> &values, StreamEntryID *id);
+ rocksdb::Status DeleteEntries(const Slice &stream_name, const std::vector<StreamEntryID> &ids, uint64_t *ret);
+ rocksdb::Status Len(const Slice &stream_name, uint64_t *ret);
+ rocksdb::Status GetStreamInfo(const Slice &stream_name, bool full, uint64_t count, StreamInfo *info);
+ rocksdb::Status Range(const Slice &stream_name, const StreamRangeOptions &options,
+ std::vector<StreamEntry> *entries);
+ rocksdb::Status Trim(const Slice &stream_name, const StreamTrimOptions &options, uint64_t *ret);
+ rocksdb::Status GetMetadata(const Slice &stream_name, StreamMetadata *metadata);
+ rocksdb::Status GetLastGeneratedID(const Slice &stream_name, StreamEntryID *id);
+
+ private:
+ rocksdb::ColumnFamilyHandle *stream_cf_handle_;
+
+ rocksdb::Status range(const std::string &ns_key, const StreamMetadata &metadata,
+ const StreamRangeOptions &options, std::vector<StreamEntry> *entries) const;
+ rocksdb::Status getEntryRawValue(const std::string &ns_key, const StreamMetadata &metadata,
+ const StreamEntryID &id, std::string *value) const;
+ StreamEntryID entryIDFromInternalKey(const rocksdb::Slice &key) const;
+ std::string internalKeyFromEntryID(const std::string &ns_key, const StreamMetadata &metadata,
+ const StreamEntryID &id) const;
+ rocksdb::Status getNextEntryID(const StreamMetadata &metadata, const StreamAddOptions &options,
+ bool first_entry, StreamEntryID *next_entry_id) const;
+ uint64_t trim(const std::string &ns_key, const StreamTrimOptions &options,
+ StreamMetadata *metadata, rocksdb::WriteBatch *batch);
+};
+
+} // namespace Redis
diff --git a/src/redis_stream_base.cc b/src/redis_stream_base.cc
new file mode 100644
index 0000000..4af68fe
--- /dev/null
+++ b/src/redis_stream_base.cc
@@ -0,0 +1,152 @@
+/*
+ * 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 "redis_stream_base.h"
+#include "encoding.h"
+#include "util.h"
+
+namespace Redis {
+
+const char* kErrLastEntryIdReached = "last possible entry id reached";
+const char* kErrInvalidEntryIdSpecified = "Invalid stream ID specified as stream command argument";
+
+rocksdb::Status IncrementStreamEntryID(StreamEntryID *id) {
+ if (id->seq == UINT64_MAX) {
+ if (id->ms == UINT64_MAX) {
+ // special case where 'id' is the last possible entry ID
+ id->ms = 0;
+ id->seq = 0;
+ return rocksdb::Status::InvalidArgument(kErrLastEntryIdReached);
+ } else {
+ id->ms++;
+ id->seq = 0;
+ }
+ } else {
+ id->seq++;
+ }
+
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status GetNextStreamEntryID(const StreamEntryID &last_id, StreamEntryID *new_id) {
+ uint64_t ms = Util::GetTimeStampMS();
+ if (ms > last_id.ms) {
+ new_id->ms = ms;
+ new_id->seq = 0;
+ return rocksdb::Status::OK();
+ } else {
+ *new_id = last_id;
+ return IncrementStreamEntryID(new_id);
+ }
+}
+
+Status ParseStreamEntryID(const std::string &input, StreamEntryID *id) {
+ auto pos = input.find("-");
+ try {
+ if (pos != std::string::npos) {
+ auto ms_str = input.substr(0, pos);
+ auto seq_str = input.substr(pos + 1);
+
+ id->ms = std::stoull(ms_str);
+ id->seq = std::stoull(seq_str);
+ } else {
+ id->ms = std::stoull(input);
+ id->seq = 0;
+ }
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, kErrInvalidEntryIdSpecified);
+ }
+
+ return Status::OK();
+}
+
+Status ParseNewStreamEntryID(const std::string &input, NewStreamEntryID *id) {
+ auto pos = input.find("-");
+ try {
+ if (pos != std::string::npos) {
+ auto ms_str = input.substr(0, pos);
+ auto seq_str = input.substr(pos + 1);
+
+ id->ms = std::stoull(ms_str);
+
+ if (seq_str == "*") {
+ id->any_seq_number = true;
+ } else {
+ id->seq = std::stoull(seq_str);
+ }
+ } else {
+ id->ms = std::stoull(input);
+ id->seq = 0;
+ }
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, kErrInvalidEntryIdSpecified);
+ }
+
+ return Status();
+}
+
+Status ParseRangeStart(const std::string &input, StreamEntryID *id) {
+ return ParseStreamEntryID(input, id);
+}
+
+Status ParseRangeEnd(const std::string &input, StreamEntryID *id) {
+ auto pos = input.find("-");
+ try {
+ if (pos != std::string::npos) {
+ auto ms_str = input.substr(0, pos);
+ auto seq_str = input.substr(pos + 1);
+
+ id->ms = std::stoull(ms_str);
+ id->seq = std::stoull(seq_str);
+ } else {
+ id->ms = std::stoull(input);
+ id->seq = UINT64_MAX;
+ }
+ } catch (const std::exception &) {
+ return Status(Status::RedisParseErr, kErrInvalidEntryIdSpecified);
+ }
+
+ return Status();
+}
+
+std::string EncodeStreamEntryValue(const std::vector<std::string> &args) {
+ std::string dst;
+ for (auto const &v : args) {
+ PutFixed32(&dst, v.size());
+ dst.append(v);
+ }
+ return dst;
+}
+
+std::vector<std::string> DecodeRawStreamEntryValue(const std::string &value) {
+ std::vector<std::string> result;
+ rocksdb::Slice s(value);
+
+ while (!s.empty()) {
+ uint32_t len;
+ GetFixed32(&s, &len);
+ result.emplace_back(s.data(), len);
+ s.remove_prefix(len);
+ }
+
+ return result;
+}
+
+} // namespace Redis
diff --git a/src/redis_stream_base.h b/src/redis_stream_base.h
new file mode 100644
index 0000000..76208ca
--- /dev/null
+++ b/src/redis_stream_base.h
@@ -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.
+ *
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <rocksdb/status.h>
+
+#include "status.h"
+
+namespace Redis {
+
+enum class StreamTrimStrategy {
+ None = 0,
+ MaxLen = 1,
+ MinID = 2,
+};
+
+struct StreamEntryID {
+ uint64_t ms = 0;
+ uint64_t seq = 0;
+
+ StreamEntryID() {}
+ StreamEntryID(uint64_t ms, uint64_t seq) : ms(ms), seq(seq) {}
+
+ void Clear() { ms = 0; seq = 0; }
+
+ bool IsMaximum() const { return ms == UINT64_MAX && seq == UINT64_MAX; }
+ bool IsMinimum() const { return ms == 0 && seq == 0; }
+
+ bool operator<(const StreamEntryID &rhs) const {
+ if (ms < rhs.ms) return true;
+ if (ms == rhs.ms) return seq < rhs.seq;
+ return false;
+ }
+
+ bool operator>=(const StreamEntryID &rhs) const {
+ return !(*this < rhs);
+ }
+
+ bool operator>(const StreamEntryID &rhs) const {
+ return rhs < *this;
+ }
+
+ bool operator<=(const StreamEntryID &rhs) const {
+ return !(rhs < *this);
+ }
+
+ bool operator==(const StreamEntryID &rhs) const {
+ return ms == rhs.ms && seq == rhs.seq;
+ }
+
+ std::string ToString() const {
+ return std::to_string(ms) + "-" + std::to_string(seq);
+ }
+
+ static StreamEntryID Minimum() { return StreamEntryID{0, 0}; }
+ static StreamEntryID Maximum() { return StreamEntryID{UINT64_MAX, UINT64_MAX}; }
+};
+
+struct NewStreamEntryID {
+ uint64_t ms = 0;
+ uint64_t seq = 0;
+ bool any_seq_number = false;
+
+ NewStreamEntryID() = default;
+ explicit NewStreamEntryID(uint64_t ms) : ms(ms), any_seq_number(true) {}
+ NewStreamEntryID(uint64_t ms, uint64_t seq) : ms(ms), seq(seq) {}
+};
+
+struct StreamEntry {
+ std::string key;
+ std::vector<std::string> values;
+
+ StreamEntry(std::string k, std::vector<std::string> vv) : key(std::move(k)), values(std::move(vv)) {}
+};
+
+struct StreamTrimOptions {
+ uint64_t max_len;
+ StreamEntryID min_id;
+ StreamTrimStrategy strategy = StreamTrimStrategy::None;
+};
+
+struct StreamAddOptions {
+ NewStreamEntryID entry_id;
+ StreamTrimOptions trim_options;
+ bool nomkstream = false;
+ bool with_entry_id = false;
+};
+
+struct StreamRangeOptions {
+ StreamEntryID start;
+ StreamEntryID end;
+ uint64_t count;
+ bool reverse = false;
+ bool exclude_start = false;
+ bool exclude_end = false;
+ bool with_count = false;
+};
+
+struct StreamInfo {
+ uint64_t size;
+ uint64_t entries_added;
+ StreamEntryID last_generated_id;
+ StreamEntryID max_deleted_entry_id;
+ StreamEntryID recorded_first_entry_id;
+ std::unique_ptr<StreamEntry> first_entry;
+ std::unique_ptr<StreamEntry> last_entry;
+ std::vector<StreamEntry> entries;
+};
+
+struct StreamReadResult {
+ std::string name;
+ std::vector<StreamEntry> entries;
+
+ StreamReadResult(std::string name, std::vector<StreamEntry> result)
+ : name(std::move(name)), entries(std::move(result)) {}
+};
+
+rocksdb::Status IncrementStreamEntryID(StreamEntryID *id);
+rocksdb::Status GetNextStreamEntryID(const StreamEntryID &last_id, StreamEntryID *new_id);
+Status ParseStreamEntryID(const std::string &input, StreamEntryID *id);
+Status ParseNewStreamEntryID(const std::string &input, NewStreamEntryID *id);
+Status ParseRangeStart(const std::string &input, StreamEntryID *id);
+Status ParseRangeEnd(const std::string &input, StreamEntryID *id);
+std::string EncodeStreamEntryValue(const std::vector<std::string> &args);
+std::vector<std::string> DecodeRawStreamEntryValue(const std::string &value);
+
+
+} // namespace Redis
diff --git a/src/replication.cc b/src/replication.cc
index 11e4640..af5dff8 100644
--- a/src/replication.cc
+++ b/src/replication.cc
@@ -958,6 +958,14 @@ rocksdb::Status ReplicationThread::ParseWriteBatch(const std::string &batch_stri
}
}
break;
+ case kBatchTypeStream:
+ InternalKey ikey(write_batch_handler.Key(), storage_->IsSlotIdEncoded());
+ Slice entry_id = ikey.GetSubKey();
+ Redis::StreamEntryID id;
+ GetFixed64(&entry_id, &id.ms);
+ GetFixed64(&entry_id, &id.seq);
+ srv_->OnEntryAddedToStream(ikey.GetNamespace().ToString(), ikey.GetKey().ToString(), id);
+ break;
}
return rocksdb::Status::OK();
}
@@ -980,6 +988,9 @@ rocksdb::Status WriteBatchHandler::PutCF(uint32_t column_family_id, const rocksd
type_ = kBatchTypePropagate;
kv_ = std::make_pair(key.ToString(), value.ToString());
return rocksdb::Status::OK();
+ } else if (column_family_id == kColumnFamilyIDStream) {
+ type_ = kBatchTypeStream;
+ kv_ = std::make_pair(key.ToString(), value.ToString());
}
return rocksdb::Status::OK();
}
diff --git a/src/replication.h b/src/replication.h
index 69e3eef..5632e1a 100644
--- a/src/replication.h
+++ b/src/replication.h
@@ -50,6 +50,7 @@ enum ReplState {
enum WriteBatchType {
kBatchTypePublish = 1,
kBatchTypePropagate,
+ kBatchTypeStream,
};
typedef std::function<void(const std::string, const uint32_t)> fetch_file_callback;
diff --git a/src/server.cc b/src/server.cc
index b48d41b..5a49d87 100644
--- a/src/server.cc
+++ b/src/server.cc
@@ -456,6 +456,7 @@ void Server::AddBlockingKey(const std::string &key, Redis::Connection *conn) {
} else {
iter->second.emplace_back(conn_ctx);
}
+ IncrBlockedClientNum();
}
void Server::UnBlockingKey(const std::string &key, Redis::Connection *conn) {
@@ -474,6 +475,47 @@ void Server::UnBlockingKey(const std::string &key, Redis::Connection *conn) {
break;
}
}
+ DecrBlockedClientNum();
+}
+
+void Server::BlockOnStreams(const std::vector<std::string> &keys,
+ const std::vector<Redis::StreamEntryID> &entry_ids, Redis::Connection *conn) {
+ std::lock_guard<std::mutex> guard(blocking_keys_mu_);
+ IncrBlockedClientNum();
+ for (size_t i = 0; i < keys.size(); ++i) {
+ auto consumer = std::make_shared<StreamConsumer>(
+ conn->Owner(), conn->GetFD(), conn->GetNamespace(), entry_ids[i]);
+ auto iter = blocked_stream_consumers_.find(keys[i]);
+ if (iter == blocked_stream_consumers_.end()) {
+ std::set<std::shared_ptr<StreamConsumer>> consumers;
+ consumers.insert(consumer);
+ blocked_stream_consumers_.insert(std::make_pair(keys[i], consumers));
+ } else {
+ iter->second.insert(consumer);
+ }
+ }
+}
+
+void Server::UnblockOnStreams(const std::vector<std::string> &keys, Redis::Connection *conn) {
+ std::lock_guard<std::mutex> guard(blocking_keys_mu_);
+ DecrBlockedClientNum();
+ for (const auto &key : keys) {
+ auto iter = blocked_stream_consumers_.find(key);
+ if (iter == blocked_stream_consumers_.end()) {
+ continue;
+ }
+
+ for (auto it = iter->second.begin(); it != iter->second.end(); ) {
+ auto consumer = *it;
+ if (conn->GetFD() == consumer->fd && conn->Owner() == consumer->owner) {
+ iter->second.erase(it);
+ if (iter->second.empty()) {
+ blocked_stream_consumers_.erase(iter);
+ }
+ break;
+ }
+ }
+ }
}
Status Server::WakeupBlockingConns(const std::string &key, size_t n_conns) {
@@ -491,6 +533,27 @@ Status Server::WakeupBlockingConns(const std::string &key, size_t n_conns) {
return Status::OK();
}
+Status Server::OnEntryAddedToStream(const std::string &ns, const std::string &key,
+ const Redis::StreamEntryID &entry_id) {
+ std::lock_guard<std::mutex> guard(blocking_keys_mu_);
+ auto iter = blocked_stream_consumers_.find(key);
+ if (iter == blocked_stream_consumers_.end() || iter->second.empty()) {
+ return Status(Status::NotOK);
+ }
+
+ for (auto it = iter->second.begin(); it != iter->second.end(); ) {
+ auto consumer = *it;
+ if (consumer->ns == ns && entry_id > consumer->last_consumed_id) {
+ consumer->owner->EnableWriteEvent(consumer->fd);
+ it = iter->second.erase(it);
+ } else {
+ ++it;
+ }
+ }
+
+ return Status::OK();
+}
+
void Server::delConnContext(ConnContext *c) {
auto conn_ctx_iter = conn_ctxs_.find(c);
if (conn_ctx_iter != conn_ctxs_.end()) {
@@ -522,6 +585,14 @@ int Server::DecrMonitorClientNum() {
return monitor_clients_.fetch_sub(1, std::memory_order_relaxed);
}
+int Server::IncrBlockedClientNum() {
+ return blocked_clients_.fetch_add(1, std::memory_order_relaxed);
+}
+
+int Server::DecrBlockedClientNum() {
+ return blocked_clients_.fetch_sub(1, std::memory_order_relaxed);
+}
+
std::unique_ptr<RWLock::ReadLock> Server::WorkConcurrencyGuard() {
return std::unique_ptr<RWLock::ReadLock>(new RWLock::ReadLock(works_concurrency_rw_lock_));
}
@@ -740,6 +811,7 @@ void Server::GetClientsInfo(std::string *info) {
string_stream << "maxclients:" << config_->maxclients << "\r\n";
string_stream << "connected_clients:" << connected_clients_ << "\r\n";
string_stream << "monitor_clients:" << monitor_clients_ << "\r\n";
+ string_stream << "blocked_clients:" << blocked_clients_ << "\r\n";
*info = string_stream.str();
}
diff --git a/src/server.h b/src/server.h
index 3452c58..5f85fd4 100644
--- a/src/server.h
+++ b/src/server.h
@@ -29,6 +29,8 @@
#include <vector>
#include <memory>
#include <unordered_map>
+#include <set>
+#include <utility>
#include "lua.hpp"
#include "stats.h"
@@ -55,6 +57,15 @@ struct ConnContext {
ConnContext(Worker *w, int fd) : owner(w), fd(fd) {}
};
+struct StreamConsumer {
+ Worker *owner;
+ int fd;
+ std::string ns;
+ Redis::StreamEntryID last_consumed_id;
+ StreamConsumer(Worker *w, int fd, std::string ns, Redis::StreamEntryID id) :
+ owner(w), fd(fd), ns(std::move(ns)), last_consumed_id(id) {}
+};
+
typedef struct {
std::string channel;
size_t subscribe_num;
@@ -138,7 +149,12 @@ class Server {
void AddBlockingKey(const std::string &key, Redis::Connection *conn);
void UnBlockingKey(const std::string &key, Redis::Connection *conn);
+ void BlockOnStreams(const std::vector<std::string> &keys,
+ const std::vector<Redis::StreamEntryID> &entry_ids, Redis::Connection *conn);
+ void UnblockOnStreams(const std::vector<std::string> &keys, Redis::Connection *conn);
Status WakeupBlockingConns(const std::string &key, size_t n_conns);
+ Status OnEntryAddedToStream(const std::string &ns, const std::string &key,
+ const Redis::StreamEntryID &entry_id);
std::string GetLastRandomKeyCursor();
void SetLastRandomKeyCursor(const std::string &cursor);
@@ -169,6 +185,8 @@ class Server {
int IncrClientNum();
int IncrMonitorClientNum();
int DecrMonitorClientNum();
+ int IncrBlockedClientNum();
+ int DecrBlockedClientNum();
std::string GetClientsStr();
std::atomic<uint64_t> *GetClientID();
void KillClient(int64_t *killed, std::string addr, uint64_t id, uint64_t type,
@@ -254,6 +272,8 @@ class Server {
std::mutex pubsub_channels_mu_;
std::map<std::string, std::list<ConnContext *>> blocking_keys_;
std::mutex blocking_keys_mu_;
+ std::atomic<int> blocked_clients_{0};
+ std::map<std::string, std::set<std::shared_ptr<StreamConsumer>>> blocked_stream_consumers_;
// threads
RWLock::ReadWriteLock works_concurrency_rw_lock_;
diff --git a/src/storage.cc b/src/storage.cc
index 8b1ebcd..5c58efd 100644
--- a/src/storage.cc
+++ b/src/storage.cc
@@ -53,6 +53,7 @@ const char *kZSetScoreColumnFamilyName = "zset_score";
const char *kMetadataColumnFamilyName = "metadata";
const char *kSubkeyColumnFamilyName = "default";
const char *kPropagateColumnFamilyName = "propagate";
+const char *kStreamColumnFamilyName = "stream";
const char *kPropagateScriptCommand = "script";
@@ -204,7 +205,8 @@ Status Storage::CreateColumnFamilies(const rocksdb::Options &options) {
std::vector<std::string> cf_names = {kMetadataColumnFamilyName,
kZSetScoreColumnFamilyName,
kPubSubColumnFamilyName,
- kPropagateColumnFamilyName};
+ kPropagateColumnFamilyName,
+ kStreamColumnFamilyName};
std::vector<rocksdb::ColumnFamilyHandle *> cf_handles;
s = tmp_db->CreateColumnFamilies(cf_options, cf_names, &cf_handles);
if (!s.ok()) {
@@ -301,6 +303,7 @@ Status Storage::Open(bool read_only) {
column_families.emplace_back(rocksdb::ColumnFamilyDescriptor(kZSetScoreColumnFamilyName, subkey_opts));
column_families.emplace_back(rocksdb::ColumnFamilyDescriptor(kPubSubColumnFamilyName, pubsub_opts));
column_families.emplace_back(rocksdb::ColumnFamilyDescriptor(kPropagateColumnFamilyName, propagate_opts));
+ column_families.emplace_back(rocksdb::ColumnFamilyDescriptor(kStreamColumnFamilyName, subkey_opts));
std::vector<std::string> old_column_families;
auto s = rocksdb::DB::ListColumnFamilies(options, config_->db_dir, &old_column_families);
if (!s.ok()) return Status(Status::NotOK, s.ToString());
@@ -568,6 +571,8 @@ rocksdb::ColumnFamilyHandle *Storage::GetCFHandle(const std::string &name) {
return cf_handles_[3];
} else if (name == kPropagateColumnFamilyName) {
return cf_handles_[4];
+ } else if (name == kStreamColumnFamilyName) {
+ return cf_handles_[5];
}
return cf_handles_[0];
}
diff --git a/src/storage.h b/src/storage.h
index 234fc69..848ddd2 100644
--- a/src/storage.h
+++ b/src/storage.h
@@ -45,6 +45,7 @@ enum ColumnFamilyID{
kColumnFamilyIDZSetScore,
kColumnFamilyIDPubSub,
kColumnFamilyIDPropagate,
+ kColumnFamilyIDStream,
};
namespace Engine {
@@ -53,6 +54,7 @@ extern const char *kZSetScoreColumnFamilyName;
extern const char *kMetadataColumnFamilyName;
extern const char *kSubkeyColumnFamilyName;
extern const char *kPropagateColumnFamilyName;
+extern const char *kStreamColumnFamilyName;
extern const char *kPropagateScriptCommand;
diff --git a/tests/cppunit/t_stream_test.cc b/tests/cppunit/t_stream_test.cc
new file mode 100644
index 0000000..927e5b1
--- /dev/null
+++ b/tests/cppunit/t_stream_test.cc
@@ -0,0 +1,2057 @@
+/*
+ * 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 <gtest/gtest.h>
+#include "test_base.h"
+#include "redis_stream.h"
+
+
+class RedisStreamTest : public TestBase {
+public:
+ void checkStreamEntryValues(const std::vector<std::string> &got, const std::vector<std::string> &expected) {
+ EXPECT_EQ(got.size(), expected.size());
+ for (size_t i = 0; i < got.size(); ++i) {
+ EXPECT_EQ(got[i], expected[i]);
+ }
+ }
+
+protected:
+ RedisStreamTest() : TestBase() {
+ stream = new Redis::Stream(storage_, "stream_ns");
+ name = "test_stream";
+ }
+
+ ~RedisStreamTest() {
+ delete stream;
+ }
+
+ void SetUp() override {
+ stream->Del(name);
+ }
+
+ void TearDown() override {
+ stream->Del(name);
+ }
+
+protected:
+ std::string name;
+ Redis::Stream *stream;
+};
+
+TEST_F(RedisStreamTest, AddEntryToNonExistingStreamWithNomkstreamOption) {
+ Redis::StreamAddOptions options;
+ options.nomkstream = true;
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.IsNotFound());
+}
+
+TEST_F(RedisStreamTest, AddEntryPredefinedIDAsZeroZero) {
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{0, 0};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(!s.ok());
+}
+
+TEST_F(RedisStreamTest, AddEntryWithPredefinedIDAsZeroMsAndAnySeq) {
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{0};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(id.ToString(), "0-1");
+}
+
+TEST_F(RedisStreamTest, AddFirstEntryWithoutPredefinedID) {
+ Redis::StreamAddOptions options;
+ options.with_entry_id = false;
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(id.seq, 0);
+ EXPECT_TRUE(id.ms <= Util::GetTimeStampMS());
+}
+
+TEST_F(RedisStreamTest, AddEntryFirstEntryWithPredefinedID) {
+ Redis::StreamEntryID expected_id{12345, 6789};
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{expected_id.ms, expected_id.seq};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(id.ms, expected_id.ms);
+ EXPECT_EQ(id.seq, expected_id.seq);
+}
+
+TEST_F(RedisStreamTest, AddFirstEntryWithPredefinedNonZeroMsAndAnySeqNo) {
+ uint64_t ms = Util::GetTimeStampMS();
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{ms};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(id.ms, ms);
+ EXPECT_EQ(id.seq, 0);
+}
+
+TEST_F(RedisStreamTest, AddEntryToNonEmptyStreamWithPredefinedMsAndAnySeqNo) {
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{12345, 678};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ options.entry_id = Redis::NewStreamEntryID{12346};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(id2.ToString(), "12346-0");
+}
+
+TEST_F(RedisStreamTest, AddEntryWithPredefinedButExistingMsAndAnySeqNo) {
+ uint64_t ms = 12345;
+ uint64_t seq = 6789;
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{ms, seq};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{ms};
+ s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(id.ms, ms);
+ EXPECT_EQ(id.seq, seq + 1);
+}
+
+TEST_F(RedisStreamTest, AddEntryWithExistingMsAnySeqNoAndExistingSeqNoIsAlreadyMax) {
+ uint64_t ms = 12345;
+ uint64_t seq = UINT64_MAX;
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{ms, seq};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{ms};
+ s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(!s.ok());
+}
+
+TEST_F(RedisStreamTest, AddEntryAndExistingMsAndSeqNoAreAlreadyMax) {
+ uint64_t ms = UINT64_MAX;
+ uint64_t seq = UINT64_MAX;
+ Redis::StreamAddOptions options;
+ options.with_entry_id = true;
+ options.entry_id = Redis::NewStreamEntryID{ms, seq};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(s.ok());
+ options.with_entry_id = false;
+ s = stream->Add(name, options, values, &id);
+ EXPECT_TRUE(!s.ok());
+}
+
+TEST_F(RedisStreamTest, AddEntryWithTrimMaxLenStrategy) {
+ Redis::StreamAddOptions add_options;
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions trim_options;
+ trim_options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ trim_options.max_len = 2;
+ add_options.trim_options = trim_options;
+ Redis::StreamEntryID id3;
+ std::vector<std::string> values3 = {"key3", "val3"};
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id3.ToString());
+ checkStreamEntryValues(entries[1].values, values3);
+}
+
+TEST_F(RedisStreamTest, AddEntryWithTrimMaxLenStrategyThatDeletesAddedEntry) {
+ Redis::StreamAddOptions add_options;
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions trim_options;
+ trim_options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ trim_options.max_len = 0;
+ add_options.trim_options = trim_options;
+ Redis::StreamEntryID id3;
+ std::vector<std::string> values3 = {"key3", "val3"};
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, AddEntryWithTrimMinIdStrategy) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{12346, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions trim_options;
+ trim_options.strategy = Redis::StreamTrimStrategy::MinID;
+ trim_options.min_id = Redis::StreamEntryID{12346, 0};
+ add_options.trim_options = trim_options;
+ add_options.entry_id = Redis::NewStreamEntryID{12347, 0};
+ Redis::StreamEntryID id3;
+ std::vector<std::string> values3 = {"key3", "val3"};
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id3.ToString());
+ checkStreamEntryValues(entries[1].values, values3);
+}
+
+TEST_F(RedisStreamTest, AddEntryWithTrimMinIdStrategyThatDeletesAddedEntry) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{12346, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions trim_options;
+ trim_options.strategy = Redis::StreamTrimStrategy::MinID;
+ trim_options.min_id = Redis::StreamEntryID{1234567, 0};
+ add_options.trim_options = trim_options;
+ add_options.entry_id = Redis::NewStreamEntryID{12347, 0};
+ Redis::StreamEntryID id3;
+ std::vector<std::string> values3 = {"key3", "val3"};
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeOnNonExistingStream) {
+ Redis::StreamRangeOptions options;
+ options.start = Redis::StreamEntryID{0, 0};
+ options.end = Redis::StreamEntryID{1234567, 0};
+ std::vector<Redis::StreamEntry> entries;
+ auto s = stream->Range(name, options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeOnEmptyStream) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = false;
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+ uint64_t removed;
+ s = stream->DeleteEntries(name, {id}, &removed);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeWithStartAndEndSameMs) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345678, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{12345678, 1};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{12345679, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{12345678, 0};
+ range_options.end = Redis::StreamEntryID{12345678, UINT64_MAX};
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id1.ToString());
+ checkStreamEntryValues(entries[0].values, values1);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+}
+
+TEST_F(RedisStreamTest, RangeInterval) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{123456, 0};
+ range_options.end = Redis::StreamEntryID{123459, 0};
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 3);
+ EXPECT_EQ(entries[0].key, id1.ToString());
+ checkStreamEntryValues(entries[0].values, values1);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+ EXPECT_EQ(entries[2].key, id3.ToString());
+ checkStreamEntryValues(entries[2].values, values3);
+}
+
+TEST_F(RedisStreamTest, RangeFromMinimumToMaximum) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 4);
+ EXPECT_EQ(entries[0].key, id1.ToString());
+ checkStreamEntryValues(entries[0].values, values1);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+ EXPECT_EQ(entries[2].key, id3.ToString());
+ checkStreamEntryValues(entries[2].values, values3);
+ EXPECT_EQ(entries[3].key, id4.ToString());
+ checkStreamEntryValues(entries[3].values, values4);
+}
+
+TEST_F(RedisStreamTest, RangeFromMinimumToMinimum) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Minimum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeWithStartGreaterThanEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Maximum();
+ range_options.end = Redis::StreamEntryID::Minimum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeWithStartAndEndAreEqual) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = id2;
+ range_options.end = id2;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 1);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+}
+
+TEST_F(RedisStreamTest, RangeWithStartAndEndAreEqualAndExludedStart) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = id2;
+ range_options.exclude_start = true;
+ range_options.end = id2;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeWithStartAndEndAreEqualAndExludedEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = id2;
+ range_options.end = id2;
+ range_options.exclude_end = true;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeWithExcludedStart) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{123456, 1};
+ range_options.exclude_start = true;
+ range_options.end = Redis::StreamEntryID{123458, 3};
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id3.ToString());
+ checkStreamEntryValues(entries[1].values, values3);
+}
+
+TEST_F(RedisStreamTest, RangeWithExcludedEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{123457, 2};
+ range_options.end = Redis::StreamEntryID{123459, 4};
+ range_options.exclude_end = true;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id3.ToString());
+ checkStreamEntryValues(entries[1].values, values3);
+}
+
+TEST_F(RedisStreamTest, RangeWithExcludedStartAndExcludedEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{123456, 1};
+ range_options.exclude_start = true;
+ range_options.end = Redis::StreamEntryID{123459, 4};
+ range_options.exclude_end = true;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id3.ToString());
+ checkStreamEntryValues(entries[1].values, values3);
+}
+
+TEST_F(RedisStreamTest, RangeWithStartAsMaximumAndExlusion) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Maximum();
+ range_options.exclude_start = true;
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(!s.ok());
+}
+
+TEST_F(RedisStreamTest, RangeWithEndAsMinimumAndExlusion) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Minimum();
+ range_options.exclude_end = true;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(!s.ok());
+}
+
+TEST_F(RedisStreamTest, RangeWithCountEqualToZero) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{123456, 0};
+ range_options.end = Redis::StreamEntryID{123459, 0};
+ range_options.with_count = true;
+ range_options.count = 0;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RangeWithCountGreaterThanRequiredElements) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{123456, 0};
+ range_options.end = Redis::StreamEntryID{123459, 0};
+ range_options.with_count = true;
+ range_options.count = 3;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 3);
+ EXPECT_EQ(entries[0].key, id1.ToString());
+ checkStreamEntryValues(entries[0].values, values1);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+ EXPECT_EQ(entries[2].key, id3.ToString());
+ checkStreamEntryValues(entries[2].values, values3);
+}
+
+TEST_F(RedisStreamTest, RangeWithCountLessThanRequiredElements) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID{123456, 0};
+ range_options.end = Redis::StreamEntryID{123459, 0};
+ range_options.with_count = true;
+ range_options.count = 2;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id1.ToString());
+ checkStreamEntryValues(entries[0].values, values1);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+}
+
+TEST_F(RedisStreamTest, RevRangeWithStartAndEndSameMs) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345678, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{12345678, 1};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{12345679, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID{12345678, UINT64_MAX};
+ range_options.end = Redis::StreamEntryID{12345678, 0};
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id1.ToString());
+ checkStreamEntryValues(entries[1].values, values1);
+}
+
+TEST_F(RedisStreamTest, RevRangeInterval) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID{123459, 0};
+ range_options.end = Redis::StreamEntryID{123456, 0};
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 3);
+ EXPECT_EQ(entries[0].key, id3.ToString());
+ checkStreamEntryValues(entries[0].values, values3);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+ EXPECT_EQ(entries[2].key, id1.ToString());
+ checkStreamEntryValues(entries[2].values, values1);
+}
+
+TEST_F(RedisStreamTest, RevRangeFromMaximumToMinimum) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID::Maximum();
+ range_options.end = Redis::StreamEntryID::Minimum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 4);
+ EXPECT_EQ(entries[0].key, id4.ToString());
+ checkStreamEntryValues(entries[0].values, values4);
+ EXPECT_EQ(entries[1].key, id3.ToString());
+ checkStreamEntryValues(entries[1].values, values3);
+ EXPECT_EQ(entries[2].key, id2.ToString());
+ checkStreamEntryValues(entries[2].values, values2);
+ EXPECT_EQ(entries[3].key, id1.ToString());
+ checkStreamEntryValues(entries[3].values, values1);
+}
+
+TEST_F(RedisStreamTest, RevRangeFromMinimumToMinimum) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Minimum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RevRangeWithStartLessThanEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RevRangeStartAndEndAreEqual) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = id2;
+ range_options.end = id2;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 1);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+}
+
+TEST_F(RedisStreamTest, RevRangeStartAndEndAreEqualAndExcludedStart) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = id2;
+ range_options.exclude_start = true;
+ range_options.end = id2;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RevRangeStartAndEndAreEqualAndExcludedEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = id2;
+ range_options.end = id2;
+ range_options.exclude_end = true;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, RevRangeWithExcludedStart) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID{123458, 3};
+ range_options.exclude_start = true;
+ range_options.end = Redis::StreamEntryID{123456, 1};
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id1.ToString());
+ checkStreamEntryValues(entries[1].values, values1);
+}
+
+TEST_F(RedisStreamTest, RevRangeWithExcludedEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID{123458, 3};
+ range_options.end = Redis::StreamEntryID{123456, 1};
+ range_options.exclude_end = true;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id3.ToString());
+ checkStreamEntryValues(entries[0].values, values3);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+}
+
+TEST_F(RedisStreamTest, RevRangeWithExcludedStartAndExcludedEnd) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 1};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 2};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 3};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 4};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamRangeOptions range_options;
+ range_options.reverse = true;
+ range_options.start = Redis::StreamEntryID{123459, 4};
+ range_options.exclude_start = true;
+ range_options.end = Redis::StreamEntryID{123456, 1};
+ range_options.exclude_end = true;
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id3.ToString());
+ checkStreamEntryValues(entries[0].values, values3);
+ EXPECT_EQ(entries[1].key, id2.ToString());
+ checkStreamEntryValues(entries[1].values, values2);
+}
+
+TEST_F(RedisStreamTest, DeleteFromNonExistingStream) {
+ std::vector<Redis::StreamEntryID> ids = {Redis::StreamEntryID{12345, 6789}};
+ uint64_t deleted;
+ auto s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(deleted, 0);
+}
+
+TEST_F(RedisStreamTest, DeleteExistingEntry) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 6789};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+
+ std::vector<Redis::StreamEntryID> ids = {id};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(deleted, 1);
+}
+
+TEST_F(RedisStreamTest, DeleteNonExistingEntry) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 6789};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+
+ std::vector<Redis::StreamEntryID> ids = {Redis::StreamEntryID{123, 456}};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(deleted, 0);
+}
+
+TEST_F(RedisStreamTest, DeleteMultipleEntries) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ std::vector<Redis::StreamEntryID> ids = {
+ Redis::StreamEntryID{123456, 0}, Redis::StreamEntryID{1234567, 89}, Redis::StreamEntryID{123458, 0}};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(deleted, 2);
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id2.ToString());
+ checkStreamEntryValues(entries[0].values, values2);
+ EXPECT_EQ(entries[1].key, id4.ToString());
+ checkStreamEntryValues(entries[1].values, values4);
+}
+
+TEST_F(RedisStreamTest, LenOnNonExistingStream) {
+ uint64_t length;
+ auto s = stream->Len(name, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 0);
+}
+
+TEST_F(RedisStreamTest, LenOnEmptyStream) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 6789};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+
+ std::vector<Redis::StreamEntryID> ids = {id};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+
+ uint64_t length;
+ s = stream->Len(name, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 0);
+}
+
+TEST_F(RedisStreamTest, Len) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ uint64_t length;
+ s = stream->Len(name, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 2);
+}
+
+TEST_F(RedisStreamTest, TrimNonExistingStream) {
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 10;
+ uint64_t trimmed;
+ auto s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 0);
+}
+
+TEST_F(RedisStreamTest, TrimEmptyStream) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 6789};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+ std::vector<Redis::StreamEntryID> ids = {id};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 10;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 0);
+}
+
+TEST_F(RedisStreamTest, TrimWithNoStrategySpecified) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 6789};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.min_id = Redis::StreamEntryID{123456, 0};
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 0);
+}
+
+TEST_F(RedisStreamTest, TrimWithMaxLenGreaterThanStreamSize) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 10;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 0);
+}
+
+TEST_F(RedisStreamTest, TrimWithMaxLenEqualToStreamSize) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 4;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 0);
+}
+
+TEST_F(RedisStreamTest, TrimWithMaxLenLessThanStreamSize) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 2;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 2);
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id3.ToString());
+ checkStreamEntryValues(entries[0].values, values3);
+ EXPECT_EQ(entries[1].key, id4.ToString());
+ checkStreamEntryValues(entries[1].values, values4);
+}
+
+TEST_F(RedisStreamTest, TrimWithMaxLenEqualTo1) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 1;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 3);
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 1);
+ EXPECT_EQ(entries[0].key, id4.ToString());
+ checkStreamEntryValues(entries[0].values, values4);
+}
+
+TEST_F(RedisStreamTest, TrimWithMaxLenZero) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 0;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 4);
+ uint64_t length;
+ s = stream->Len(name, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 0);
+}
+
+TEST_F(RedisStreamTest, TrimWithMinIdLessThanFirstEntryID) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MinID;
+ options.min_id = Redis::StreamEntryID{12345, 0};
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 0);
+}
+
+TEST_F(RedisStreamTest, TrimWithMinIdEqualToFirstEntryID) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MinID;
+ options.min_id = Redis::StreamEntryID{123456, 0};
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 0);
+}
+
+TEST_F(RedisStreamTest, TrimWithMinId) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MinID;
+ options.min_id = Redis::StreamEntryID{123457, 10};
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 2);
+
+ Redis::StreamRangeOptions range_options;
+ range_options.start = Redis::StreamEntryID::Minimum();
+ range_options.end = Redis::StreamEntryID::Maximum();
+ std::vector<Redis::StreamEntry> entries;
+ s = stream->Range(name, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id3.ToString());
+ checkStreamEntryValues(entries[0].values, values3);
+ EXPECT_EQ(entries[1].key, id4.ToString());
+ checkStreamEntryValues(entries[1].values, values4);
+}
+
+TEST_F(RedisStreamTest, TrimWithMinIdGreaterThanLastEntryID) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MinID;
+ options.min_id = Redis::StreamEntryID{12345678, 0};
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 4);
+
+ uint64_t length;
+ s = stream->Len(name, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 0);
+}
+
+TEST_F(RedisStreamTest, StreamInfoOnNonExistingStream) {
+ Redis::StreamInfo info;
+ auto s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.IsNotFound());
+}
+
+TEST_F(RedisStreamTest, StreamInfoOnEmptyStream) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 6789};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+
+ std::vector<Redis::StreamEntryID> ids = {id};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 0);
+ EXPECT_EQ(info.last_generated_id.ToString(), id.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), id.ToString());
+ EXPECT_EQ(info.entries_added, 1);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), "0-0");
+ EXPECT_FALSE(info.first_entry);
+ EXPECT_FALSE(info.last_entry);
+}
+
+TEST_F(RedisStreamTest, StreamInfoOneEntry) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{12345, 6789};
+ std::vector<std::string> values = {"key1", "val1"};
+ Redis::StreamEntryID id;
+ auto s = stream->Add(name, add_options, values, &id);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 1);
+ EXPECT_EQ(info.last_generated_id.ToString(), id.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), "0-0");
+ EXPECT_EQ(info.entries_added, 1);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), id.ToString());
+ EXPECT_TRUE(info.first_entry);
+ EXPECT_EQ(info.first_entry->key, id.ToString());
+ checkStreamEntryValues(info.first_entry->values, values);
+ EXPECT_TRUE(info.last_entry);
+ EXPECT_EQ(info.last_entry->key, id.ToString());
+ checkStreamEntryValues(info.last_entry->values, values);
+}
+
+TEST_F(RedisStreamTest, StreamInfoOnStreamWithElements) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 3);
+ EXPECT_EQ(info.last_generated_id.ToString(), id3.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), "0-0");
+ EXPECT_EQ(info.entries_added, 3);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), id1.ToString());
+ EXPECT_TRUE(info.first_entry);
+ EXPECT_EQ(info.first_entry->key, id1.ToString());
+ checkStreamEntryValues(info.first_entry->values, values1);
+ EXPECT_TRUE(info.last_entry);
+ EXPECT_EQ(info.last_entry->key, id3.ToString());
+ checkStreamEntryValues(info.last_entry->values, values3);
+ EXPECT_EQ(info.entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, StreamInfoOnStreamWithElementsFullOption) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, true, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 3);
+ EXPECT_EQ(info.last_generated_id.ToString(), id3.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), "0-0");
+ EXPECT_EQ(info.entries_added, 3);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), id1.ToString());
+ EXPECT_FALSE(info.first_entry);
+ EXPECT_FALSE(info.last_entry);
+ EXPECT_EQ(info.entries.size(), 3);
+ EXPECT_EQ(info.entries[0].key, id1.ToString());
+ checkStreamEntryValues(info.entries[0].values, values1);
+ EXPECT_EQ(info.entries[1].key, id2.ToString());
+ checkStreamEntryValues(info.entries[1].values, values2);
+ EXPECT_EQ(info.entries[2].key, id3.ToString());
+ checkStreamEntryValues(info.entries[2].values, values3);
+}
+
+TEST_F(RedisStreamTest, StreamInfoCheckAfterLastEntryDeletion) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ std::vector<Redis::StreamEntryID> ids = {id3};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 2);
+ EXPECT_EQ(info.last_generated_id.ToString(), id3.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), id3.ToString());
+ EXPECT_EQ(info.entries_added, 3);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), id1.ToString());
+ EXPECT_TRUE(info.first_entry);
+ EXPECT_EQ(info.first_entry->key, id1.ToString());
+ checkStreamEntryValues(info.first_entry->values, values1);
+ EXPECT_TRUE(info.last_entry);
+ EXPECT_EQ(info.last_entry->key, id2.ToString());
+ checkStreamEntryValues(info.last_entry->values, values2);
+ EXPECT_EQ(info.entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, StreamInfoCheckAfterFirstEntryDeletion) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+
+ std::vector<Redis::StreamEntryID> ids = {id1};
+ uint64_t deleted;
+ s = stream->DeleteEntries(name, ids, &deleted);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 2);
+ EXPECT_EQ(info.last_generated_id.ToString(), id3.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), id1.ToString());
+ EXPECT_EQ(info.entries_added, 3);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), id2.ToString());
+ EXPECT_TRUE(info.first_entry);
+ EXPECT_EQ(info.first_entry->key, id2.ToString());
+ checkStreamEntryValues(info.first_entry->values, values2);
+ EXPECT_TRUE(info.last_entry);
+ EXPECT_EQ(info.last_entry->key, id3.ToString());
+ checkStreamEntryValues(info.last_entry->values, values3);
+ EXPECT_EQ(info.entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, StreamInfoCheckAfterTrimMinId) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MinID;
+ options.min_id = Redis::StreamEntryID{123458, 0};
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 2);
+ EXPECT_EQ(info.last_generated_id.ToString(), id4.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), id2.ToString());
+ EXPECT_EQ(info.entries_added, 4);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), id3.ToString());
+ EXPECT_TRUE(info.first_entry);
+ EXPECT_EQ(info.first_entry->key, id3.ToString());
+ checkStreamEntryValues(info.first_entry->values, values3);
+ EXPECT_TRUE(info.last_entry);
+ EXPECT_EQ(info.last_entry->key, id4.ToString());
+ checkStreamEntryValues(info.last_entry->values, values4);
+ EXPECT_EQ(info.entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, StreamInfoCheckAfterTrimMaxLen) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 2;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 2);
+ EXPECT_EQ(info.last_generated_id.ToString(), id4.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), id2.ToString());
+ EXPECT_EQ(info.entries_added, 4);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), id3.ToString());
+ EXPECT_TRUE(info.first_entry);
+ EXPECT_EQ(info.first_entry->key, id3.ToString());
+ checkStreamEntryValues(info.first_entry->values, values3);
+ EXPECT_TRUE(info.last_entry);
+ EXPECT_EQ(info.last_entry->key, id4.ToString());
+ checkStreamEntryValues(info.last_entry->values, values4);
+ EXPECT_EQ(info.entries.size(), 0);
+}
+
+TEST_F(RedisStreamTest, StreamInfoCheckAfterTrimAllEntries) {
+ Redis::StreamAddOptions add_options;
+ add_options.with_entry_id = true;
+ add_options.entry_id = Redis::NewStreamEntryID{123456, 0};
+ std::vector<std::string> values1 = {"key1", "val1"};
+ Redis::StreamEntryID id1;
+ auto s = stream->Add(name, add_options, values1, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123457, 0};
+ std::vector<std::string> values2 = {"key2", "val2"};
+ Redis::StreamEntryID id2;
+ s = stream->Add(name, add_options, values2, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123458, 0};
+ std::vector<std::string> values3 = {"key3", "val3"};
+ Redis::StreamEntryID id3;
+ s = stream->Add(name, add_options, values3, &id3);
+ EXPECT_TRUE(s.ok());
+ add_options.entry_id = Redis::NewStreamEntryID{123459, 0};
+ std::vector<std::string> values4 = {"key4", "val4"};
+ Redis::StreamEntryID id4;
+ s = stream->Add(name, add_options, values4, &id4);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamTrimOptions options;
+ options.strategy = Redis::StreamTrimStrategy::MaxLen;
+ options.max_len = 0;
+ uint64_t trimmed;
+ s = stream->Trim(name, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+
+ Redis::StreamInfo info;
+ s = stream->GetStreamInfo(name, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.size, 0);
+ EXPECT_EQ(info.last_generated_id.ToString(), id4.ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), id4.ToString());
+ EXPECT_EQ(info.entries_added, 4);
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), "0-0");
+ EXPECT_FALSE(info.first_entry);
+ EXPECT_FALSE(info.last_entry);
+ EXPECT_EQ(info.entries.size(), 0);
+}
diff --git a/tests/tcl/tests/support/util.tcl b/tests/tcl/tests/support/util.tcl
index 9dc6204..b59139c 100644
--- a/tests/tcl/tests/support/util.tcl
+++ b/tests/tcl/tests/support/util.tcl
@@ -711,3 +711,19 @@ proc string2printable s {
set res "\"$res\""
return $res
}
+
+proc wait_for_blocked_client {} {
+ wait_for_condition 50 100 {
+ [s blocked_clients] ne 0
+ } else {
+ fail "no blocked clients"
+ }
+}
+
+proc wait_for_blocked_clients_count {count {maxtries 100} {delay 10}} {
+ wait_for_condition $maxtries $delay {
+ [s blocked_clients] == $count
+ } else {
+ fail "Timeout waiting for blocked clients"
+ }
+}
diff --git a/tests/tcl/tests/test_helper.tcl b/tests/tcl/tests/test_helper.tcl
index eea6b4f..8d07885 100644
--- a/tests/tcl/tests/test_helper.tcl
+++ b/tests/tcl/tests/test_helper.tcl
@@ -47,6 +47,7 @@ set ::all_tests {
unit/type/hash
unit/type/sint
unit/type/bitmap
+ unit/type/stream
unit/multi
unit/expire
unit/quit
diff --git a/tests/tcl/tests/unit/command.tcl b/tests/tcl/tests/unit/command.tcl
index 8579ac9..5d3da15 100644
--- a/tests/tcl/tests/unit/command.tcl
+++ b/tests/tcl/tests/unit/command.tcl
@@ -16,9 +16,9 @@
# under the License.
start_server {tags {"command"}} {
- test {kvrocks has 172 commands currently} {
+ test {kvrocks has 180 commands currently} {
r command count
- } {172}
+ } {180}
test {acquire GET command info by COMMAND INFO} {
set e [lindex [r command info get] 0]
diff --git a/tests/tcl/tests/unit/type/stream.tcl b/tests/tcl/tests/unit/type/stream.tcl
new file mode 100644
index 0000000..48b02ce
--- /dev/null
+++ b/tests/tcl/tests/unit/type/stream.tcl
@@ -0,0 +1,658 @@
+# 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.
+
+# Copyright (c) 2006-2020, Salvatore Sanfilippo
+# See bundled license file licenses/LICENSE.redis for details.
+
+# This file is copied and modified from the Redis project
+# https://github.com/redis/redis/blob/unstable/tests/unit/type/stream.tcl
+
+# return value is like strcmp() and similar.
+proc streamCompareID {a b} {
+ if {$a eq $b} {return 0}
+ lassign [split $a -] a_ms a_seq
+ lassign [split $b -] b_ms b_seq
+ if {$a_ms > $b_ms} {return 1}
+ if {$a_ms < $b_ms} {return -1}
+ # Same ms case, compare seq.
+ if {$a_seq > $b_seq} {return 1}
+ if {$a_seq < $b_seq} {return -1}
+}
+
+# return the ID immediately greater than the specified one.
+# Note that this function does not care to handle 'seq' overflow
+# since it's a 64 bit value.
+proc streamNextID {id} {
+ lassign [split $id -] ms seq
+ incr seq
+ join [list $ms $seq] -
+}
+
+# Generate a random stream entry ID with the ms part between min and max
+# and a low sequence number (0 - 999 range), in order to stress test
+# XRANGE against a Tcl implementation implementing the same concept
+# with Tcl-only code in a linear array.
+proc streamRandomID {min_id max_id} {
+ lassign [split $min_id -] min_ms min_seq
+ lassign [split $max_id -] max_ms max_seq
+ set delta [expr {$max_ms-$min_ms+1}]
+ set ms [expr {$min_ms+[randomInt $delta]}]
+ set seq [randomInt 1000]
+ return $ms-$seq
+}
+
+# Tcl-side implementation of XRANGE to perform fuzz testing in the Redis
+# XRANGE implementation.
+proc streamSimulateXRANGE {items start end} {
+ set res {}
+ foreach i $items {
+ set this_id [lindex $i 0]
+ if {[streamCompareID $this_id $start] >= 0} {
+ if {[streamCompareID $this_id $end] <= 0} {
+ lappend res $i
+ }
+ }
+ }
+ return $res
+}
+
+set content {} ;# Will be populated with Tcl side copy of the stream content.
+
+start_server {
+ tags {"stream"}
+} {
+ test "XADD wrong number of args" {
+ assert_error {*wrong number of arguments*} {r XADD mystream}
+ assert_error {*wrong number of arguments*} {r XADD mystream *}
+ assert_error {*wrong number of arguments*} {r XADD mystream * field}
+ }
+
+ test {XADD can add entries into a stream that XRANGE can fetch} {
+ r XADD mystream * item 1 value a
+ r XADD mystream * item 2 value b
+ assert_equal 2 [r XLEN mystream]
+ set items [r XRANGE mystream - +]
+ assert_equal [lindex $items 0 1] {item 1 value a}
+ assert_equal [lindex $items 1 1] {item 2 value b}
+ }
+
+ test {XADD IDs are incremental} {
+ set id1 [r XADD mystream * item 1 value a]
+ set id2 [r XADD mystream * item 2 value b]
+ set id3 [r XADD mystream * item 3 value c]
+ assert {[streamCompareID $id1 $id2] == -1}
+ assert {[streamCompareID $id2 $id3] == -1}
+ }
+
+ test {XADD IDs are incremental when ms is the same as well} {
+ r multi
+ r XADD mystream * item 1 value a
+ r XADD mystream * item 2 value b
+ r XADD mystream * item 3 value c
+ lassign [r exec] id1 id2 id3
+ assert {[streamCompareID $id1 $id2] == -1}
+ assert {[streamCompareID $id2 $id3] == -1}
+ }
+
+ test {XADD IDs correctly report an error when overflowing} {
+ r DEL mystream
+ r xadd mystream 18446744073709551615-18446744073709551615 a b
+ assert_error ERR* {r xadd mystream * c d}
+ }
+
+ test {XADD auto-generated sequence is incremented for last ID} {
+ r DEL mystream
+ set id1 [r XADD mystream 123-456 item 1 value a]
+ set id2 [r XADD mystream 123-* item 2 value b]
+ lassign [split $id2 -] _ seq
+ assert {$seq == 457}
+ assert {[streamCompareID $id1 $id2] == -1}
+ }
+
+ test {XADD auto-generated sequence is zero for future timestamp ID} {
+ r DEL mystream
+ set id1 [r XADD mystream 123-456 item 1 value a]
+ set id2 [r XADD mystream 789-* item 2 value b]
+ lassign [split $id2 -] _ seq
+ assert {$seq == 0}
+ assert {[streamCompareID $id1 $id2] == -1}
+ }
+
+ test {XADD auto-generated sequence can't be smaller than last ID} {
+ r DEL mystream
+ r XADD mystream 123-456 item 1 value a
+ assert_error ERR* {r XADD mystream 42-* item 2 value b}
+ }
+
+ test {XADD auto-generated sequence can't overflow} {
+ r DEL mystream
+ r xadd mystream 1-18446744073709551615 a b
+ assert_error ERR* {r xadd mystream 1-* c d}
+ }
+
+ test {XADD 0-* should succeed} {
+ r DEL mystream
+ set id [r xadd mystream 0-* a b]
+ lassign [split $id -] _ seq
+ assert {$seq == 1}
+ }
+
+ test {XADD with MAXLEN option} {
+ r DEL mystream
+ for {set j 0} {$j < 1000} {incr j} {
+ if {rand() < 0.9} {
+ r XADD mystream MAXLEN 5 * xitem $j
+ } else {
+ r XADD mystream MAXLEN 5 * yitem $j
+ }
+ }
+ assert {[r xlen mystream] == 5}
+ set res [r xrange mystream - +]
+ set expected 995
+ foreach r $res {
+ assert {[lindex $r 1 1] == $expected}
+ incr expected
+ }
+ }
+
+ test {XADD with MAXLEN option and the '=' argument} {
+ r DEL mystream
+ for {set j 0} {$j < 1000} {incr j} {
+ if {rand() < 0.9} {
+ r XADD mystream MAXLEN = 5 * xitem $j
+ } else {
+ r XADD mystream MAXLEN = 5 * yitem $j
+ }
+ }
+ assert {[r XLEN mystream] == 5}
+ }
+
+ test {XADD with NOMKSTREAM option} {
+ r DEL mystream
+ assert_equal "" [r XADD mystream NOMKSTREAM * item 1 value a]
+ assert_equal 0 [r EXISTS mystream]
+ r XADD mystream * item 1 value a
+ r XADD mystream NOMKSTREAM * item 2 value b
+ assert_equal 2 [r XLEN mystream]
+ set items [r XRANGE mystream - +]
+ assert_equal [lindex $items 0 1] {item 1 value a}
+ assert_equal [lindex $items 1 1] {item 2 value b}
+ }
+
+ test {XADD with MINID option} {
+ r DEL mystream
+ for {set j 1} {$j < 1001} {incr j} {
+ set minid 1000
+ if {$j >= 5} {
+ set minid [expr {$j-5}]
+ }
+ if {rand() < 0.9} {
+ r XADD mystream MINID $minid $j xitem $j
+ } else {
+ r XADD mystream MINID $minid $j yitem $j
+ }
+ }
+ assert {[r xlen mystream] == 6}
+ set res [r xrange mystream - +]
+ set expected 995
+ foreach r $res {
+ assert {[lindex $r 1 1] == $expected}
+ incr expected
+ }
+ }
+
+ test {XTRIM with MINID option} {
+ r DEL mystream
+ r XADD mystream 1-0 f v
+ r XADD mystream 2-0 f v
+ r XADD mystream 3-0 f v
+ r XADD mystream 4-0 f v
+ r XADD mystream 5-0 f v
+ r XTRIM mystream MINID = 3-0
+ assert_equal [r XRANGE mystream - +] {{3-0 {f v}} {4-0 {f v}} {5-0 {f v}}}
+ }
+
+ test {XTRIM with MINID option, big delta from master record} {
+ r DEL mystream
+ r XADD mystream 1-0 f v
+ r XADD mystream 1641544570597-0 f v
+ r XADD mystream 1641544570597-1 f v
+ r XTRIM mystream MINID 1641544570597-0
+ assert_equal [r XRANGE mystream - +] {{1641544570597-0 {f v}} {1641544570597-1 {f v}}}
+ }
+
+ proc insert_into_stream_key {key {count 10000}} {
+ r multi
+ for {set j 0} {$j < $count} {incr j} {
+ # From time to time insert a field with a different set
+ # of fields in order to stress the stream compression code.
+ if {rand() < 0.9} {
+ r XADD $key * item $j
+ } else {
+ r XADD $key * item $j otherfield foo
+ }
+ }
+ r exec
+ }
+
+ test {XADD mass insertion and XLEN} {
+ r DEL mystream
+ insert_into_stream_key mystream
+
+ set items [r XRANGE mystream - +]
+ for {set j 0} {$j < 10000} {incr j} {
+ assert {[lrange [lindex $items $j 1] 0 1] eq [list item $j]}
+ }
+ assert {[r xlen mystream] == $j}
+ }
+
+ test {XADD with ID 0-0} {
+ r DEL otherstream
+ catch {r XADD otherstream 0-0 k v} err
+ assert {[r EXISTS otherstream] == 0}
+ }
+
+ test {XRANGE COUNT works as expected} {
+ assert {[llength [r xrange mystream - + COUNT 10]] == 10}
+ }
+
+ test {XREVRANGE COUNT works as expected} {
+ assert {[llength [r xrevrange mystream + - COUNT 10]] == 10}
+ }
+
+ test {XRANGE can be used to iterate the whole stream} {
+ set last_id "-"
+ set j 0
+ while 1 {
+ set elements [r xrange mystream $last_id + COUNT 100]
+ if {[llength $elements] == 0} break
+ foreach e $elements {
+ assert {[lrange [lindex $e 1] 0 1] eq [list item $j]}
+ incr j;
+ }
+ set last_id [streamNextID [lindex $elements end 0]]
+ }
+ assert {$j == 10000}
+ }
+
+ test {XREVRANGE returns the reverse of XRANGE} {
+ assert {[r xrange mystream - +] == [lreverse [r xrevrange mystream + -]]}
+ }
+
+ test {XRANGE exclusive ranges} {
+ set ids {0-1 0-18446744073709551615 1-0 42-0 42-42
+ 18446744073709551615-18446744073709551614
+ 18446744073709551615-18446744073709551615}
+ set total [llength $ids]
+ r multi
+ r DEL vipstream
+ foreach id $ids {
+ r XADD vipstream $id foo bar
+ }
+ r exec
+ assert {[llength [r xrange vipstream - +]] == $total}
+ assert {[llength [r xrange vipstream ([lindex $ids 0] +]] == $total-1}
+ assert {[llength [r xrange vipstream - ([lindex $ids $total-1]]] == $total-1}
+ assert {[llength [r xrange vipstream (0-1 (1-0]] == 1}
+ assert {[llength [r xrange vipstream (1-0 (42-42]] == 1}
+ catch {r xrange vipstream (- +} e
+ assert_match {ERR*} $e
+ catch {r xrange vipstream - (+} e
+ assert_match {ERR*} $e
+ catch {r xrange vipstream (18446744073709551615-18446744073709551615 +} e
+ assert_match {ERR*} $e
+ catch {r xrange vipstream - (0-0} e
+ assert_match {ERR*} $e
+ }
+
+ test {XREAD with non empty stream} {
+ set res [r XREAD COUNT 1 STREAMS mystream 0-0]
+ assert {[lrange [lindex $res 0 1 0 1] 0 1] eq {item 0}}
+ }
+
+ test {Non blocking XREAD with empty streams} {
+ set res [r XREAD STREAMS s1{t} s2{t} 0-0 0-0]
+ assert {$res eq {}}
+ }
+
+ test {XREAD with non empty second stream} {
+ insert_into_stream_key mystream{t}
+ set res [r XREAD COUNT 1 STREAMS nostream{t} mystream{t} 0-0 0-0]
+ assert {[lindex $res 0 0] eq {mystream{t}}}
+ assert {[lrange [lindex $res 0 1 0 1] 0 1] eq {item 0}}
+ }
+
+ test {Blocking XREAD waiting new data} {
+ r XADD s2{t} * old abcd1234
+ set rd [redis_deferring_client]
+ $rd XREAD BLOCK 20000 STREAMS s1{t} s2{t} s3{t} $ $ $
+ wait_for_blocked_client
+ r XADD s2{t} * new abcd1234
+ set res [$rd read]
+ assert {[lindex $res 0 0] eq {s2{t}}}
+ assert {[lindex $res 0 1 0 1] eq {new abcd1234}}
+ $rd close
+ }
+
+ test {Blocking XREAD waiting old data} {
+ set rd [redis_deferring_client]
+ $rd XREAD BLOCK 20000 STREAMS s1{t} s2{t} s3{t} $ 0-0 $
+ r XADD s2{t} * foo abcd1234
+ set res [$rd read]
+ assert {[lindex $res 0 0] eq {s2{t}}}
+ assert {[lindex $res 0 1 0 1] eq {old abcd1234}}
+ $rd close
+ }
+
+ test {Blocking XREAD will not reply with an empty array} {
+ r del s1
+ r XADD s1 666 f v
+ r XADD s1 667 f2 v2
+ r XDEL s1 667
+ set rd [redis_deferring_client]
+ $rd XREAD BLOCK 10 STREAMS s1 666
+ after 20
+ assert {[$rd read] == {}} ;# before the fix, client didn't even block, but was served synchronously with {s1 {}}
+ $rd close
+ }
+
+ test "Blocking XREAD for stream that ran dry (issue #5299)" {
+ set rd [redis_deferring_client]
+
+ # Add a entry then delete it, now stream's last_id is 666.
+ r DEL mystream
+ r XADD mystream 666 key value
+ r XDEL mystream 666
+
+ # Pass a ID smaller than stream's last_id, released on timeout.
+ $rd XREAD BLOCK 10 STREAMS mystream 665
+ assert_equal [$rd read] {}
+
+ # Throw an error if the ID equal or smaller than the last_id.
+ assert_error ERR*equal*smaller* {r XADD mystream 665 key value}
+ assert_error ERR*equal*smaller* {r XADD mystream 666 key value}
+
+ # Entered blocking state and then release because of the new entry.
+ $rd XREAD BLOCK 0 STREAMS mystream 665
+ wait_for_blocked_clients_count 1
+ r XADD mystream 667 key value
+ assert_equal [$rd read] {{mystream {{667-0 {key value}}}}}
+
+ $rd close
+ }
+
+ #test "XREAD: XADD + DEL should not awake client" {
+ # set rd [redis_deferring_client]
+ # r del s1
+ # $rd XREAD BLOCK 20000 STREAMS s1 $
+ # wait_for_blocked_clients_count 1
+ # r multi
+ # r XADD s1 * old abcd1234
+ # r DEL s1
+ # r exec
+ # r XADD s1 * new abcd1234
+ # set res [$rd read]
+ # assert {[lindex $res 0 0] eq {s1}}
+ # assert {[lindex $res 0 1 0 1] eq {new abcd1234}}
+ # $rd close
+ #}
+
+ #test "XREAD: XADD + DEL + LPUSH should not awake client" {
+ # set rd [redis_deferring_client]
+ # r del s1
+ # $rd XREAD BLOCK 20000 STREAMS s1 $
+ # wait_for_blocked_clients_count 1
+ # r multi
+ # r XADD s1 * old abcd1234
+ # r DEL s1
+ # r LPUSH s1 foo bar
+ # r exec
+ # r DEL s1
+ # r XADD s1 * new abcd1234
+ # set res [$rd read]
+ # assert {[lindex $res 0 0] eq {s1}}
+ # assert {[lindex $res 0 1 0 1] eq {new abcd1234}}
+ # $rd close
+ #}
+
+ test {XREAD with same stream name multiple times should work} {
+ r XADD s2 * old abcd1234
+ set rd [redis_deferring_client]
+ $rd XREAD BLOCK 20000 STREAMS s2 s2 s2 $ $ $
+ wait_for_blocked_clients_count 1
+ r XADD s2 * new abcd1234
+ set res [$rd read]
+ assert {[lindex $res 0 0] eq {s2}}
+ assert {[lindex $res 0 1 0 1] eq {new abcd1234}}
+ $rd close
+ }
+
+ #test {XREAD + multiple XADD inside transaction} {
+ # r XADD s2 * old abcd1234
+ # set rd [redis_deferring_client]
+ # $rd XREAD BLOCK 20000 STREAMS s2 s2 s2 $ $ $
+ # wait_for_blocked_clients_count 1
+ # r MULTI
+ # r XADD s2 * field one
+ # r XADD s2 * field two
+ # r XADD s2 * field three
+ # r EXEC
+ # set res [$rd read]
+ # assert {[lindex $res 0 0] eq {s2}}
+ # assert {[lindex $res 0 1 0 1] eq {field one}}
+ # assert {[lindex $res 0 1 1 1] eq {field two}}
+ # $rd close
+ #}
+
+ test {XDEL basic test} {
+ r del somestream
+ r xadd somestream * foo value0
+ set id [r xadd somestream * foo value1]
+ r xadd somestream * foo value2
+ r xdel somestream $id
+ assert {[r xlen somestream] == 2}
+ set result [r xrange somestream - +]
+ assert {[lindex $result 0 1 1] eq {value0}}
+ assert {[lindex $result 1 1 1] eq {value2}}
+ }
+
+ # Here the idea is to check the consistency of the stream data structure
+ # as we remove all the elements down to zero elements.
+ test {XDEL fuzz test} {
+ r del somestream
+ set ids {}
+ set x 0; # Length of the stream
+ while 1 {
+ lappend ids [r xadd somestream * item $x]
+ incr x
+ # Add enough elements to have a few radix tree nodes inside the stream.
+ if {[dict get [r xinfo stream somestream] length] > 500} break
+ }
+
+ # Now remove all the elements till we reach an empty stream
+ # and after every deletion, check that the stream is sane enough
+ # to report the right number of elements with XRANGE: this will also
+ # force accessing the whole data structure to check sanity.
+ assert {[r xlen somestream] == $x}
+
+ # We want to remove elements in random order to really test the
+ # implementation in a better way.
+ set ids [lshuffle $ids]
+ foreach id $ids {
+ assert {[r xdel somestream $id] == 1}
+ incr x -1
+ assert {[r xlen somestream] == $x}
+ # The test would be too slow calling XRANGE for every iteration.
+ # Do it every 100 removal.
+ if {$x % 100 == 0} {
+ set res [r xrange somestream - +]
+ assert {[llength $res] == $x}
+ }
+ }
+ }
+
+ test {XRANGE fuzzing} {
+ set items [r XRANGE mystream{t} - +]
+ set low_id [lindex $items 0 0]
+ set high_id [lindex $items end 0]
+ for {set j 0} {$j < 100} {incr j} {
+ set start [streamRandomID $low_id $high_id]
+ set end [streamRandomID $low_id $high_id]
+ set range [r xrange mystream{t} $start $end]
+ set tcl_range [streamSimulateXRANGE $items $start $end]
+ if {$range ne $tcl_range} {
+ puts "*** WARNING *** - XRANGE fuzzing mismatch: $start - $end"
+ puts "---"
+ puts "XRANGE: '$range'"
+ puts "---"
+ puts "TCL: '$tcl_range'"
+ puts "---"
+ fail "XRANGE fuzzing failed, check logs for details"
+ }
+ }
+ }
+
+ test {XREVRANGE regression test for issue #5006} {
+ # Add non compressed entries
+ r xadd teststream 1234567891230 key1 value1
+ r xadd teststream 1234567891240 key2 value2
+ r xadd teststream 1234567891250 key3 value3
+
+ # Add SAMEFIELD compressed entries
+ r xadd teststream2 1234567891230 key1 value1
+ r xadd teststream2 1234567891240 key1 value2
+ r xadd teststream2 1234567891250 key1 value3
+
+ assert_equal [r xrevrange teststream 1234567891245 -] {{1234567891240-0 {key2 value2}} {1234567891230-0 {key1 value1}}}
+
+ assert_equal [r xrevrange teststream2 1234567891245 -] {{1234567891240-0 {key1 value2}} {1234567891230-0 {key1 value1}}}
+ }
+
+ test {XREAD streamID edge (no-blocking)} {
+ r del x
+ r XADD x 1-1 f v
+ r XADD x 1-18446744073709551615 f v
+ r XADD x 2-1 f v
+ set res [r XREAD BLOCK 0 STREAMS x 1-18446744073709551615]
+ assert {[lindex $res 0 1 0] == {2-1 {f v}}}
+ }
+
+ test {XREAD streamID edge (blocking)} {
+ r del x
+ set rd [redis_deferring_client]
+ $rd XREAD BLOCK 0 STREAMS x 1-18446744073709551615
+ wait_for_blocked_clients_count 1
+ r XADD x 1-1 f v
+ r XADD x 1-18446744073709551615 f v
+ r XADD x 2-1 f v
+ set res [$rd read]
+ assert {[lindex $res 0 1 0] == {2-1 {f v}}}
+ $rd close
+ }
+
+ test {XADD streamID edge} {
+ r del x
+ r XADD x 2577343934890-18446744073709551615 f v ;# we need the timestamp to be in the future
+ r XADD x * f2 v2
+ assert_equal [r XRANGE x - +] {{2577343934890-18446744073709551615 {f v}} {2577343934891-0 {f2 v2}}}
+ }
+
+ test {XTRIM with MAXLEN option basic test} {
+ r DEL mystream
+ for {set j 0} {$j < 1000} {incr j} {
+ if {rand() < 0.9} {
+ r XADD mystream * xitem $j
+ } else {
+ r XADD mystream * yitem $j
+ }
+ }
+ r XTRIM mystream MAXLEN 666
+ assert {[r XLEN mystream] == 666}
+ r XTRIM mystream MAXLEN = 555
+ assert {[r XLEN mystream] == 555}
+ }
+
+ test {XADD with LIMIT consecutive calls} {
+ r del mystream
+ for {set j 0} {$j < 100} {incr j} {
+ r XADD mystream * xitem v
+ }
+ r XADD mystream MAXLEN = 55 * xitem v
+ assert {[r xlen mystream] == 55}
+ r XADD mystream MAXLEN = 55 * xitem v
+ assert {[r xlen mystream] == 55}
+ }
+}
+
+start_server {tags {"stream offset"}} {
+ test {XADD advances the entries-added counter and sets the recorded-first-entry-id} {
+ r DEL x
+ r XADD x 1-0 data a
+
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply entries-added] 1
+ assert_equal [dict get $reply recorded-first-entry-id] "1-0"
+
+ r XADD x 2-0 data a
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply entries-added] 2
+ assert_equal [dict get $reply recorded-first-entry-id] "1-0"
+ }
+
+ test {XDEL/TRIM are reflected by recorded first entry} {
+ r DEL x
+ r XADD x 1-0 data a
+ r XADD x 2-0 data a
+ r XADD x 3-0 data a
+ r XADD x 4-0 data a
+ r XADD x 5-0 data a
+
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply entries-added] 5
+ assert_equal [dict get $reply recorded-first-entry-id] "1-0"
+
+ r XDEL x 2-0
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply recorded-first-entry-id] "1-0"
+
+ r XDEL x 1-0
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply recorded-first-entry-id] "3-0"
+
+ r XTRIM x MAXLEN = 2
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply recorded-first-entry-id] "4-0"
+ }
+
+ test {Maxmimum XDEL ID behaves correctly} {
+ r DEL x
+ r XADD x 1-0 data a
+ r XADD x 2-0 data b
+ r XADD x 3-0 data c
+
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply max-deleted-entry-id] "0-0"
+
+ r XDEL x 2-0
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply max-deleted-entry-id] "2-0"
+
+ r XDEL x 1-0
+ set reply [r XINFO STREAM x FULL]
+ assert_equal [dict get $reply max-deleted-entry-id] "2-0"
+ }
+}