From 0dcac993972c4e5cbf0b9f36ee46be7c6093a71e Mon Sep 17 00:00:00 2001 From: ithewei Date: Fri, 10 Jul 2026 20:14:21 +0800 Subject: [PATCH 1/4] feat: add redis client module Add a standalone single-node Redis C++ client under the repository root redis/, following the same .h + .cpp split as other libhv modules. - RedisMessage: RESP2 encoder/parser, RedisReply/RedisResult model - AsyncRedisClient: async long-connection client with FIFO pending queue, timeout, reconnect and batch command support - RedisClient: sync + async facade with typed helpers (get/set/del/exists/ expire/hget/hset/publish) and raw command interfaces - RedisPipeline / RedisTransaction: batch and MULTI/EXEC/DISCARD support - RedisSubscriber: dedicated pub/sub connection with resubscribe on reconnect Gate the module behind WITH_REDIS (default off), wired across Makefile, CMake and Bazel, and only available when WITH_EVPP is enabled. Add unit tests, example programs and Chinese docs (docs/cn/RedisClient.md). --- BUILD.bazel | 28 ++ BUILD.md | 6 + CMakeLists.txt | 7 +- Makefile | 34 +- Makefile.vars | 7 + README-CN.md | 15 + README.md | 15 + cmake/vars.cmake | 9 + config.ini | 1 + configure | 1 + docs/PLAN.md | 2 +- docs/cn/README.md | 1 + docs/cn/RedisClient.md | 304 ++++++++++++++ examples/BUILD.bazel | 23 ++ examples/CMakeLists.txt | 12 + examples/redis_client_test.cpp | 56 +++ examples/redis_subscriber_test.cpp | 53 +++ redis/AsyncRedisClient.cpp | 581 ++++++++++++++++++++++++++ redis/AsyncRedisClient.h | 53 +++ redis/RedisClient.cpp | 296 ++++++++++++++ redis/RedisClient.h | 101 +++++ redis/RedisMessage.cpp | 296 ++++++++++++++ redis/RedisMessage.h | 85 ++++ redis/RedisPipeline.cpp | 70 ++++ redis/RedisPipeline.h | 27 ++ redis/RedisSubscriber.cpp | 588 +++++++++++++++++++++++++++ redis/RedisSubscriber.h | 45 ++ redis/RedisTransaction.cpp | 80 ++++ redis/RedisTransaction.h | 28 ++ scripts/unittest.sh | 10 + unittest/CMakeLists.txt | 19 + unittest/redis_async_client_test.cpp | 388 ++++++++++++++++++ unittest/redis_batch_test.cpp | 356 ++++++++++++++++ unittest/redis_client_test.cpp | 358 ++++++++++++++++ unittest/redis_protocol_test.cpp | 167 ++++++++ unittest/redis_subscriber_test.cpp | 128 ++++++ unittest/redis_test_server.cpp | 227 +++++++++++ unittest/redis_test_server.h | 30 ++ 38 files changed, 4504 insertions(+), 3 deletions(-) create mode 100644 docs/cn/RedisClient.md create mode 100644 examples/redis_client_test.cpp create mode 100644 examples/redis_subscriber_test.cpp create mode 100644 redis/AsyncRedisClient.cpp create mode 100644 redis/AsyncRedisClient.h create mode 100644 redis/RedisClient.cpp create mode 100644 redis/RedisClient.h create mode 100644 redis/RedisMessage.cpp create mode 100644 redis/RedisMessage.h create mode 100644 redis/RedisPipeline.cpp create mode 100644 redis/RedisPipeline.h create mode 100644 redis/RedisSubscriber.cpp create mode 100644 redis/RedisSubscriber.h create mode 100644 redis/RedisTransaction.cpp create mode 100644 redis/RedisTransaction.h create mode 100644 unittest/redis_async_client_test.cpp create mode 100644 unittest/redis_batch_test.cpp create mode 100644 unittest/redis_client_test.cpp create mode 100644 unittest/redis_protocol_test.cpp create mode 100644 unittest/redis_subscriber_test.cpp create mode 100644 unittest/redis_test_server.cpp create mode 100644 unittest/redis_test_server.h diff --git a/BUILD.bazel b/BUILD.bazel index 0e5b8b6d0..1a5374565 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -100,6 +100,15 @@ config_setting( visibility = [":__subpackages__"], ) +config_setting( + name = "with_redis", + define_values = { + "WITH_EVPP": "ON", + "WITH_REDIS": "ON", + }, + visibility = [":__subpackages__"], +) + config_setting( name = "enable_uds", define_values = {"ENABLE_UDS": "ON"} @@ -189,6 +198,9 @@ HEADERS_DIRS = ["base", "ssl", "event"] + select({ }) + select({ "with_evpp": ["cpputil", "evpp"], "//conditions:default": [], +}) + select({ + "with_redis": ["redis"], + "//conditions:default": [], }) + select({ "with_http": ["http"], "//conditions:default": [], @@ -305,6 +317,7 @@ CPPUTIL_HEADERS = [ "cpputil/iniparser.h", "cpputil/json.hpp", "cpputil/singleton.h", + "cpputil/LRUCache.h", "cpputil/ThreadLocalStorage.h", ] @@ -322,6 +335,15 @@ EVPP_HEADERS = [ "evpp/UdpServer.h", ] +REDIS_HEADERS = [ + "redis/RedisMessage.h", + "redis/AsyncRedisClient.h", + "redis/RedisClient.h", + "redis/RedisPipeline.h", + "redis/RedisTransaction.h", + "redis/RedisSubscriber.h", +] + PROTOCOL_HEADERS = [ "protocol/icmp.h", "protocol/dns.h", @@ -372,6 +394,9 @@ HEADERS = ["hv.h", ":config", "hexport.h"] + BASE_HEADERS + SSL_HEADERS + EVENT_ }) + select({ "with_evpp": CPPUTIL_HEADERS + EVPP_HEADERS, "//conditions:default": [], +}) + select({ + "with_redis": REDIS_HEADERS, + "//conditions:default": [], }) + select({ "with_http": HTTP_HEADERS, "//conditions:default": [], @@ -412,6 +437,9 @@ SRCS = CORE_SRCS + glob(["util/*.h", "util/*.c", "util/*.cpp"], exclude = ["util }) + select({ "with_evpp": glob(["cpputil/*.h", "cpputil/*.c", "cpputil/*.cpp", "evpp/*.h", "evpp/*.c", "evpp/*.cpp"], exclude = ["cpputil/*_test.c", "evpp/*_test.c", "evpp/*_test.cpp"]), "//conditions:default": [], +}) + select({ + "with_redis": glob(["redis/*.h", "redis/*.c", "redis/*.cpp"], exclude = ["redis/*_test.c", "redis/*_test.cpp"]), + "//conditions:default": [], }) + select({ "with_http": glob(["http/*.h", "http/*.c", "http/*.cpp"], exclude = ["http/*_test.c"]), "//conditions:default": [], diff --git a/BUILD.md b/BUILD.md index c2219ae13..d501437e1 100644 --- a/BUILD.md +++ b/BUILD.md @@ -156,6 +156,12 @@ make clean && make make clean && make ``` +### compile WITH_REDIS +``` +./configure --with-redis +make clean && make +``` + ### More ``` ./configure --help diff --git a/CMakeLists.txt b/CMakeLists.txt index 8d8502e34..ffbe683fe 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -17,6 +17,7 @@ option(WITH_HTTP "compile http" ON) option(WITH_HTTP_SERVER "compile http/server" ON) option(WITH_HTTP_CLIENT "compile http/client" ON) option(WITH_MQTT "compile mqtt" OFF) +option(WITH_REDIS "compile redis" OFF) option(ENABLE_UDS "Unix Domain Socket" OFF) option(USE_MULTIMAP "MultiMap" OFF) @@ -203,7 +204,7 @@ if(APPLE) endif() # see Makefile -set(ALL_SRCDIRS . base ssl event event/kcp util cpputil evpp protocol http http/client http/server mqtt) +set(ALL_SRCDIRS . base ssl event event/kcp util cpputil evpp redis protocol http http/client http/server mqtt) set(CORE_SRCDIRS . base ssl event) if(WIN32 OR MINGW) if(WITH_WEPOLL) @@ -225,6 +226,10 @@ endif() if(WITH_EVPP) set(LIBHV_HEADERS ${LIBHV_HEADERS} ${CPPUTIL_HEADERS} ${EVPP_HEADERS}) set(LIBHV_SRCDIRS ${LIBHV_SRCDIRS} cpputil evpp) + if(WITH_REDIS) + set(LIBHV_HEADERS ${LIBHV_HEADERS} ${REDIS_HEADERS}) + set(LIBHV_SRCDIRS ${LIBHV_SRCDIRS} redis) + endif() if(WITH_HTTP) set(LIBHV_HEADERS ${LIBHV_HEADERS} ${HTTP_HEADERS}) set(LIBHV_SRCDIRS ${LIBHV_SRCDIRS} http) diff --git a/Makefile b/Makefile index 894c530ef..f6fa6e583 100644 --- a/Makefile +++ b/Makefile @@ -2,7 +2,7 @@ include config.mk include Makefile.vars MAKEF=$(MAKE) -f Makefile.in -ALL_SRCDIRS=. base ssl event event/kcp util cpputil evpp protocol http http/client http/server mqtt +ALL_SRCDIRS=. base ssl event event/kcp util cpputil evpp redis protocol http http/client http/server mqtt CORE_SRCDIRS=. base ssl event ifeq ($(WITH_KCP), yes) CORE_SRCDIRS += event/kcp @@ -24,6 +24,11 @@ ifeq ($(WITH_EVPP), yes) LIBHV_HEADERS += $(CPPUTIL_HEADERS) $(EVPP_HEADERS) LIBHV_SRCDIRS += cpputil evpp +ifeq ($(WITH_REDIS), yes) +LIBHV_HEADERS += $(REDIS_HEADERS) +LIBHV_SRCDIRS += redis +endif + ifeq ($(WITH_HTTP), yes) LIBHV_HEADERS += $(HTTP_HEADERS) LIBHV_SRCDIRS += http @@ -71,6 +76,9 @@ EXAMPLES = hmain_test htimer_test hloop_test pipe_test \ ifeq ($(WITH_EVPP), yes) EXAMPLES += nmap +ifeq ($(WITH_REDIS), yes) +EXAMPLES += redis_client_example redis_subscriber_example +endif ifeq ($(WITH_HTTP), yes) EXAMPLES += wrk ifeq ($(WITH_HTTP_SERVER), yes) @@ -185,6 +193,16 @@ tinyproxyd: prepare nmap: prepare libhv $(MAKEF) TARGET=$@ SRCDIRS="$(CORE_SRCDIRS) cpputil examples/nmap" DEFINES="PRINT_DEBUG" +ifeq ($(WITH_EVPP), yes) +ifeq ($(WITH_REDIS), yes) +redis_client_example: prepare + $(MAKEF) TARGET=$@ SRCDIRS="$(CORE_SRCDIRS) cpputil evpp redis" SRCS="examples/redis_client_test.cpp" + +redis_subscriber_example: prepare + $(MAKEF) TARGET=$@ SRCDIRS="$(CORE_SRCDIRS) cpputil evpp redis" SRCS="examples/redis_subscriber_test.cpp" +endif +endif + wrk: prepare $(MAKEF) TARGET=$@ SRCDIRS="$(CORE_SRCDIRS) util cpputil evpp http" SRCS="examples/wrk.cpp" @@ -287,6 +305,20 @@ unittest: prepare $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -Iprotocol -o bin/ping unittest/ping_test.c protocol/icmp.c base/hsocket.c base/htime.c -DPRINT_DEBUG $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -Iprotocol -o bin/ftp unittest/ftp_test.c protocol/ftp.c base/hsocket.c base/htime.c $(CC) -g -Wall -O0 -std=c99 -I. -Ibase -Iprotocol -Iutil -o bin/sendmail unittest/sendmail_test.c protocol/smtp.c base/hsocket.c base/htime.c util/base64.c +ifeq ($(WITH_EVPP), yes) +ifeq ($(WITH_REDIS), yes) + $(MAKE) libhv + $(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Ievent -Icpputil -Iredis -o bin/redis_protocol_test unittest/redis_protocol_test.cpp redis/RedisMessage.cpp + $(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Issl -Ievent -Icpputil -Iredis -Ievpp -o bin/redis_async_client_test unittest/redis_async_client_test.cpp unittest/redis_test_server.cpp -Llib -lhv -pthread + $(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Issl -Ievent -Icpputil -Iredis -Ievpp -o bin/redis_client_test unittest/redis_client_test.cpp unittest/redis_test_server.cpp -Llib -lhv -pthread + $(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Issl -Ievent -Icpputil -Iredis -Ievpp -o bin/redis_batch_test unittest/redis_batch_test.cpp unittest/redis_test_server.cpp -Llib -lhv -pthread + $(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Issl -Ievent -Icpputil -Iredis -Ievpp -o bin/redis_subscriber_test unittest/redis_subscriber_test.cpp unittest/redis_test_server.cpp -Llib -lhv -pthread +else + $(RM) bin/redis_protocol_test bin/redis_async_client_test bin/redis_client_test bin/redis_batch_test bin/redis_subscriber_test +endif +else + $(RM) bin/redis_protocol_test bin/redis_async_client_test bin/redis_client_test bin/redis_batch_test bin/redis_subscriber_test +endif run-unittest: unittest bash scripts/unittest.sh diff --git a/Makefile.vars b/Makefile.vars index c21d9aa5a..1760c1e8b 100644 --- a/Makefile.vars +++ b/Makefile.vars @@ -64,6 +64,13 @@ EVPP_HEADERS = evpp/Buffer.h\ evpp/UdpClient.h\ evpp/UdpServer.h\ +REDIS_HEADERS = redis/RedisMessage.h\ + redis/AsyncRedisClient.h\ + redis/RedisClient.h\ + redis/RedisPipeline.h\ + redis/RedisTransaction.h\ + redis/RedisSubscriber.h\ + PROTOCOL_HEADERS = protocol/icmp.h\ protocol/dns.h\ protocol/ftp.h\ diff --git a/README-CN.md b/README-CN.md index 81af89c01..1dca3a315 100644 --- a/README-CN.md +++ b/README-CN.md @@ -63,6 +63,7 @@ - HTTP支持RESTful风格、路由、中间件、keep-alive长连接、chunked分块、SSE等特性 - WebSocket服务端/客户端 - MQTT客户端 +- Redis客户端 ## ⌛️ 构建 @@ -397,6 +398,18 @@ int main(int argc, char** argv) { } ``` +### Redis +见 [examples/redis_client_test.cpp](examples/redis_client_test.cpp) 和 [examples/redis_subscriber_test.cpp](examples/redis_subscriber_test.cpp) + +Redis C++ 模块位于仓库根目录 `redis/`，并采用 `.h + .cpp` 分离结构。Redis 默认不编译，需要显式使用 `./configure --with-redis` 或 `cmake -DWITH_REDIS=ON` 打开。 + +```shell +./configure --with-redis +make unittest +cmake -S . -B build -DWITH_EVPP=ON -DWITH_REDIS=ON -DBUILD_UNITTEST=ON +cmake --build build --target redis_protocol_test redis_async_client_test redis_client_test redis_batch_test redis_subscriber_test +``` + ## 🍭 更多示例 ### c版本 @@ -430,6 +443,8 @@ int main(int argc, char** argv) { - HTTP客户端: [examples/http_client_test.cpp](examples/http_client_test.cpp) - WebSocket服务端: [examples/websocket_server_test.cpp](examples/websocket_server_test.cpp) - WebSocket客户端: [examples/websocket_client_test.cpp](examples/websocket_client_test.cpp) +- Redis客户端示例: [examples/redis_client_test.cpp](examples/redis_client_test.cpp) +- Redis订阅示例: [examples/redis_subscriber_test.cpp](examples/redis_subscriber_test.cpp) - protobufRPC示例: [examples/protorpc](examples/protorpc) - Qt中使用libhv示例: [hv-projects/QtDemo](https://github.com/hv-projects/QtDemo) diff --git a/README.md b/README.md index f200cd742..9970198d5 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ but simpler api and richer protocols. - HTTP supports RESTful, router, middleware, keep-alive, chunked, SSE, etc. - WebSocket client/server - MQTT client +- Redis client ## ⌛️ Build @@ -338,6 +339,18 @@ int main(int argc, char** argv) { } ``` +### Redis +see [examples/redis_client_test.cpp](examples/redis_client_test.cpp) and [examples/redis_subscriber_test.cpp](examples/redis_subscriber_test.cpp) + +The Redis C++ module lives in the repository root `redis/` and follows the same `.h + .cpp` split used by other libhv modules. Redis is disabled by default; enable it explicitly with `./configure --with-redis` or `cmake -DWITH_REDIS=ON`. + +```shell +./configure --with-redis +make unittest +cmake -S . -B build -DWITH_EVPP=ON -DWITH_REDIS=ON -DBUILD_UNITTEST=ON +cmake --build build --target redis_protocol_test redis_async_client_test redis_client_test redis_batch_test redis_subscriber_test +``` + ## 🍭 More examples ### c version - [examples/hloop_test.c](examples/hloop_test.c) @@ -370,6 +383,8 @@ int main(int argc, char** argv) { - [examples/http_client_test.cpp](examples/http_client_test.cpp) - [examples/websocket_server_test.cpp](examples/websocket_server_test.cpp) - [examples/websocket_client_test.cpp](examples/websocket_client_test.cpp) +- [examples/redis_client_test.cpp](examples/redis_client_test.cpp) +- [examples/redis_subscriber_test.cpp](examples/redis_subscriber_test.cpp) - [examples/protorpc](examples/protorpc) - [hv-projects/QtDemo](https://github.com/hv-projects/QtDemo) diff --git a/cmake/vars.cmake b/cmake/vars.cmake index 1acf5565f..c3b0451c6 100644 --- a/cmake/vars.cmake +++ b/cmake/vars.cmake @@ -68,6 +68,15 @@ set(EVPP_HEADERS evpp/UdpServer.h ) +set(REDIS_HEADERS + redis/RedisMessage.h + redis/AsyncRedisClient.h + redis/RedisClient.h + redis/RedisPipeline.h + redis/RedisTransaction.h + redis/RedisSubscriber.h +) + set(PROTOCOL_HEADERS protocol/icmp.h protocol/dns.h diff --git a/config.ini b/config.ini index f9a5a43b4..466d3b936 100644 --- a/config.ini +++ b/config.ini @@ -17,6 +17,7 @@ WITH_HTTP=yes WITH_HTTP_SERVER=yes WITH_HTTP_CLIENT=yes WITH_MQTT=no +WITH_REDIS=no # features # base/hsocket.h: Unix Domain Socket diff --git a/configure b/configure index 396762389..9cf1b15f3 100755 --- a/configure +++ b/configure @@ -30,6 +30,7 @@ modules: --with-http-client compile http client module? (DEFAULT: $WITH_HTTP_CLIENT) --with-http-server compile http server module? (DEFAULT: $WITH_HTTP_SERVER) --with-mqtt compile mqtt module? (DEFAULT: $WITH_MQTT) + --with-redis compile redis module? (DEFAULT: $WITH_REDIS) features: --enable-uds enable Unix Domain Socket? (DEFAULT: $ENABLE_UDS) diff --git a/docs/PLAN.md b/docs/PLAN.md index c33d5fb4b..f589d57f8 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -8,10 +8,10 @@ - http client/server: include https http1/x http2 - websocket client/server - mqtt client +- redis client ## Plan -- redis client - async DNS - lua binding - js binding diff --git a/docs/cn/README.md b/docs/cn/README.md index d93c806cc..0f676b4ae 100644 --- a/docs/cn/README.md +++ b/docs/cn/README.md @@ -16,3 +16,4 @@ - [class HttpClient: HTTP客户端类](HttpClient.md) - [class WebSocketServer: WebSocket服务端类](WebSocketServer.md) - [class WebSocketClient: WebSocket客户端类](WebSocketClient.md) +- [class RedisClient: Redis客户端类](RedisClient.md) diff --git a/docs/cn/RedisClient.md b/docs/cn/RedisClient.md new file mode 100644 index 000000000..96b7e3578 --- /dev/null +++ b/docs/cn/RedisClient.md @@ -0,0 +1,304 @@ +Redis 客户端类 + +libhv 的 Redis C++ 模块位于仓库根目录 `redis/`，采用 `.h + .cpp` 分离结构。第一版只支持单机 Redis 与 RESP2 协议，提供同步 / 异步命令、typed helpers、pipeline、transaction 和 pub/sub。 + +Redis 模块默认不编译，需要显式开启： + +```shell +./configure --with-redis && make libhv +# 或 +cmake -S . -B build -DWITH_EVPP=ON -DWITH_REDIS=ON && cmake --build build +``` + +## 结果模型 + +命令的结果严格区分三类语义：客户端错误、Redis 服务端错误回复、nil / 空结果。 + +```c++ +// RESP2 回复类型 +enum RedisReplyType { + REDIS_REPLY_NIL, // nil + REDIS_REPLY_STRING, // 字符串 (SimpleString / BulkString) + REDIS_REPLY_ERROR, // 服务端错误回复 + REDIS_REPLY_INTEGER, // 整数 + REDIS_REPLY_ARRAY, // 数组 +}; + +// 统一底层回复对象 +struct RedisReply { + RedisReplyType type; + std::string str; // 字符串 / 错误信息 + int64_t integer; // 整数 + std::vector elements; // 数组元素 + + bool isNil() const; + bool isError() const; + bool isArray() const; + bool isString() const; + const std::string& error() const; // 错误信息 + const std::string& asString() const; + int64_t asInt() const; + const std::vector& asArray() const; +}; + +// 一次命令调用的统一结果 +struct RedisResult { + int code; // 0 表示客户端侧成功 + RedisReply reply; // 客户端侧成功时保存 Redis 回复 + + bool ok() const; // code == 0 且 reply 不是错误回复 +}; + +// typed helper 的结果 (带具体值 value) +template +struct RedisValueResult { + int code; + RedisReply reply; + T value; + bool has_value; + + bool ok() const; // code == 0 且非错误回复且 has_value + bool isNil() const; // code == 0 且回复为 nil (例如 GET 不存在的 key) +}; +``` + +语义约定： + +- `code == 0`：客户端侧调用成功，`reply` 有效。 +- `code != 0`：客户端侧失败（未连接、连接失败、超时、断线、协议错误、参数非法等），`reply` 无效。 +- `code == 0 && reply.isError()`：成功收到 Redis 服务端错误回复（如 `-ERR`、`-WRONGTYPE`、`-NOAUTH`）。 +- `isNil()`：不是错误，例如 `GET` 不存在的 key。 + +## class RedisClient + +面向多数使用者的主入口，同步 + 异步接口共存，用法与 `HttpClient` 类似。内部使用独立线程运行事件循环，同步接口在内部等待完成，对外无需操作 future。 + +```c++ +class RedisClient { + + RedisClient(); + + // 连接配置 + void setHost(const std::string& host); + void setPort(int port); + void setAuth(const std::string& password); // AUTH + void setDb(int db); // SELECT + void setConnectTimeout(int ms); // 连接超时 + void setTimeout(int ms); // 命令读写超时 + void setReconnect(reconn_setting_t* setting); // 断线重连 + + // 原始命令 (参数列表风格, 主推, binary-safe) + RedisResult command(const RedisCommand& command); // 同步 + int commandAsync(const RedisCommand& command, RedisCallback cb); // 异步 + + // 原始命令 (format 风格, 兼容重载) + RedisResult commandf(const char* fmt, ...); + template + int commandfAsync(const char* fmt, RedisCallback cb, Args... args); + + // pipeline / transaction 工厂 + RedisPipeline pipeline(); + RedisTransaction transaction(); + + // typed helpers (同步 + 异步) + RedisValueResult get(const std::string& key); + int getAsync(const std::string& key, RedisValueCallback cb); + + RedisResult set(const std::string& key, const std::string& value); + int setAsync(const std::string& key, const std::string& value, RedisCallback cb); + + RedisValueResult del(const std::string& key); + RedisValueResult exists(const std::string& key); + RedisValueResult expire(const std::string& key, int seconds); + RedisValueResult hget(const std::string& key, const std::string& field); + RedisValueResult hset(const std::string& key, const std::string& field, const std::string& value); + RedisValueResult publish(const std::string& channel, const std::string& message); + // ...对应的 delAsync / existsAsync / expireAsync / hgetAsync / hsetAsync / publishAsync +}; +``` + +其中 `RedisCommand` 即 `std::vector`，`RedisCallback` 为 `std::function`。 + +typed helpers 是对原始命令的薄封装，复用统一的命令编码、reply 解析与错误语义，不构成第二套协议实现。 + +### 同步用法 + +```c++ +using namespace hv; + +RedisClient client; +client.setHost("127.0.0.1"); +client.setPort(6379); +client.setConnectTimeout(3000); +client.setTimeout(3000); + +// typed helper +if (client.set("key", "hello").ok()) { + RedisValueResult v = client.get("key"); + if (v.ok()) { + printf("key => %s\n", v.value.c_str()); + } else if (v.isNil()) { + printf("key not exists\n"); + } +} + +// 原始命令 +RedisResult r = client.command(RedisCommand{"INCR", "counter"}); +if (r.ok()) { + printf("counter = %lld\n", (long long)r.reply.asInt()); +} +``` + +### 异步用法 + +异步命令采用单连接、FIFO 配对模型：按发送顺序取出对应 callback。回调保证只触发一次。 + +```c++ +client.getAsync("key", [](const RedisValueResult& v) { + if (v.ok()) { + printf("key => %s\n", v.value.c_str()); + } +}); +``` + +> 注意：不要在异步回调线程里再调用同步接口（如 `client.get(...)`），会因处于事件循环线程内被拒绝并返回 `ERR_INVALID_HANDLE`。 + +## class RedisPipeline + +批量命令对象：先累积命令，`exec` 时一次性发送，按顺序返回 N 条回复。整体失败（发送失败 / 超时 / 断线 / 回复不完整）通过 `RedisResult.code` 体现；单条命令的错误回复是 `replies` 中某一项的合法 error reply。 + +```c++ +class RedisPipeline { + void appendCommand(const RedisCommand& command); + RedisResult exec(std::vector* replies = NULL); // 同步 + int execAsync(const RedisRepliesCallback& cb); // 异步 +}; +``` + +```c++ +RedisPipeline pipe = client.pipeline(); +pipe.appendCommand(RedisCommand{"SET", "counter", "1"}); +pipe.appendCommand(RedisCommand{"INCR", "counter"}); + +std::vector replies; +RedisResult result = pipe.exec(&replies); +if (result.ok()) { + printf("INCR => %lld\n", (long long)replies[1].asInt()); // 2 +} +``` + +## class RedisTransaction + +封装 `MULTI` / `EXEC` / `DISCARD`。`exec` 返回事务结果数组。第一版不把 `WATCH` 及其冲突重试作为重点能力。 + +```c++ +class RedisTransaction { + void appendCommand(const RedisCommand& command); + RedisResult exec(std::vector* replies = NULL); // MULTI + 命令 + EXEC + RedisResult discard(); // DISCARD +}; +``` + +```c++ +RedisTransaction tx = client.transaction(); +tx.appendCommand(RedisCommand{"SET", "k", "7"}); +tx.appendCommand(RedisCommand{"GET", "k"}); + +std::vector replies; +RedisResult result = tx.exec(&replies); +if (result.ok()) { + printf("GET => %s\n", replies[1].asString().c_str()); // 7 +} +``` + +## class AsyncRedisClient + +面向事件循环模型的纯异步客户端，风格贴近 `TcpClient` / `AsyncHttpClient`。`RedisClient` 的异步能力即基于它实现。适合高并发、长连接、批量发送场景。 + +```c++ +class AsyncRedisClient { + + AsyncRedisClient(EventLoopPtr loop = NULL); + + // 连接配置 (同 RedisClient) + void setHost(const std::string& host); + void setPort(int port); + void setAuth(const std::string& password); + void setDb(int db); + void setConnectTimeout(int ms); + void setTimeout(int ms); + void setReconnect(reconn_setting_t* setting); + + // 生命周期 + void start(bool wait_threads_started = true); + void stop(bool wait_threads_stopped = true); + bool isConnected() const; + bool isStarted() const; + bool isInLoopThread(); + + // 异步命令 + int command(const RedisCommand& command, RedisCallback cb); + int commandBatch(const std::vector& commands, RedisRepliesCallback cb); + + // 事件回调 + std::function onConnect; + std::function onClose; + std::function onError; +}; +``` + +> 连接断开时，所有尚未完成的 pending 异步请求会统一以客户端错误失败，且不会自动重放未完成命令（Redis 命令可能有副作用，自动重放不安全）。 + +## class RedisSubscriber + +独立订阅客户端，专门处理 Pub/Sub。使用独立连接，不与普通命令连接复用状态机。 + +```c++ +class RedisSubscriber { + + RedisSubscriber(EventLoopPtr loop = NULL); + + // 连接配置 + void setHost(const std::string& host); + void setPort(int port); + void setAuth(const std::string& password); + void setDb(int db); + void setReconnect(reconn_setting_t* setting); + + // 生命周期 + void start(bool wait_threads_started = true); + void stop(bool wait_threads_stopped = true); + + // 订阅 / 退订 + int subscribe(const std::string& channel); + int psubscribe(const std::string& pattern); + int unsubscribe(const std::string& channel); + int punsubscribe(const std::string& pattern); + + // 事件回调 + std::function onMessage; + std::function onSubscribe; + std::function onUnsubscribe; + std::function onError; +}; +``` + +```c++ +RedisSubscriber subscriber; +subscriber.setHost("127.0.0.1"); +subscriber.setPort(6379); +subscriber.onMessage = [](const std::string& channel, const std::string& message) { + printf("%s => %s\n", channel.c_str(), message.c_str()); +}; +subscriber.start(); +subscriber.subscribe("news"); +``` + +> 配置了 reconnect 时，连接恢复后会重放当前订阅集合并继续接收推送；已经显式退订的条目不会自动恢复。 + +## 说明 + +- 第一版明确不做：Redis Sentinel、Redis Cluster、RESP3、coroutine / future-first 的公开 API、`WATCH` 冲突重试、自动重放未完成命令、大而全的全命令 typed API 覆盖。 +- 普通命令流与订阅流分离，是第一版设计中的硬边界。 + +测试代码见 [examples/redis_client_test.cpp](../../examples/redis_client_test.cpp) 和 [examples/redis_subscriber_test.cpp](../../examples/redis_subscriber_test.cpp) diff --git a/examples/BUILD.bazel b/examples/BUILD.bazel index 340de96fe..b8ca983f8 100644 --- a/examples/BUILD.bazel +++ b/examples/BUILD.bazel @@ -100,6 +100,26 @@ cc_binary( deps = ["//:hv"] ) +cc_binary( + name = "redis_client_example", + srcs = ["redis_client_test.cpp"], + deps = ["//:hv"], + target_compatible_with = select({ + "//:with_redis": [], + "//conditions:default": ["@platforms//:incompatible"], + }), +) + +cc_binary( + name = "redis_subscriber_example", + srcs = ["redis_subscriber_test.cpp"], + deps = ["//:hv"], + target_compatible_with = select({ + "//:with_redis": [], + "//conditions:default": ["@platforms//:incompatible"], + }), +) + cc_binary( name = "wrk", srcs = ["wrk.cpp"], @@ -205,6 +225,9 @@ filegroup( ] + select({ "//:with_evpp": [":hmain_test", ":nmap"], "//conditions:default": [], + }) + select({ + "//:with_redis": [":redis_client_example", ":redis_subscriber_example"], + "//conditions:default": [], }) + select({ "//:with_http": [":wrk"], "//conditions:default": [], diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 1e4df94f7..422e61eb4 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -107,6 +107,18 @@ if(WITH_EVPP) target_link_libraries(nmap ${HV_LIBRARIES}) list(APPEND EXAMPLES hmain_test nmap) + + if(WITH_REDIS) + include_directories(../redis) + + add_executable(redis_client_example redis_client_test.cpp) + target_link_libraries(redis_client_example ${HV_LIBRARIES}) + + add_executable(redis_subscriber_example redis_subscriber_test.cpp) + target_link_libraries(redis_subscriber_example ${HV_LIBRARIES}) + + list(APPEND EXAMPLES redis_client_example redis_subscriber_example) + endif() if(WITH_HTTP) include_directories(../http) diff --git a/examples/redis_client_test.cpp b/examples/redis_client_test.cpp new file mode 100644 index 000000000..c339adbdc --- /dev/null +++ b/examples/redis_client_test.cpp @@ -0,0 +1,56 @@ +#include +#include +#include +#include + +#include "redis/RedisClient.h" + +using namespace hv; + +int main(int argc, char** argv) { + std::string host = argc > 1 ? argv[1] : "127.0.0.1"; + int port = argc > 2 ? atoi(argv[2]) : 6379; + + RedisClient client; + client.setHost(host); + client.setPort(port); + client.setConnectTimeout(3000); + client.setTimeout(3000); + + RedisResult set_result = client.set("libhv:redis:example", "hello"); + if (!set_result.ok()) { + std::cerr << "SET failed, code=" << set_result.code << std::endl; + return set_result.code != 0 ? set_result.code : -1; + } + + RedisValueResult value = client.get("libhv:redis:example"); + if (!value.ok()) { + std::cerr << "GET failed, code=" << value.code << std::endl; + return value.code != 0 ? value.code : -1; + } + std::cout << "GET libhv:redis:example => " << value.value << std::endl; + + RedisPipeline pipeline = client.pipeline(); + pipeline.appendCommand(RedisCommand{"SET", "libhv:redis:pipeline", "1"}); + pipeline.appendCommand(RedisCommand{"GET", "libhv:redis:pipeline"}); + std::vector replies; + RedisResult pipe_result = pipeline.exec(&replies); + if (!pipe_result.ok()) { + std::cerr << "Pipeline failed, code=" << pipe_result.code << std::endl; + return pipe_result.code != 0 ? pipe_result.code : -1; + } + std::cout << "Pipeline replies=" << replies.size() << std::endl; + + RedisTransaction tx = client.transaction(); + tx.appendCommand(RedisCommand{"SET", "libhv:redis:tx", "7"}); + tx.appendCommand(RedisCommand{"GET", "libhv:redis:tx"}); + replies.clear(); + RedisResult tx_result = tx.exec(&replies); + if (!tx_result.ok()) { + std::cerr << "Transaction failed, code=" << tx_result.code << std::endl; + return tx_result.code != 0 ? tx_result.code : -1; + } + std::cout << "Transaction replies=" << replies.size() << std::endl; + + return 0; +} diff --git a/examples/redis_subscriber_test.cpp b/examples/redis_subscriber_test.cpp new file mode 100644 index 000000000..6b766436e --- /dev/null +++ b/examples/redis_subscriber_test.cpp @@ -0,0 +1,53 @@ +#include +#include +#include + +#include "redis/RedisClient.h" +#include "redis/RedisSubscriber.h" + +using namespace hv; + +int main(int argc, char** argv) { + std::string host = argc > 1 ? argv[1] : "127.0.0.1"; + int port = argc > 2 ? atoi(argv[2]) : 6379; + std::string channel = argc > 3 ? argv[3] : "libhv:redis:channel"; + + RedisSubscriber subscriber; + subscriber.setHost(host); + subscriber.setPort(port); + subscriber.onSubscribe = [&](const std::string& name) { + std::cout << "subscribed: " << name << std::endl; + }; + subscriber.onUnsubscribe = [&](const std::string& name) { + std::cout << "unsubscribed: " << name << std::endl; + }; + subscriber.onMessage = [&](const std::string& recv_channel, const std::string& message) { + std::cout << recv_channel << " => " << message << std::endl; + }; + subscriber.onError = [](int code) { + std::cerr << "subscriber error: " << code << std::endl; + }; + + subscriber.start(); + int ret = subscriber.subscribe(channel); + if (ret != 0) { + std::cerr << "subscribe failed: " << ret << std::endl; + return ret; + } + + if (argc > 4) { + RedisClient publisher; + publisher.setHost(host); + publisher.setPort(port); + RedisValueResult published = publisher.publish(channel, argv[4]); + if (!published.ok()) { + std::cerr << "publish failed, code=" << published.code << std::endl; + } + } + + std::cout << "Press Enter to exit..." << std::endl; + std::string line; + std::getline(std::cin, line); + subscriber.stop(true); + return 0; +} diff --git a/redis/AsyncRedisClient.cpp b/redis/AsyncRedisClient.cpp new file mode 100644 index 000000000..b4a0fcacd --- /dev/null +++ b/redis/AsyncRedisClient.cpp @@ -0,0 +1,581 @@ +#include "AsyncRedisClient.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "TcpClient.h" + +namespace hv { + +struct AsyncRedisClient::Impl { + struct PendingRequest { + size_t expected_replies; + RedisCallback callback; + RedisRepliesCallback batch_callback; + std::vector replies; + TimerID timer_id; + std::string payload; + bool sent; + + PendingRequest() + : expected_replies(1) + , timer_id(INVALID_TIMER_ID) + , sent(false) {} + }; + + struct EnqueueState { + enum Owner { + kPending, + kLoop, + kCancelled, + }; + + std::mutex mutex; + std::condition_variable cv; + std::atomic owner; + bool done; + int code; + + EnqueueState() + : owner(kPending) + , done(false) + , code(ERR_CONNECT) {} + }; + + struct CleanupState { + std::mutex mutex; + std::condition_variable cv; + bool done; + + CleanupState() + : done(false) {} + }; + + AsyncRedisClient* self; + TcpClientEventLoopTmpl tcp_client; + std::deque > pending; + RedisParser parser; + bool is_loop_owner; + std::string host; + int port; + int connect_timeout_ms; + int timeout_ms; + std::string password; + int db; + bool handshake_pending; + std::atomic started; + std::atomic accept_requests; + std::atomic destroyed; + std::atomic stop_in_progress; + size_t handshake_index; + std::vector handshake_commands; + + Impl(AsyncRedisClient* client, const EventLoopPtr& loop, bool loop_owner) + : self(client) + , tcp_client(loop) + , is_loop_owner(loop_owner) + , port(6379) + , connect_timeout_ms(5000) + , timeout_ms(5000) + , db(0) + , handshake_pending(false) + , started(false) + , accept_requests(true) + , destroyed(false) + , stop_in_progress(false) + , handshake_index(0) {} + + static bool tryCancelEnqueue(const std::shared_ptr& state) { + int expected = EnqueueState::kPending; + return state->owner.compare_exchange_strong(expected, EnqueueState::kCancelled); + } + + bool acceptsRequests() { + if (!started || destroyed || stop_in_progress || !accept_requests || self->loop() == NULL || self->loop()->loop() == NULL) { + return false; + } + if (!is_loop_owner && !self->loop()->isRunning()) { + return false; + } + return true; + } + + void initCallbacks() { + tcp_client.onConnection = [this](const SocketChannelPtr& channel) { + if (destroyed) { + return; + } + if (channel->isConnected()) { + if (timeout_ms > 0) { + channel->setReadTimeout(timeout_ms); + channel->setWriteTimeout(timeout_ms); + } + clearProtocolState(); + beginHandshake(); + return; + } + clearProtocolState(); + failPending(ERR_CONNECT); + if (self->onClose) { + self->onClose(); + } + }; + + tcp_client.onMessage = [this](const SocketChannelPtr&, Buffer* buf) { + if (destroyed) { + return; + } + parser.Feed((const char*)buf->data(), buf->size()); + if (parser.HasError()) { + handleClientError(ERR_INVALID_PROTOCOL); + return; + } + while (parser.HasReply()) { + handleReply(parser.NextReply()); + } + }; + } + + int applySettings() { + tcp_client.remote_host = host.empty() ? "127.0.0.1" : host; + tcp_client.remote_port = port; + tcp_client.connect_timeout = connect_timeout_ms; + memset(&tcp_client.remote_addr, 0, sizeof(tcp_client.remote_addr)); + int ret = sockaddr_set_ipport(&tcp_client.remote_addr, tcp_client.remote_host.c_str(), tcp_client.remote_port); + if (ret != 0) { + return NABS(ret); + } + return 0; + } + + int startConnectInLoop() { + if (!accept_requests || destroyed || self->loop() == NULL || self->loop()->loop() == NULL) { + return ERR_CONNECT; + } + if (!is_loop_owner && !self->loop()->isRunning()) { + return ERR_CONNECT; + } + int ret = applySettings(); + if (ret != 0) { + notifyError(ret); + return ret; + } + ret = tcp_client.startConnect(); + if (ret != 0) { + notifyError(ret); + } + return ret; + } + + void clearCallbacks() { + tcp_client.onConnection = NULL; + tcp_client.onMessage = NULL; + tcp_client.onWriteComplete = NULL; + if (tcp_client.channel) { + tcp_client.channel->onconnect = NULL; + tcp_client.channel->onread = NULL; + tcp_client.channel->onwrite = NULL; + tcp_client.channel->onclose = NULL; + } + } + + void cleanupInPlace() { + tcp_client.setReconnect(NULL); + failPending(ERR_CONNECT); + clearProtocolState(); + clearCallbacks(); + if (tcp_client.channel && !tcp_client.channel->isClosed()) { + tcp_client.channel->close(); + } + } + + void runCleanupOnLoopAndWait() { + if (self->loop()->isInLoopThread()) { + cleanupInPlace(); + return; + } + std::shared_ptr state = std::make_shared(); + self->loop()->queueInLoop([this, state]() { + cleanupInPlace(); + { + std::lock_guard lock(state->mutex); + state->done = true; + } + state->cv.notify_one(); + }); + std::unique_lock lock(state->mutex); + while (!state->done) { + if (state->cv.wait_for(lock, std::chrono::milliseconds(10), [state]() { return state->done; })) { + break; + } + if (!self->loop()->isRunning()) { + break; + } + } + lock.unlock(); + if (!state->done) { + cleanupInPlace(); + } + } + + void clearProtocolState() { + parser.Reset(); + handshake_pending = false; + handshake_index = 0; + handshake_commands.clear(); + } + + int enqueueRequest(const std::shared_ptr& request) { + if (!acceptsRequests()) { + return ERR_CONNECT; + } + if (self->loop()->isInLoopThread()) { + return enqueueRequestInLoop(request, NULL); + } + std::shared_ptr state = std::make_shared(); + self->loop()->queueInLoop([this, request, state]() { + int code = enqueueRequestInLoop(request, state); + { + std::lock_guard lock(state->mutex); + state->code = code; + state->done = true; + } + state->cv.notify_one(); + }); + std::unique_lock lock(state->mutex); + while (!state->done) { + if (state->cv.wait_for(lock, std::chrono::milliseconds(10), [state]() { return state->done; })) { + break; + } + if (!acceptsRequests()) { + if (tryCancelEnqueue(state)) { + break; + } + } + } + if (!state->done) { + return ERR_CONNECT; + } + return state->code; + } + + int enqueueRequestInLoop(const std::shared_ptr& request, const std::shared_ptr& state) { + if (state) { + int expected = EnqueueState::kPending; + if (!state->owner.compare_exchange_strong(expected, EnqueueState::kLoop)) { + return ERR_CONNECT; + } + } + if (!acceptsRequests()) { + return ERR_CONNECT; + } + pending.push_back(request); + armTimeout(request); + if (tcp_client.isConnected() && !handshake_pending) { + flushPending(); + } + else if (started && !handshake_pending) { + tcp_client.start(); + } + return 0; + } + + void beginHandshake() { + handshake_commands.clear(); + if (!password.empty()) { + handshake_commands.push_back(RedisCommand{"AUTH", password}); + } + if (db > 0) { + handshake_commands.push_back(RedisCommand{"SELECT", std::to_string(db)}); + } + handshake_pending = !handshake_commands.empty(); + if (!handshake_pending) { + finishHandshake(); + return; + } + sendHandshakeCommand(0); + } + + void sendHandshakeCommand(size_t index) { + if (index >= handshake_commands.size()) { + finishHandshake(); + return; + } + int ret = tcp_client.send(RedisEncodeCommand(handshake_commands[index])); + if (ret < 0) { + handleClientError(ret); + return; + } + handshake_index = index; + } + + void armTimeout(const std::shared_ptr& request) { + if (timeout_ms <= 0 || !self->loop()) { + return; + } + request->timer_id = self->loop()->setTimeout(timeout_ms, [this, request](TimerID timerID) { + if (request->timer_id != timerID) { + return; + } + handleClientError(ERR_TASK_TIMEOUT); + }); + } + + void flushPending() { + if (!tcp_client.isConnected()) { + if (started) { + tcp_client.start(); + } + return; + } + for (size_t i = 0; i < pending.size(); ++i) { + const std::shared_ptr& request = pending[i]; + if (request->sent) { + continue; + } + int ret = tcp_client.send(request->payload); + if (ret < 0) { + handleClientError(ret); + return; + } + request->sent = true; + } + } + + void failPending(int code) { + while (!pending.empty()) { + const std::shared_ptr& request = pending.front(); + cancelTimeout(request); + invokeRequestCallback(request, code); + pending.pop_front(); + } + } + + void cancelTimeout(const std::shared_ptr& request) { + if (request->timer_id != INVALID_TIMER_ID && self->loop()) { + self->loop()->killTimer(request->timer_id); + request->timer_id = INVALID_TIMER_ID; + } + } + + void invokeRequestCallback(const std::shared_ptr& request, int code) { + if (request->callback) { + RedisResult result(code); + if (code == 0 && !request->replies.empty()) { + result.reply = request->replies.front(); + } + request->callback(result); + } + if (request->batch_callback) { + std::vector replies; + if (code == 0) { + replies = request->replies; + } + request->batch_callback(code, replies); + } + } + + void handleHandshakeReply(const RedisReply& reply) { + if (reply.isError()) { + handleClientError(ERR_RESPONSE); + return; + } + size_t next = handshake_index + 1; + if (next >= handshake_commands.size()) { + finishHandshake(); + return; + } + sendHandshakeCommand(next); + } + + void handleReply(const RedisReply& reply) { + if (handshake_pending) { + handleHandshakeReply(reply); + return; + } + if (pending.empty()) { + handleClientError(ERR_RESPONSE); + return; + } + const std::shared_ptr& request = pending.front(); + request->replies.push_back(reply); + if (request->replies.size() < request->expected_replies) { + return; + } + cancelTimeout(request); + invokeRequestCallback(request, 0); + pending.pop_front(); + } + + void handleClientError(int code) { + notifyError(code); + failPending(code); + clearProtocolState(); + if (tcp_client.channel && !tcp_client.channel->isClosed()) { + tcp_client.channel->close(); + } + } + + void finishHandshake() { + clearProtocolState(); + if (self->onConnect) { + self->onConnect(); + } + flushPending(); + } + + void notifyError(int code) { + if (self->onError) { + self->onError(code); + } + } +}; + +AsyncRedisClient::AsyncRedisClient(EventLoopPtr loop) + : EventLoopThread(loop) + , impl_(std::make_shared(this, EventLoopThread::loop(), loop == NULL)) { + impl_->initCallbacks(); +} + +AsyncRedisClient::~AsyncRedisClient() { + stop(true); +} + +void AsyncRedisClient::setHost(const std::string& host) { + impl_->host = host; +} + +void AsyncRedisClient::setPort(int port) { + impl_->port = port; +} + +void AsyncRedisClient::setAuth(const std::string& password) { + impl_->password = password; +} + +void AsyncRedisClient::setDb(int db) { + impl_->db = db; +} + +void AsyncRedisClient::setConnectTimeout(int ms) { + impl_->connect_timeout_ms = ms; +} + +void AsyncRedisClient::setTimeout(int ms) { + impl_->timeout_ms = ms; +} + +void AsyncRedisClient::setReconnect(reconn_setting_t* setting) { + impl_->tcp_client.setReconnect(setting); +} + +void AsyncRedisClient::start(bool wait_threads_started) { + impl_->destroyed = false; + impl_->stop_in_progress = false; + impl_->started = true; + impl_->accept_requests = true; + if (!impl_->is_loop_owner) { + if (!loop() || !loop()->loop() || !loop()->isRunning()) { + impl_->started = false; + impl_->accept_requests = false; + impl_->notifyError(ERR_CONNECT); + return; + } + loop()->runInLoop([this]() { + impl_->startConnectInLoop(); + }); + return; + } + if (isRunning()) { + loop()->runInLoop([this]() { + impl_->startConnectInLoop(); + }); + return; + } + EventLoopThread::start(wait_threads_started, [this]() { + return impl_->startConnectInLoop(); + }); +} + +void AsyncRedisClient::stop(bool wait_threads_stopped) { + impl_->accept_requests = false; + impl_->started = false; + impl_->destroyed = true; + impl_->stop_in_progress = true; + if (!loop()) { + impl_->cleanupInPlace(); + impl_->stop_in_progress = false; + return; + } + if (!impl_->is_loop_owner) { + if (loop()->isRunning()) { + impl_->runCleanupOnLoopAndWait(); + } + else { + impl_->cleanupInPlace(); + } + impl_->stop_in_progress = false; + return; + } + if (loop()->isRunning()) { + impl_->runCleanupOnLoopAndWait(); + } + else { + impl_->cleanupInPlace(); + } + EventLoopThread::stop(wait_threads_stopped); + impl_->stop_in_progress = false; +} + +bool AsyncRedisClient::isConnected() const { + return impl_->tcp_client.channel && impl_->tcp_client.channel->isConnected(); +} + +bool AsyncRedisClient::isStarted() const { + return impl_->started; +} + +bool AsyncRedisClient::isInLoopThread() { + return loop() && loop()->isInLoopThread(); +} + +int AsyncRedisClient::command(const RedisCommand& command, RedisCallback cb) { + if (command.empty()) { + return ERR_INVALID_PARAM; + } + auto request = std::make_shared(); + request->payload = RedisEncodeCommand(command); + request->callback = std::move(cb); + int ret = impl_->enqueueRequest(request); + if (ret != 0) { + impl_->invokeRequestCallback(request, ret); + } + return ret; +} + +int AsyncRedisClient::commandBatch(const std::vector& commands, RedisRepliesCallback cb) { + if (commands.empty()) { + return ERR_INVALID_PARAM; + } + auto request = std::make_shared(); + request->expected_replies = commands.size(); + request->batch_callback = std::move(cb); + for (size_t i = 0; i < commands.size(); ++i) { + if (commands[i].empty()) { + return ERR_INVALID_PARAM; + } + request->payload += RedisEncodeCommand(commands[i]); + } + int ret = impl_->enqueueRequest(request); + if (ret != 0) { + impl_->invokeRequestCallback(request, ret); + } + return ret; +} + +} // namespace hv diff --git a/redis/AsyncRedisClient.h b/redis/AsyncRedisClient.h new file mode 100644 index 000000000..dfd793a56 --- /dev/null +++ b/redis/AsyncRedisClient.h @@ -0,0 +1,53 @@ +#ifndef HV_ASYNC_REDIS_CLIENT_HPP_ +#define HV_ASYNC_REDIS_CLIENT_HPP_ + +#include +#include +#include +#include + +#include "herr.h" + +#include "EventLoopThread.h" +#include "RedisMessage.h" + +namespace hv { + +using RedisCallback = std::function; +using RedisRepliesCallback = std::function&)>; + +class HV_EXPORT AsyncRedisClient : private EventLoopThread { +public: + AsyncRedisClient(EventLoopPtr loop = NULL); + ~AsyncRedisClient(); + + void setHost(const std::string& host); + void setPort(int port); + void setAuth(const std::string& password); + void setDb(int db); + void setConnectTimeout(int ms); + void setTimeout(int ms); + void setReconnect(reconn_setting_t* setting); + + void start(bool wait_threads_started = true); + void stop(bool wait_threads_stopped = true); + + bool isConnected() const; + bool isStarted() const; + bool isInLoopThread(); + + int command(const RedisCommand& command, RedisCallback cb); + int commandBatch(const std::vector& commands, RedisRepliesCallback cb); + + std::function onConnect; + std::function onClose; + std::function onError; + +private: + struct Impl; + std::shared_ptr impl_; +}; + +} // namespace hv + +#endif // HV_ASYNC_REDIS_CLIENT_HPP_ diff --git a/redis/RedisClient.cpp b/redis/RedisClient.cpp new file mode 100644 index 000000000..213f25595 --- /dev/null +++ b/redis/RedisClient.cpp @@ -0,0 +1,296 @@ +#include "RedisClient.h" + +#include +#include +#include +#include +#include +#include + +#include "RedisPipeline.h" +#include "RedisTransaction.h" + +namespace hv { + +namespace { + +template +RedisValueResult makeValueResult(const RedisResult& result); + +template<> +RedisValueResult makeValueResult(const RedisResult& result) { + RedisValueResult out; + out.code = result.code; + out.reply = result.reply; + if (result.code == 0 && result.reply.type == REDIS_REPLY_STRING && !result.reply.isError()) { + out.value = result.reply.asString(); + out.has_value = true; + } + return out; +} + +template<> +RedisValueResult makeValueResult(const RedisResult& result) { + RedisValueResult out; + out.code = result.code; + out.reply = result.reply; + if (result.code == 0 && result.reply.type == REDIS_REPLY_INTEGER && !result.reply.isError()) { + out.value = result.reply.asInt(); + out.has_value = true; + } + return out; +} + +template +RedisValueResult commandValue(RedisClient* client, const RedisCommand& command) { + return makeValueResult(client->command(command)); +} + +template +int commandAsyncValue(RedisClient* client, const RedisCommand& command, RedisValueCallback cb) { + return client->commandAsync(command, [cb](const RedisResult& result) { + if (cb) { + cb(makeValueResult(result)); + } + }); +} + +} // namespace + +RedisClient::RedisClient() {} + +void RedisClient::setHost(const std::string& host) { + async_.setHost(host); +} + +void RedisClient::setPort(int port) { + async_.setPort(port); +} + +void RedisClient::setAuth(const std::string& password) { + async_.setAuth(password); +} + +void RedisClient::setDb(int db) { + async_.setDb(db); +} + +void RedisClient::setConnectTimeout(int ms) { + async_.setConnectTimeout(ms); +} + +void RedisClient::setTimeout(int ms) { + async_.setTimeout(ms); +} + +void RedisClient::setReconnect(reconn_setting_t* setting) { + async_.setReconnect(setting); +} + +RedisResult RedisClient::command(const RedisCommand& command) { + std::mutex mutex; + std::condition_variable cv; + bool done = false; + RedisResult result; + + if (async_.isStarted() && async_.isInLoopThread()) { + return RedisResult(ERR_INVALID_HANDLE); + } + if (!async_.isStarted()) { + async_.start(); + } + int ret = async_.command(command, [&](const RedisResult& value) { + std::lock_guard lock(mutex); + result = value; + done = true; + cv.notify_one(); + }); + if (ret != 0) { + std::lock_guard lock(mutex); + if (!done) { + result.code = ret; + } + return result; + } + + std::unique_lock lock(mutex); + cv.wait(lock, [&done]() { return done; }); + return result; +} + +int RedisClient::commandAsync(const RedisCommand& command, RedisCallback cb) { + async_.start(); + return async_.command(command, std::move(cb)); +} + +RedisResult RedisClient::commandf(const char* fmt, ...) { + va_list ap; + va_start(ap, fmt); + va_list ap_copy; + va_copy(ap_copy, ap); + int len = vsnprintf(NULL, 0, fmt, ap_copy); + va_end(ap_copy); + if (len < 0) { + va_end(ap); + return RedisResult(ERR_INVALID_PARAM); + } + std::vector buf((size_t)len + 1); + if (vsnprintf(buf.data(), buf.size(), fmt, ap) < 0) { + va_end(ap); + return RedisResult(ERR_INVALID_PARAM); + } + va_end(ap); + + RedisCommand command = tokenize(buf.data()); + if (command.empty()) { + return RedisResult(ERR_INVALID_PARAM); + } + return this->command(command); +} + +RedisPipeline RedisClient::pipeline() { + return RedisPipeline(this); +} + +RedisTransaction RedisClient::transaction() { + return RedisTransaction(this); +} + +RedisValueResult RedisClient::get(const std::string& key) { + return commandValue(this, RedisCommand{"GET", key}); +} + +int RedisClient::getAsync(const std::string& key, RedisValueCallback cb) { + return commandAsyncValue(this, RedisCommand{"GET", key}, std::move(cb)); +} + +RedisResult RedisClient::set(const std::string& key, const std::string& value) { + return command(RedisCommand{"SET", key, value}); +} + +int RedisClient::setAsync(const std::string& key, const std::string& value, RedisCallback cb) { + return commandAsync(RedisCommand{"SET", key, value}, std::move(cb)); +} + +RedisValueResult RedisClient::del(const std::string& key) { + return commandValue(this, RedisCommand{"DEL", key}); +} + +int RedisClient::delAsync(const std::string& key, RedisValueCallback cb) { + return commandAsyncValue(this, RedisCommand{"DEL", key}, std::move(cb)); +} + +RedisValueResult RedisClient::exists(const std::string& key) { + return commandValue(this, RedisCommand{"EXISTS", key}); +} + +int RedisClient::existsAsync(const std::string& key, RedisValueCallback cb) { + return commandAsyncValue(this, RedisCommand{"EXISTS", key}, std::move(cb)); +} + +RedisValueResult RedisClient::expire(const std::string& key, int seconds) { + return commandValue(this, RedisCommand{"EXPIRE", key, std::to_string(seconds)}); +} + +int RedisClient::expireAsync(const std::string& key, int seconds, RedisValueCallback cb) { + return commandAsyncValue(this, RedisCommand{"EXPIRE", key, std::to_string(seconds)}, std::move(cb)); +} + +RedisValueResult RedisClient::hget(const std::string& key, const std::string& field) { + return commandValue(this, RedisCommand{"HGET", key, field}); +} + +int RedisClient::hgetAsync(const std::string& key, const std::string& field, RedisValueCallback cb) { + return commandAsyncValue(this, RedisCommand{"HGET", key, field}, std::move(cb)); +} + +RedisValueResult RedisClient::hset(const std::string& key, const std::string& field, const std::string& value) { + return commandValue(this, RedisCommand{"HSET", key, field, value}); +} + +int RedisClient::hsetAsync(const std::string& key, const std::string& field, const std::string& value, RedisValueCallback cb) { + return commandAsyncValue(this, RedisCommand{"HSET", key, field, value}, std::move(cb)); +} + +RedisValueResult RedisClient::publish(const std::string& channel, const std::string& message) { + return commandValue(this, RedisCommand{"PUBLISH", channel, message}); +} + +int RedisClient::publishAsync(const std::string& channel, const std::string& message, RedisValueCallback cb) { + return commandAsyncValue(this, RedisCommand{"PUBLISH", channel, message}, std::move(cb)); +} + +RedisResult RedisClient::commandBatch(const std::vector& commands, std::vector* replies) { + if (replies) { + replies->clear(); + } + std::mutex mutex; + std::condition_variable cv; + bool done = false; + RedisResult result; + std::vector batch_replies; + + if (async_.isStarted() && async_.isInLoopThread()) { + return RedisResult(ERR_INVALID_HANDLE); + } + if (!async_.isStarted()) { + async_.start(); + } + int ret = async_.commandBatch(commands, [&](int code, const std::vector& value) { + std::lock_guard lock(mutex); + result.code = code; + if (code == 0) { + batch_replies = value; + } + done = true; + cv.notify_one(); + }); + if (ret != 0) { + std::lock_guard lock(mutex); + if (!done) { + result.code = ret; + } + return result; + } + + std::unique_lock lock(mutex); + cv.wait(lock, [&done]() { return done; }); + lock.unlock(); + + if (result.code == 0) { + if (replies) { + *replies = batch_replies; + } + if (!batch_replies.empty()) { + result.reply = batch_replies.back(); + } + } + return result; +} + +int RedisClient::commandBatchAsync(const std::vector& commands, RedisRepliesCallback cb) { + async_.start(); + return async_.commandBatch(commands, std::move(cb)); +} + +RedisCommand RedisClient::tokenize(const std::string& line) { + RedisCommand command; + std::string token; + for (size_t i = 0; i < line.size(); ++i) { + unsigned char ch = (unsigned char)line[i]; + if (std::isspace(ch)) { + if (!token.empty()) { + command.push_back(token); + token.clear(); + } + continue; + } + token.push_back((char)ch); + } + if (!token.empty()) { + command.push_back(token); + } + return command; +} + +} // namespace hv diff --git a/redis/RedisClient.h b/redis/RedisClient.h new file mode 100644 index 000000000..43f3eb4e2 --- /dev/null +++ b/redis/RedisClient.h @@ -0,0 +1,101 @@ +#ifndef HV_REDIS_CLIENT_HPP_ +#define HV_REDIS_CLIENT_HPP_ + +#include +#include +#include +#include +#include + +#include "AsyncRedisClient.h" +#include "RedisPipeline.h" +#include "RedisTransaction.h" + +namespace hv { + +template +using RedisValueCallback = std::function&)>; + +class HV_EXPORT RedisClient { +public: + RedisClient(); + + void setHost(const std::string& host); + void setPort(int port); + void setAuth(const std::string& password); + void setDb(int db); + void setConnectTimeout(int ms); + void setTimeout(int ms); + void setReconnect(reconn_setting_t* setting); + + RedisResult command(const RedisCommand& command); + int commandAsync(const RedisCommand& command, RedisCallback cb); + RedisResult commandf(const char* fmt, ...); + + template + int commandfAsync(const char* fmt, RedisCallback cb, Args... args) { + RedisCommand command; + if (!formatCommandArgs(fmt, &command, args...)) { + return ERR_INVALID_PARAM; + } + return commandAsync(command, std::move(cb)); + } + + RedisPipeline pipeline(); + RedisTransaction transaction(); + + RedisValueResult get(const std::string& key); + int getAsync(const std::string& key, RedisValueCallback cb); + + RedisResult set(const std::string& key, const std::string& value); + int setAsync(const std::string& key, const std::string& value, RedisCallback cb); + + RedisValueResult del(const std::string& key); + int delAsync(const std::string& key, RedisValueCallback cb); + + RedisValueResult exists(const std::string& key); + int existsAsync(const std::string& key, RedisValueCallback cb); + + RedisValueResult expire(const std::string& key, int seconds); + int expireAsync(const std::string& key, int seconds, RedisValueCallback cb); + + RedisValueResult hget(const std::string& key, const std::string& field); + int hgetAsync(const std::string& key, const std::string& field, RedisValueCallback cb); + + RedisValueResult hset(const std::string& key, const std::string& field, const std::string& value); + int hsetAsync(const std::string& key, const std::string& field, const std::string& value, RedisValueCallback cb); + + RedisValueResult publish(const std::string& channel, const std::string& message); + int publishAsync(const std::string& channel, const std::string& message, RedisValueCallback cb); + +private: + template + static bool formatCommandArgs(const char* fmt, RedisCommand* command, Args... args) { + if (fmt == nullptr || command == nullptr) { + return false; + } + int len = std::snprintf(NULL, 0, fmt, args...); + if (len < 0) { + return false; + } + std::vector buf((size_t)len + 1); + if (std::snprintf(buf.data(), buf.size(), fmt, args...) < 0) { + return false; + } + *command = tokenize(buf.data()); + return !command->empty(); + } + + static RedisCommand tokenize(const std::string& line); + RedisResult commandBatch(const std::vector& commands, std::vector* replies); + int commandBatchAsync(const std::vector& commands, RedisRepliesCallback cb); + + friend class RedisPipeline; + friend class RedisTransaction; + + AsyncRedisClient async_; +}; + +} // namespace hv + +#endif // HV_REDIS_CLIENT_HPP_ diff --git a/redis/RedisMessage.cpp b/redis/RedisMessage.cpp new file mode 100644 index 000000000..2ad6d3a9a --- /dev/null +++ b/redis/RedisMessage.cpp @@ -0,0 +1,296 @@ +#include "RedisMessage.h" + +#include +#include +#include +#include + +namespace hv { + +namespace { + +void RedisAppendBulk(std::string& out, const std::string& value) { + out += "$"; + out += std::to_string(value.size()); + out += "\r\n"; + out += value; + out += "\r\n"; +} + +void RedisAppendReply(std::string& out, const RedisReply& reply) { + switch (reply.type) { + case REDIS_REPLY_STRING: + if (reply.bulk) { + RedisAppendBulk(out, reply.str); + } + else { + out += "+"; + out += reply.str; + out += "\r\n"; + } + return; + case REDIS_REPLY_ERROR: + out += "-"; + out += reply.str; + out += "\r\n"; + return; + case REDIS_REPLY_INTEGER: + out += ":"; + out += std::to_string(reply.integer); + out += "\r\n"; + return; + case REDIS_REPLY_ARRAY: + out += "*"; + out += std::to_string(reply.elements.size()); + out += "\r\n"; + for (size_t i = 0; i < reply.elements.size(); ++i) { + RedisAppendReply(out, reply.elements[i]); + } + return; + case REDIS_REPLY_NIL: + out += reply.null_array ? "*-1\r\n" : "$-1\r\n"; + return; + default: + out += "$-1\r\n"; + return; + } +} + +} // namespace + +struct RedisParser::Impl { + enum ParseStatus { + kParseOk, + kParseIncomplete, + kParseError, + }; + + std::string buffer; + std::deque replies; + bool error; + + Impl() + : error(false) {} + + static bool ParseInt64(const std::string& str, int64_t* value) { + if (value == NULL || str.empty()) return false; + char* end = NULL; + errno = 0; + long long parsed = strtoll(str.c_str(), &end, 10); + if (errno != 0 || end == str.c_str() || *end != '\0') { + return false; + } + *value = (int64_t)parsed; + return true; + } + + bool ParseLine(size_t start, std::string& line, size_t& next) const { + size_t pos = buffer.find("\r\n", start); + if (pos == std::string::npos) { + return false; + } + line.assign(buffer, start, pos - start); + next = pos + 2; + return true; + } + + ParseStatus ParseOne(size_t& off, RedisReply& reply) { + size_t start = off; + if (off >= buffer.size()) { + return kParseIncomplete; + } + + char prefix = buffer[off++]; + std::string line; + size_t next = off; + int64_t number = 0; + + switch (prefix) { + case '+': + if (!ParseLine(off, line, next)) { + off = start; + return kParseIncomplete; + } + reply = RedisReply(); + reply.type = REDIS_REPLY_STRING; + reply.str = line; + off = next; + return kParseOk; + case '-': + if (!ParseLine(off, line, next)) { + off = start; + return kParseIncomplete; + } + reply = RedisReply(); + reply.type = REDIS_REPLY_ERROR; + reply.str = line; + off = next; + return kParseOk; + case ':': + if (!ParseLine(off, line, next)) { + off = start; + return kParseIncomplete; + } + if (!ParseInt64(line, &number)) { + off = next; + return kParseError; + } + reply = RedisReply(); + reply.type = REDIS_REPLY_INTEGER; + reply.integer = number; + off = next; + return kParseOk; + case '$': { + if (!ParseLine(off, line, next)) { + off = start; + return kParseIncomplete; + } + if (!ParseInt64(line, &number)) { + off = next; + return kParseError; + } + if (number == -1) { + reply = RedisReply(); + off = next; + return kParseOk; + } + if (number < 0) { + off = next; + return kParseError; + } + size_t bulk_len = (size_t)number; + if (buffer.size() < next + bulk_len + 2) { + off = start; + return kParseIncomplete; + } + if (buffer[next + bulk_len] != '\r' || buffer[next + bulk_len + 1] != '\n') { + off = next + bulk_len + 2; + return kParseError; + } + reply = RedisReply(); + reply.type = REDIS_REPLY_STRING; + reply.bulk = true; + reply.str.assign(buffer.data() + next, bulk_len); + off = next + bulk_len + 2; + return kParseOk; + } + case '*': { + if (!ParseLine(off, line, next)) { + off = start; + return kParseIncomplete; + } + if (!ParseInt64(line, &number)) { + off = next; + return kParseError; + } + if (number == -1) { + reply = RedisReply(); + reply.null_array = true; + off = next; + return kParseOk; + } + if (number < 0) { + off = next; + return kParseError; + } + size_t cursor = next; + std::vector elements; + elements.reserve((size_t)number); + for (int64_t i = 0; i < number; ++i) { + RedisReply element; + ParseStatus status = ParseOne(cursor, element); + if (status == kParseIncomplete) { + off = start; + return status; + } + if (status == kParseError) { + off = cursor; + return status; + } + elements.push_back(std::move(element)); + } + reply = RedisReply(); + reply.type = REDIS_REPLY_ARRAY; + reply.elements.swap(elements); + off = cursor; + return kParseOk; + } + default: + off = start; + return kParseError; + } + } + + void ParseAll() { + size_t off = 0; + error = false; + while (off < buffer.size()) { + RedisReply reply; + size_t next = off; + ParseStatus status = ParseOne(next, reply); + if (status == kParseOk) { + replies.push_back(std::move(reply)); + off = next; + continue; + } + if (status == kParseIncomplete) { + break; + } + error = true; + off = next > off ? next : off + 1; + } + if (off != 0) { + buffer.erase(0, off); + } + } +}; + +std::string RedisEncodeCommand(const RedisCommand& command) { + std::string out = "*" + std::to_string(command.size()) + "\r\n"; + for (size_t i = 0; i < command.size(); ++i) { + RedisAppendBulk(out, command[i]); + } + return out; +} + +std::string RedisEncodeReply(const RedisReply& reply) { + std::string out; + RedisAppendReply(out, reply); + return out; +} + +RedisParser::RedisParser() + : impl_(std::make_shared()) {} + +void RedisParser::Reset() { + impl_->buffer.clear(); + impl_->replies.clear(); + impl_->error = false; +} + +size_t RedisParser::Feed(const char* data, size_t len) { + if (data != NULL && len != 0) { + impl_->buffer.append(data, len); + } + impl_->ParseAll(); + return len; +} + +bool RedisParser::HasReply() const { + return !impl_->replies.empty(); +} + +RedisReply RedisParser::NextReply() { + if (impl_->replies.empty()) { + return RedisReply(); + } + RedisReply reply = std::move(impl_->replies.front()); + impl_->replies.pop_front(); + return reply; +} + +bool RedisParser::HasError() const { + return impl_->error; +} + +} // namespace hv diff --git a/redis/RedisMessage.h b/redis/RedisMessage.h new file mode 100644 index 000000000..5da2cdcfd --- /dev/null +++ b/redis/RedisMessage.h @@ -0,0 +1,85 @@ +#ifndef HV_REDIS_MESSAGE_HPP_ +#define HV_REDIS_MESSAGE_HPP_ + +#include +#include + +#include +#include +#include + +namespace hv { + +using RedisCommand = std::vector; + +enum RedisReplyType { + REDIS_REPLY_NIL, + REDIS_REPLY_STRING, + REDIS_REPLY_ERROR, + REDIS_REPLY_INTEGER, + REDIS_REPLY_ARRAY, +}; + +struct RedisReply { + RedisReplyType type = REDIS_REPLY_NIL; + std::string str; + int64_t integer = 0; + std::vector elements; + bool bulk = false; + bool null_array = false; + + bool isNil() const { return type == REDIS_REPLY_NIL; } + bool isError() const { return type == REDIS_REPLY_ERROR; } + bool isArray() const { return type == REDIS_REPLY_ARRAY; } + bool isString() const { return type == REDIS_REPLY_STRING; } + const std::string& error() const { return str; } + const std::string& asString() const { return str; } + int64_t asInt() const { return integer; } + const std::vector& asArray() const { return elements; } +}; + +struct RedisResult { + int code; + RedisReply reply; + + RedisResult(int c = 0) + : code(c) {} + + bool ok() const { return code == 0 && !reply.isError(); } +}; + +template +struct RedisValueResult { + int code = 0; + RedisReply reply; + T value = T(); + bool has_value = false; + + bool ok() const { return code == 0 && !reply.isError() && has_value; } + bool isNil() const { return code == 0 && reply.isNil(); } +}; + +std::string RedisEncodeCommand(const RedisCommand& command); +std::string RedisEncodeReply(const RedisReply& reply); + +class RedisParser { +public: + RedisParser(); + RedisParser(const RedisParser&) = delete; + RedisParser& operator=(const RedisParser&) = delete; + RedisParser(RedisParser&&) = default; + RedisParser& operator=(RedisParser&&) = default; + void Reset(); + size_t Feed(const char* data, size_t len); + bool HasReply() const; + RedisReply NextReply(); + bool HasError() const; + +private: + struct Impl; + std::shared_ptr impl_; +}; + +} // namespace hv + +#endif // HV_REDIS_MESSAGE_HPP_ diff --git a/redis/RedisPipeline.cpp b/redis/RedisPipeline.cpp new file mode 100644 index 000000000..6d53c8b67 --- /dev/null +++ b/redis/RedisPipeline.cpp @@ -0,0 +1,70 @@ +#include "RedisPipeline.h" + +#include + +#include "RedisClient.h" + +namespace hv { + +namespace { + +const RedisReply* firstErrorReply(const std::vector& replies) { + for (size_t i = 0; i < replies.size(); ++i) { + if (replies[i].isError()) { + return &replies[i]; + } + } + return NULL; +} + +} // namespace + +RedisPipeline::RedisPipeline(RedisClient* client) + : client_(client) {} + +void RedisPipeline::appendCommand(const RedisCommand& command) { + commands_.push_back(command); +} + +RedisResult RedisPipeline::exec(std::vector* replies) { + if (client_ == NULL || commands_.empty()) { + if (replies) { + replies->clear(); + } + return RedisResult(ERR_INVALID_PARAM); + } + + std::vector batch_replies; + std::vector* output = replies ? replies : &batch_replies; + RedisResult result = client_->commandBatch(commands_, output); + if (result.code != 0) { + return result; + } + + commands_.clear(); + + const RedisReply* error_reply = firstErrorReply(*output); + if (error_reply != NULL) { + result.reply = *error_reply; + } + else if (!output->empty()) { + result.reply = output->back(); + } + return result; +} + +int RedisPipeline::execAsync(const RedisRepliesCallback& cb) { + if (client_ == NULL || commands_.empty()) { + if (cb) { + cb(ERR_INVALID_PARAM, std::vector()); + } + return ERR_INVALID_PARAM; + } + int ret = client_->commandBatchAsync(commands_, cb); + if (ret == 0) { + commands_.clear(); + } + return ret; +} + +} // namespace hv diff --git a/redis/RedisPipeline.h b/redis/RedisPipeline.h new file mode 100644 index 000000000..ac766b530 --- /dev/null +++ b/redis/RedisPipeline.h @@ -0,0 +1,27 @@ +#ifndef HV_REDIS_PIPELINE_HPP_ +#define HV_REDIS_PIPELINE_HPP_ + +#include + +#include "AsyncRedisClient.h" + +namespace hv { + +class RedisClient; + +class HV_EXPORT RedisPipeline { +public: + explicit RedisPipeline(RedisClient* client = NULL); + + void appendCommand(const RedisCommand& command); + RedisResult exec(std::vector* replies = NULL); + int execAsync(const RedisRepliesCallback& cb); + +private: + RedisClient* client_; + std::vector commands_; +}; + +} // namespace hv + +#endif // HV_REDIS_PIPELINE_HPP_ diff --git a/redis/RedisSubscriber.cpp b/redis/RedisSubscriber.cpp new file mode 100644 index 000000000..8eb284991 --- /dev/null +++ b/redis/RedisSubscriber.cpp @@ -0,0 +1,588 @@ +#include "RedisSubscriber.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "RedisMessage.h" +#include "TcpClient.h" + +namespace hv { + +struct RedisSubscriber::Impl { + struct EnqueueState { + enum Owner { + kPending, + kLoop, + kCancelled, + }; + + std::mutex mutex; + std::condition_variable cv; + std::atomic owner; + bool done; + int code; + + EnqueueState() + : owner(kPending) + , done(false) + , code(ERR_CONNECT) {} + }; + + struct CleanupState { + std::mutex mutex; + std::condition_variable cv; + bool done; + + CleanupState() + : done(false) {} + }; + + enum OperationType { + kSubscribe, + kPSubscribe, + kUnsubscribe, + kPUnsubscribe, + }; + + RedisSubscriber* self; + TcpClientEventLoopTmpl tcp_client; + RedisParser parser; + bool is_loop_owner; + std::string host; + int port; + std::string password; + int db; + bool handshake_pending; + std::atomic started; + std::atomic accept_requests; + std::atomic destroyed; + std::atomic stop_in_progress; + size_t handshake_index; + std::vector handshake_commands; + std::set channels; + std::set patterns; + + Impl(RedisSubscriber* subscriber, const EventLoopPtr& loop, bool loop_owner) + : self(subscriber) + , tcp_client(loop) + , is_loop_owner(loop_owner) + , port(6379) + , db(0) + , handshake_pending(false) + , started(false) + , accept_requests(true) + , destroyed(false) + , stop_in_progress(false) + , handshake_index(0) {} + + static bool tryCancelEnqueue(const std::shared_ptr& state) { + int expected = EnqueueState::kPending; + return state->owner.compare_exchange_strong(expected, EnqueueState::kCancelled); + } + + bool acceptsRequests() { + if (!started || destroyed || stop_in_progress || !accept_requests || self->loop() == NULL || self->loop()->loop() == NULL) { + return false; + } + if (!is_loop_owner && !self->loop()->isRunning()) { + return false; + } + return true; + } + + void initCallbacks() { + tcp_client.onConnection = [this](const SocketChannelPtr& channel) { + if (destroyed) { + return; + } + if (channel->isConnected()) { + clearProtocolState(); + beginHandshake(); + return; + } + clearProtocolState(); + if (!stop_in_progress) { + notifyError(ERR_CONNECT); + } + }; + + tcp_client.onMessage = [this](const SocketChannelPtr&, Buffer* buf) { + if (destroyed) { + return; + } + parser.Feed((const char*)buf->data(), buf->size()); + if (parser.HasError()) { + handleClientError(ERR_INVALID_PROTOCOL); + return; + } + while (parser.HasReply()) { + handleReply(parser.NextReply()); + } + }; + } + + int applySettings() { + tcp_client.remote_host = host.empty() ? "127.0.0.1" : host; + tcp_client.remote_port = port; + memset(&tcp_client.remote_addr, 0, sizeof(tcp_client.remote_addr)); + int ret = sockaddr_set_ipport(&tcp_client.remote_addr, tcp_client.remote_host.c_str(), tcp_client.remote_port); + if (ret != 0) { + return NABS(ret); + } + return 0; + } + + int startConnectInLoop() { + if (!accept_requests || destroyed || self->loop() == NULL || self->loop()->loop() == NULL) { + return ERR_CONNECT; + } + if (!is_loop_owner && !self->loop()->isRunning()) { + return ERR_CONNECT; + } + int ret = applySettings(); + if (ret != 0) { + notifyError(ret); + return ret; + } + ret = tcp_client.startConnect(); + if (ret != 0) { + notifyError(ret); + } + return ret; + } + + void clearCallbacks() { + tcp_client.onConnection = NULL; + tcp_client.onMessage = NULL; + tcp_client.onWriteComplete = NULL; + if (tcp_client.channel) { + tcp_client.channel->onconnect = NULL; + tcp_client.channel->onread = NULL; + tcp_client.channel->onwrite = NULL; + tcp_client.channel->onclose = NULL; + } + } + + void cleanupInPlace() { + tcp_client.setReconnect(NULL); + clearProtocolState(); + clearCallbacks(); + if (tcp_client.channel && !tcp_client.channel->isClosed()) { + tcp_client.channel->close(); + } + } + + void runCleanupOnLoopAndWait() { + if (self->loop()->isInLoopThread()) { + cleanupInPlace(); + return; + } + std::shared_ptr state = std::make_shared(); + self->loop()->queueInLoop([this, state]() { + cleanupInPlace(); + { + std::lock_guard lock(state->mutex); + state->done = true; + } + state->cv.notify_one(); + }); + std::unique_lock lock(state->mutex); + while (!state->done) { + if (state->cv.wait_for(lock, std::chrono::milliseconds(10), [state]() { return state->done; })) { + break; + } + if (!self->loop()->isRunning()) { + break; + } + } + lock.unlock(); + if (!state->done) { + cleanupInPlace(); + } + } + + void clearProtocolState() { + parser.Reset(); + handshake_pending = false; + handshake_index = 0; + handshake_commands.clear(); + } + + int queueOperation(OperationType type, const std::string& name) { + if (name.empty()) { + return ERR_INVALID_PARAM; + } + if (!acceptsRequests()) { + return ERR_CONNECT; + } + if (self->loop()->isInLoopThread()) { + return performOperation(type, name, NULL); + } + std::shared_ptr state = std::make_shared(); + self->loop()->queueInLoop([this, type, name, state]() { + int code = performOperation(type, name, state); + { + std::lock_guard lock(state->mutex); + state->code = code; + state->done = true; + } + state->cv.notify_one(); + }); + std::unique_lock lock(state->mutex); + while (!state->done) { + if (state->cv.wait_for(lock, std::chrono::milliseconds(10), [state]() { return state->done; })) { + break; + } + if (!acceptsRequests()) { + if (tryCancelEnqueue(state)) { + break; + } + } + } + if (!state->done) { + return ERR_CONNECT; + } + return state->code; + } + + int performOperation(OperationType type, const std::string& name, const std::shared_ptr& state) { + if (state) { + int expected = EnqueueState::kPending; + if (!state->owner.compare_exchange_strong(expected, EnqueueState::kLoop)) { + return ERR_CONNECT; + } + } + if (!acceptsRequests()) { + return ERR_CONNECT; + } + + RedisCommand command; + switch (type) { + case kSubscribe: + if (!channels.insert(name).second) { + return 0; + } + command = RedisCommand{"SUBSCRIBE", name}; + break; + case kPSubscribe: + if (!patterns.insert(name).second) { + return 0; + } + command = RedisCommand{"PSUBSCRIBE", name}; + break; + case kUnsubscribe: + if (channels.erase(name) == 0) { + return 0; + } + if (tcp_client.isConnected() && !handshake_pending) { + command = RedisCommand{"UNSUBSCRIBE", name}; + } + break; + case kPUnsubscribe: + if (patterns.erase(name) == 0) { + return 0; + } + if (tcp_client.isConnected() && !handshake_pending) { + command = RedisCommand{"PUNSUBSCRIBE", name}; + } + break; + } + + if (!command.empty() && tcp_client.isConnected() && !handshake_pending) { + return sendCommand(command); + } + if (!tcp_client.isConnected() && started && !handshake_pending) { + tcp_client.start(); + } + return 0; + } + + void beginHandshake() { + handshake_commands.clear(); + if (!password.empty()) { + handshake_commands.push_back(RedisCommand{"AUTH", password}); + } + if (db > 0) { + handshake_commands.push_back(RedisCommand{"SELECT", std::to_string(db)}); + } + handshake_pending = !handshake_commands.empty(); + if (!handshake_pending) { + finishHandshake(); + return; + } + sendHandshakeCommand(0); + } + + void sendHandshakeCommand(size_t index) { + if (index >= handshake_commands.size()) { + finishHandshake(); + return; + } + int ret = sendCommand(handshake_commands[index]); + if (ret < 0) { + handleClientError(ret); + return; + } + handshake_index = index; + } + + int sendCommand(const RedisCommand& command) { + int ret = tcp_client.send(RedisEncodeCommand(command)); + if (ret < 0) { + handleClientError(ret); + return ret; + } + return 0; + } + + int sendSubscribeSet(const std::set& values, const char* verb) { + for (std::set::const_iterator it = values.begin(); it != values.end(); ++it) { + int ret = sendCommand(RedisCommand{verb, *it}); + if (ret < 0) { + return ret; + } + } + return 0; + } + + int syncSubscriptions() { + int ret = sendSubscribeSet(channels, "SUBSCRIBE"); + if (ret < 0) { + return ret; + } + return sendSubscribeSet(patterns, "PSUBSCRIBE"); + } + + void handleHandshakeReply(const RedisReply& reply) { + if (reply.isError()) { + handleClientError(ERR_RESPONSE); + return; + } + size_t next = handshake_index + 1; + if (next >= handshake_commands.size()) { + finishHandshake(); + return; + } + sendHandshakeCommand(next); + } + + static const RedisReply* arrayElement(const RedisReply& reply, size_t index) { + if (!reply.isArray() || reply.elements.size() <= index) { + return NULL; + } + return &reply.elements[index]; + } + + static bool arrayString(const RedisReply& reply, size_t index, std::string* out) { + const RedisReply* element = arrayElement(reply, index); + if (element == NULL || !element->isString()) { + return false; + } + if (out) { + *out = element->asString(); + } + return true; + } + + void handleReply(const RedisReply& reply) { + if (handshake_pending) { + handleHandshakeReply(reply); + return; + } + if (!reply.isArray()) { + handleClientError(ERR_RESPONSE); + return; + } + + std::string kind; + if (!arrayString(reply, 0, &kind)) { + handleClientError(ERR_RESPONSE); + return; + } + + if (kind == "message") { + std::string channel; + std::string message; + if (!arrayString(reply, 1, &channel) || !arrayString(reply, 2, &message)) { + handleClientError(ERR_RESPONSE); + return; + } + if (self->onMessage) { + self->onMessage(channel, message); + } + return; + } + + if (kind == "pmessage") { + std::string channel; + std::string message; + if (!arrayString(reply, 2, &channel) || !arrayString(reply, 3, &message)) { + handleClientError(ERR_RESPONSE); + return; + } + if (self->onMessage) { + self->onMessage(channel, message); + } + return; + } + + if (kind == "subscribe" || kind == "psubscribe") { + std::string name; + if (!arrayString(reply, 1, &name)) { + handleClientError(ERR_RESPONSE); + return; + } + if (self->onSubscribe) { + self->onSubscribe(name); + } + return; + } + + if (kind == "unsubscribe" || kind == "punsubscribe") { + std::string name; + if (!arrayString(reply, 1, &name)) { + handleClientError(ERR_RESPONSE); + return; + } + if (self->onUnsubscribe) { + self->onUnsubscribe(name); + } + return; + } + + handleClientError(ERR_RESPONSE); + } + + void handleClientError(int code) { + notifyError(code); + clearProtocolState(); + if (tcp_client.channel && !tcp_client.channel->isClosed()) { + tcp_client.channel->close(); + } + } + + void finishHandshake() { + clearProtocolState(); + if (syncSubscriptions() < 0) { + return; + } + } + + void notifyError(int code) { + if (self->onError) { + self->onError(code); + } + } +}; + +RedisSubscriber::RedisSubscriber(EventLoopPtr loop) + : EventLoopThread(loop) + , impl_(std::make_shared(this, EventLoopThread::loop(), loop == NULL)) { + impl_->initCallbacks(); +} + +RedisSubscriber::~RedisSubscriber() { + stop(true); +} + +void RedisSubscriber::setHost(const std::string& host) { + impl_->host = host; +} + +void RedisSubscriber::setPort(int port) { + impl_->port = port; +} + +void RedisSubscriber::setAuth(const std::string& password) { + impl_->password = password; +} + +void RedisSubscriber::setDb(int db) { + impl_->db = db; +} + +void RedisSubscriber::setReconnect(reconn_setting_t* setting) { + impl_->tcp_client.setReconnect(setting); +} + +void RedisSubscriber::start(bool wait_threads_started) { + impl_->destroyed = false; + impl_->stop_in_progress = false; + impl_->started = true; + impl_->accept_requests = true; + if (!impl_->is_loop_owner) { + if (!loop() || !loop()->loop() || !loop()->isRunning()) { + impl_->started = false; + impl_->accept_requests = false; + impl_->notifyError(ERR_CONNECT); + return; + } + loop()->runInLoop([this]() { + impl_->startConnectInLoop(); + }); + return; + } + if (isRunning()) { + loop()->runInLoop([this]() { + impl_->startConnectInLoop(); + }); + return; + } + EventLoopThread::start(wait_threads_started, [this]() { + return impl_->startConnectInLoop(); + }); +} + +void RedisSubscriber::stop(bool wait_threads_stopped) { + impl_->accept_requests = false; + impl_->started = false; + impl_->destroyed = true; + impl_->stop_in_progress = true; + if (!loop()) { + impl_->cleanupInPlace(); + impl_->stop_in_progress = false; + return; + } + if (!impl_->is_loop_owner) { + if (loop()->isRunning()) { + impl_->runCleanupOnLoopAndWait(); + } + else { + impl_->cleanupInPlace(); + } + impl_->stop_in_progress = false; + return; + } + if (loop()->isRunning()) { + impl_->runCleanupOnLoopAndWait(); + } + else { + impl_->cleanupInPlace(); + } + EventLoopThread::stop(wait_threads_stopped); + impl_->stop_in_progress = false; +} + +int RedisSubscriber::subscribe(const std::string& channel) { + return impl_->queueOperation(Impl::kSubscribe, channel); +} + +int RedisSubscriber::psubscribe(const std::string& pattern) { + return impl_->queueOperation(Impl::kPSubscribe, pattern); +} + +int RedisSubscriber::unsubscribe(const std::string& channel) { + return impl_->queueOperation(Impl::kUnsubscribe, channel); +} + +int RedisSubscriber::punsubscribe(const std::string& pattern) { + return impl_->queueOperation(Impl::kPUnsubscribe, pattern); +} + +} // namespace hv diff --git a/redis/RedisSubscriber.h b/redis/RedisSubscriber.h new file mode 100644 index 000000000..d9b7fb79a --- /dev/null +++ b/redis/RedisSubscriber.h @@ -0,0 +1,45 @@ +#ifndef HV_REDIS_SUBSCRIBER_HPP_ +#define HV_REDIS_SUBSCRIBER_HPP_ + +#include +#include +#include + +#include "herr.h" + +#include "EventLoopThread.h" + +namespace hv { + +class HV_EXPORT RedisSubscriber : private EventLoopThread { +public: + RedisSubscriber(EventLoopPtr loop = NULL); + ~RedisSubscriber(); + + void setHost(const std::string& host); + void setPort(int port); + void setAuth(const std::string& password); + void setDb(int db); + void setReconnect(reconn_setting_t* setting); + + void start(bool wait_threads_started = true); + void stop(bool wait_threads_stopped = true); + + int subscribe(const std::string& channel); + int psubscribe(const std::string& pattern); + int unsubscribe(const std::string& channel); + int punsubscribe(const std::string& pattern); + + std::function onMessage; + std::function onSubscribe; + std::function onUnsubscribe; + std::function onError; + +private: + struct Impl; + std::shared_ptr impl_; +}; + +} // namespace hv + +#endif // HV_REDIS_SUBSCRIBER_HPP_ diff --git a/redis/RedisTransaction.cpp b/redis/RedisTransaction.cpp new file mode 100644 index 000000000..5ddb20f47 --- /dev/null +++ b/redis/RedisTransaction.cpp @@ -0,0 +1,80 @@ +#include "RedisTransaction.h" + +#include "RedisClient.h" + +namespace hv { + +RedisTransaction::RedisTransaction(RedisClient* client) + : client_(client) {} + +void RedisTransaction::appendCommand(const RedisCommand& command) { + commands_.push_back(command); +} + +RedisResult RedisTransaction::exec(std::vector* replies) { + if (replies) { + replies->clear(); + } + if (client_ == NULL || commands_.empty()) { + return RedisResult(ERR_INVALID_PARAM); + } + + std::vector batch; + batch.reserve(commands_.size() + 2); + batch.push_back(RedisCommand{"MULTI"}); + batch.insert(batch.end(), commands_.begin(), commands_.end()); + batch.push_back(RedisCommand{"EXEC"}); + + std::vector batch_replies; + RedisResult result = client_->commandBatch(batch, &batch_replies); + if (result.code != 0) { + return result; + } + commands_.clear(); + if (batch_replies.size() != batch.size()) { + return RedisResult(ERR_RESPONSE); + } + + const RedisReply& multi_reply = batch_replies.front(); + if (multi_reply.isError()) { + result.reply = multi_reply; + return result; + } + + for (size_t i = 1; i + 1 < batch_replies.size(); ++i) { + if (batch_replies[i].isError()) { + result.reply = batch_replies[i]; + return result; + } + } + + const RedisReply& exec_reply = batch_replies.back(); + result.reply = exec_reply; + if (exec_reply.isNil() || !exec_reply.isArray()) { + result.code = ERR_RESPONSE; + return result; + } + if (replies != NULL) { + *replies = exec_reply.elements; + } + for (size_t i = 0; i < exec_reply.elements.size(); ++i) { + if (exec_reply.elements[i].isError()) { + result.reply = exec_reply.elements[i]; + return result; + } + } + return result; +} + +RedisResult RedisTransaction::discard() { + if (client_ == NULL || commands_.empty()) { + return RedisResult(ERR_INVALID_PARAM); + } + commands_.clear(); + RedisResult result; + result.reply.type = REDIS_REPLY_STRING; + result.reply.str = "OK"; + return result; +} + +} // namespace hv diff --git a/redis/RedisTransaction.h b/redis/RedisTransaction.h new file mode 100644 index 000000000..20434e04d --- /dev/null +++ b/redis/RedisTransaction.h @@ -0,0 +1,28 @@ +#ifndef HV_REDIS_TRANSACTION_HPP_ +#define HV_REDIS_TRANSACTION_HPP_ + +#include + +#include "hexport.h" +#include "RedisMessage.h" + +namespace hv { + +class RedisClient; + +class HV_EXPORT RedisTransaction { +public: + explicit RedisTransaction(RedisClient* client = NULL); + + void appendCommand(const RedisCommand& command); + RedisResult exec(std::vector* replies = NULL); + RedisResult discard(); + +private: + RedisClient* client_; + std::vector commands_; +}; + +} // namespace hv + +#endif // HV_REDIS_TRANSACTION_HPP_ diff --git a/scripts/unittest.sh b/scripts/unittest.sh index 940243443..dec6d3c87 100755 --- a/scripts/unittest.sh +++ b/scripts/unittest.sh @@ -2,6 +2,8 @@ SCRIPT_DIR=$(cd `dirname $0`; pwd) ROOT_DIR=${SCRIPT_DIR}/.. +export DYLD_LIBRARY_PATH=${ROOT_DIR}/lib:${DYLD_LIBRARY_PATH} +export LD_LIBRARY_PATH=${ROOT_DIR}/lib:${LD_LIBRARY_PATH} cd ${ROOT_DIR} bin/rbtree_test @@ -30,3 +32,11 @@ bin/socketpair_test # bin/objectpool_test bin/sizeof_test bin/http_router_test +for redis_test in redis_async_client_test redis_client_test redis_batch_test redis_subscriber_test; do + if [ -x bin/${redis_test} ]; then + bin/${redis_test} + fi +done +if [ -x bin/redis_protocol_test ]; then + bin/redis_protocol_test +fi diff --git a/unittest/CMakeLists.txt b/unittest/CMakeLists.txt index a2c5ff255..db466c0fb 100644 --- a/unittest/CMakeLists.txt +++ b/unittest/CMakeLists.txt @@ -90,6 +90,24 @@ target_include_directories(sendmail PRIVATE .. ../base ../protocol ../util) add_executable(http_router_test http_router_test.cpp) target_include_directories(http_router_test PRIVATE ../http/server) +# ------redis------ +if(WITH_EVPP AND WITH_REDIS) +add_executable(redis_protocol_test redis_protocol_test.cpp ../redis/RedisMessage.cpp) +target_include_directories(redis_protocol_test PRIVATE .. ../base ../event ../cpputil ../redis) + +set(REDIS_UNITTEST_TARGETS redis_protocol_test) +set(REDIS_TEST_INCLUDE_DIRS .. ../base ../ssl ../event ../cpputil ../redis ../evpp) +set(REDIS_LINKED_UNITTEST_TARGETS redis_async_client_test redis_client_test redis_batch_test redis_subscriber_test) + +foreach(redis_target ${REDIS_LINKED_UNITTEST_TARGETS}) + add_executable(${redis_target} ${redis_target}.cpp redis_test_server.cpp) + target_include_directories(${redis_target} PRIVATE ${REDIS_TEST_INCLUDE_DIRS}) + target_link_libraries(${redis_target} ${HV_LIBRARIES}) +endforeach() + +list(APPEND REDIS_UNITTEST_TARGETS ${REDIS_LINKED_UNITTEST_TARGETS}) +endif() + if(UNIX) add_executable(webbench webbench.c) endif() @@ -120,4 +138,5 @@ add_custom_target(unittest DEPENDS ftp sendmail http_router_test + ${REDIS_UNITTEST_TARGETS} ) diff --git a/unittest/redis_async_client_test.cpp b/unittest/redis_async_client_test.cpp new file mode 100644 index 000000000..ef8999dce --- /dev/null +++ b/unittest/redis_async_client_test.cpp @@ -0,0 +1,388 @@ +#include "redis/AsyncRedisClient.h" +#include "EventLoop.h" +#include "redis_test_server.h" + +#include +#include +#include +#include +#include +#include +#include + +using namespace hv; + +static void test_async_fifo_and_error_reply() { + FakeRedisServer server; + server.setCommandHandler([](const RedisCommand& cmd) { + RedisReply reply; + if (cmd[0] == "PING") { + reply.type = REDIS_REPLY_STRING; + reply.str = "PONG"; + } + else { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR unsupported"; + } + return reply; + }); + server.start(); + + AsyncRedisClient client; + client.setHost("127.0.0.1"); + client.setPort(server.port()); + client.start(); + + std::mutex mutex; + std::condition_variable cv; + std::vector results; + + auto collect = [&](const RedisResult& result) { + std::lock_guard lock(mutex); + results.push_back(result); + cv.notify_one(); + }; + + client.command(RedisCommand{"PING"}, collect); + client.command(RedisCommand{"NOPE"}, collect); + + std::unique_lock lock(mutex); + bool done = cv.wait_for(lock, std::chrono::seconds(3), [&results]() { return results.size() == 2; }); + assert(done); + assert(results[0].code == 0); + assert(results[0].reply.asString() == "PONG"); + assert(results[1].code == 0); + assert(results[1].reply.isError()); + assert(results[1].reply.error() == "ERR unsupported"); + + client.stop(true); + server.stop(); +} + + +static void test_async_batch_replies() { + FakeRedisServer server; + server.setCommandHandler([](const RedisCommand& cmd) { + RedisReply reply; + if (cmd[0] == "PING") { + reply.type = REDIS_REPLY_STRING; + reply.str = "PONG"; + } + else if (cmd[0] == "INCR") { + reply.type = REDIS_REPLY_INTEGER; + reply.integer = 1; + } + else { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR unsupported"; + } + return reply; + }); + server.start(); + + AsyncRedisClient client; + client.setHost("127.0.0.1"); + client.setPort(server.port()); + client.start(); + + std::mutex mutex; + std::condition_variable cv; + bool done = false; + int code = ERR_UNKNOWN; + std::vector replies; + + client.commandBatch({RedisCommand{"PING"}, RedisCommand{"INCR", "counter"}}, + [&](int batch_code, const std::vector& batch_replies) { + std::lock_guard lock(mutex); + code = batch_code; + replies = batch_replies; + done = true; + cv.notify_one(); + }); + + std::unique_lock lock(mutex); + bool completed = cv.wait_for(lock, std::chrono::seconds(3), [&done]() { return done; }); + assert(completed); + assert(code == 0); + assert(replies.size() == 2); + assert(replies[0].asString() == "PONG"); + assert(replies[1].type == REDIS_REPLY_INTEGER); + assert(replies[1].asInt() == 1); + + client.stop(true); + server.stop(); +} + +static void test_command_without_start_is_rejected() { + AsyncRedisClient client; + + bool called = false; + int ret = client.command(RedisCommand{"PING"}, [&](const RedisResult& result) { + called = true; + assert(result.code == ERR_CONNECT); + }); + + assert(ret == ERR_CONNECT); + assert(called); + + bool batch_called = false; + ret = client.commandBatch({RedisCommand{"PING"}}, [&](int code, const std::vector& replies) { + batch_called = true; + assert(code == ERR_CONNECT); + assert(replies.empty()); + }); + + assert(ret == ERR_CONNECT); + assert(batch_called); +} + +static void test_command_after_stop_is_rejected() { + AsyncRedisClient client; + client.stop(true); + + bool called = false; + int ret = client.command(RedisCommand{"PING"}, [&](const RedisResult& result) { + called = true; + assert(result.code == ERR_CONNECT); + }); + + assert(ret == ERR_CONNECT); + assert(called); +} + +static void test_external_loop_mode_can_start_and_command() { + FakeRedisServer server; + server.setCommandHandler([](const RedisCommand& cmd) { + RedisReply reply; + if (cmd[0] == "PING") { + reply.type = REDIS_REPLY_STRING; + reply.str = "PONG"; + } + else { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR unsupported"; + } + return reply; + }); + server.start(); + + EventLoopPtr loop = std::make_shared(); + AsyncRedisClient client(loop); + client.setHost("127.0.0.1"); + client.setPort(server.port()); + + std::mutex mutex; + std::condition_variable cv; + bool connected = false; + bool done = false; + RedisResult result; + + client.onConnect = [&]() { + std::lock_guard lock(mutex); + connected = true; + cv.notify_one(); + }; + + loop->queueInLoop([&client]() { + client.start(false); + }); + std::thread loop_thread([&loop]() { + loop->run(); + }); + + { + std::unique_lock lock(mutex); + bool started = cv.wait_for(lock, std::chrono::seconds(3), [&connected]() { return connected; }); + assert(started); + } + + int ret = client.command(RedisCommand{"PING"}, [&](const RedisResult& redis_result) { + std::lock_guard lock(mutex); + result = redis_result; + done = true; + cv.notify_one(); + }); + assert(ret == 0); + + { + std::unique_lock lock(mutex); + bool completed = cv.wait_for(lock, std::chrono::seconds(3), [&done]() { return done; }); + assert(completed); + } + assert(result.code == 0); + assert(result.reply.asString() == "PONG"); + + client.stop(true); + loop->stop(); + loop_thread.join(); + server.stop(); +} + +static void test_external_loop_stopped_rejects_command() { + EventLoopPtr loop = std::make_shared(); + AsyncRedisClient client(loop); + + std::thread loop_thread([&loop]() { + loop->run(); + }); + while (!loop->isRunning()) { + hv_delay(1); + } + loop->stop(); + loop_thread.join(); + + bool called = false; + int ret = client.command(RedisCommand{"PING"}, [&](const RedisResult& result) { + called = true; + assert(result.code == ERR_CONNECT); + }); + + assert(ret == ERR_CONNECT); + assert(called); +} + +static void test_external_loop_stopped_request_completes_once() { + EventLoopPtr loop = std::make_shared(); + AsyncRedisClient client(loop); + + std::thread loop_thread([&loop]() { + loop->run(); + }); + while (!loop->isRunning()) { + hv_delay(1); + } + + std::atomic callback_count(0); + std::atomic stop_now(false); + std::thread stop_thread([&]() { + while (!stop_now.load()) { + hv_delay(1); + } + loop->stop(); + }); + + stop_now = true; + int ret = client.command(RedisCommand{"PING"}, [&](const RedisResult& result) { + ++callback_count; + assert(result.code == ERR_CONNECT || result.code == 0); + }); + + stop_thread.join(); + loop_thread.join(); + hv_delay(20); + assert(callback_count.load() <= 1); + assert(ret == ERR_CONNECT || ret == 0); +} + +static void test_external_loop_not_running_rejects_start_and_command() { + EventLoopPtr loop = std::make_shared(); + AsyncRedisClient client(loop); + + int error_code = 0; + client.onError = [&](int code) { + error_code = code; + }; + client.start(false); + assert(error_code == ERR_CONNECT); + + int callback_count = 0; + int ret = client.command(RedisCommand{"PING"}, [&](const RedisResult& result) { + ++callback_count; + assert(result.code == ERR_CONNECT); + }); + + assert(ret == ERR_CONNECT); + assert(callback_count == 1); +} + +static void test_external_loop_running_without_start_rejects_command() { + EventLoopPtr loop = std::make_shared(); + AsyncRedisClient client(loop); + + std::thread loop_thread([&loop]() { + loop->run(); + }); + while (!loop->isRunning()) { + hv_delay(1); + } + + int callback_count = 0; + int ret = client.command(RedisCommand{"PING"}, [&](const RedisResult& result) { + ++callback_count; + assert(result.code == ERR_CONNECT); + }); + assert(ret == ERR_CONNECT); + assert(callback_count == 1); + + int batch_callback_count = 0; + ret = client.commandBatch({RedisCommand{"PING"}}, [&](int code, const std::vector& replies) { + ++batch_callback_count; + assert(code == ERR_CONNECT); + assert(replies.empty()); + }); + assert(ret == ERR_CONNECT); + assert(batch_callback_count == 1); + + loop->stop(); + loop_thread.join(); +} + +static void test_close_after_reply_fails_pending_once() { + FakeRedisServer server; + server.setCommandHandler([](const RedisCommand& cmd) { + RedisReply reply; + if (cmd[0] == "PING") { + reply.type = REDIS_REPLY_STRING; + reply.str = "PONG"; + } + else { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR unsupported"; + } + return reply; + }); + server.closeClientAfterReply(true); + server.start(); + + AsyncRedisClient client; + client.setHost("127.0.0.1"); + client.setPort(server.port()); + client.start(); + + std::mutex mutex; + std::condition_variable cv; + std::vector results; + + auto collect = [&](const RedisResult& result) { + std::lock_guard lock(mutex); + results.push_back(result); + cv.notify_one(); + }; + + assert(client.command(RedisCommand{"PING"}, collect) == 0); + assert(client.command(RedisCommand{"PING"}, collect) == 0); + + std::unique_lock lock(mutex); + bool completed = cv.wait_for(lock, std::chrono::seconds(3), [&results]() { return results.size() == 2; }); + assert(completed); + assert(results[0].code == 0); + assert(results[0].reply.asString() == "PONG"); + assert(results[1].code == ERR_CONNECT); + + client.stop(true); + server.stop(); +} + +int main() { + test_async_fifo_and_error_reply(); + test_async_batch_replies(); + test_command_without_start_is_rejected(); + test_command_after_stop_is_rejected(); + test_external_loop_mode_can_start_and_command(); + test_external_loop_stopped_rejects_command(); + test_external_loop_stopped_request_completes_once(); + test_external_loop_not_running_rejects_start_and_command(); + test_external_loop_running_without_start_rejects_command(); + test_close_after_reply_fails_pending_once(); + return 0; +} diff --git a/unittest/redis_batch_test.cpp b/unittest/redis_batch_test.cpp new file mode 100644 index 000000000..55ccace08 --- /dev/null +++ b/unittest/redis_batch_test.cpp @@ -0,0 +1,356 @@ +#include "redis/RedisClient.h" +#include "redis_test_server.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace hv; + +struct RedisBatchFixture { + std::map kv; + std::vector commands; + std::vector transaction_queue; + bool in_multi = false; + bool abort_exec = false; + bool queue_error = false; + bool exec_error_reply = false; + FakeRedisServer server; + + RedisBatchFixture() { + server.setCommandHandler([this](const RedisCommand& cmd) { + commands.push_back(cmd); + RedisReply reply; + if (cmd.empty()) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR empty command"; + return reply; + } + + if (cmd[0] == "AUTH" || cmd[0] == "SELECT") { + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + return reply; + } + + if (cmd[0] == "MULTI") { + in_multi = true; + transaction_queue.clear(); + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + return reply; + } + + if (cmd[0] == "DISCARD") { + in_multi = false; + transaction_queue.clear(); + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + return reply; + } + + if (cmd[0] == "EXEC") { + if (abort_exec) { + in_multi = false; + transaction_queue.clear(); + reply.type = REDIS_REPLY_NIL; + return reply; + } + reply.type = REDIS_REPLY_ARRAY; + if (exec_error_reply) { + RedisReply error; + error.type = REDIS_REPLY_ERROR; + error.str = "ERR exec failed"; + reply.elements.push_back(error); + } + else { + for (size_t i = 0; i < transaction_queue.size(); ++i) { + reply.elements.push_back(execute(transaction_queue[i])); + } + } + in_multi = false; + transaction_queue.clear(); + return reply; + } + + if (in_multi) { + transaction_queue.push_back(cmd); + if (queue_error && cmd[0] == "NOPE") { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR queue rejected"; + return reply; + } + reply.type = REDIS_REPLY_STRING; + reply.str = "QUEUED"; + return reply; + } + + return execute(cmd); + }); + server.start(); + } + + ~RedisBatchFixture() { + server.stop(); + } + + RedisReply execute(const RedisCommand& cmd) { + RedisReply reply; + if (cmd[0] == "SET") { + kv[cmd[1]] = atoll(cmd[2].c_str()); + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + } + else if (cmd[0] == "GET") { + std::map::const_iterator it = kv.find(cmd[1]); + if (it == kv.end()) { + reply.type = REDIS_REPLY_NIL; + } + else { + reply.type = REDIS_REPLY_STRING; + reply.str = std::to_string(it->second); + reply.bulk = true; + } + } + else if (cmd[0] == "INCR") { + reply.type = REDIS_REPLY_INTEGER; + reply.integer = ++kv[cmd[1]]; + } + else { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR unsupported"; + } + return reply; + } +}; + +static void setupClient(RedisClient* client, int port) { + client->setHost("127.0.0.1"); + client->setPort(port); + client->setAuth("secret"); + client->setDb(3); + client->setTimeout(3000); + client->setConnectTimeout(3000); +} + +static void test_pipeline_sync_exec() { + RedisBatchFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + + std::vector replies; + RedisPipeline pipeline = client.pipeline(); + pipeline.appendCommand(RedisCommand{"SET", "counter", "1"}); + pipeline.appendCommand(RedisCommand{"INCR", "counter"}); + + RedisResult result = pipeline.exec(&replies); + assert(result.code == 0); + assert(result.ok()); + assert(replies.size() == 2); + assert(replies[0].asString() == "OK"); + assert(replies[1].type == REDIS_REPLY_INTEGER); + assert(replies[1].asInt() == 2); + assert(fixture.kv["counter"] == 2); + + std::vector second_replies; + RedisResult second = pipeline.exec(&second_replies); + assert(second.code == ERR_INVALID_PARAM); + assert(second_replies.empty()); +} + +static void test_pipeline_async_exec() { + RedisBatchFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + + RedisPipeline pipeline = client.pipeline(); + pipeline.appendCommand(RedisCommand{"SET", "async-counter", "4"}); + pipeline.appendCommand(RedisCommand{"INCR", "async-counter"}); + + std::mutex mutex; + std::condition_variable cv; + bool done = false; + int code = ERR_UNKNOWN; + std::vector replies; + + int ret = pipeline.execAsync([&](int batch_code, const std::vector& batch_replies) { + std::lock_guard lock(mutex); + code = batch_code; + replies = batch_replies; + done = true; + cv.notify_one(); + }); + assert(ret == 0); + + std::unique_lock lock(mutex); + bool completed = cv.wait_for(lock, std::chrono::seconds(3), [&done]() { return done; }); + assert(completed); + assert(code == 0); + assert(replies.size() == 2); + assert(replies[0].asString() == "OK"); + assert(replies[1].asInt() == 5); + assert(fixture.kv["async-counter"] == 5); + + int second = pipeline.execAsync(NULL); + assert(second == ERR_INVALID_PARAM); +} + +static void test_pipeline_error_reply_marks_failure() { + RedisBatchFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + + std::vector replies; + RedisPipeline pipeline = client.pipeline(); + pipeline.appendCommand(RedisCommand{"SET", "counter", "1"}); + pipeline.appendCommand(RedisCommand{"NOPE"}); + pipeline.appendCommand(RedisCommand{"INCR", "counter"}); + + RedisResult result = pipeline.exec(&replies); + assert(result.code == 0); + assert(!result.ok()); + assert(result.reply.isError()); + assert(result.reply.error() == "ERR unsupported"); + assert(replies.size() == 3); + assert(replies[0].asString() == "OK"); + assert(replies[1].isError()); + assert(replies[2].type == REDIS_REPLY_INTEGER); + assert(replies[2].asInt() == 2); +} + +static void test_transaction_exec() { + RedisBatchFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + + RedisTransaction transaction = client.transaction(); + transaction.appendCommand(RedisCommand{"SET", "tx-counter", "10"}); + transaction.appendCommand(RedisCommand{"INCR", "tx-counter"}); + + std::vector replies; + RedisResult result = transaction.exec(&replies); + assert(result.code == 0); + assert(result.ok()); + assert(result.reply.type == REDIS_REPLY_ARRAY); + assert(replies.size() == 2); + assert(replies[0].asString() == "OK"); + assert(replies[1].type == REDIS_REPLY_INTEGER); + assert(replies[1].asInt() == 11); + assert(fixture.kv["tx-counter"] == 11); + assert(fixture.commands.size() >= 6); + assert(fixture.commands[fixture.commands.size() - 4][0] == "MULTI"); + assert(fixture.commands[fixture.commands.size() - 3][0] == "SET"); + assert(fixture.commands[fixture.commands.size() - 2][0] == "INCR"); + assert(fixture.commands[fixture.commands.size() - 1][0] == "EXEC"); + + std::vector second_replies; + RedisResult second = transaction.exec(&second_replies); + assert(second.code == ERR_INVALID_PARAM); + assert(second_replies.empty()); +} + +static void test_transaction_exec_nil_reply_fails() { + RedisBatchFixture fixture; + fixture.abort_exec = true; + RedisClient client; + setupClient(&client, fixture.server.port()); + + RedisTransaction transaction = client.transaction(); + transaction.appendCommand(RedisCommand{"SET", "aborted", "1"}); + + std::vector replies; + RedisResult result = transaction.exec(&replies); + assert(result.code == ERR_RESPONSE); + assert(!result.ok()); + assert(result.reply.isNil()); + assert(replies.empty()); +} + +static void test_transaction_exec_error_reply_fails() { + RedisBatchFixture fixture; + fixture.exec_error_reply = true; + RedisClient client; + setupClient(&client, fixture.server.port()); + + RedisTransaction transaction = client.transaction(); + transaction.appendCommand(RedisCommand{"SET", "broken", "1"}); + + std::vector replies; + RedisResult result = transaction.exec(&replies); + assert(result.code == 0); + assert(!result.ok()); + assert(result.reply.isError()); + assert(result.reply.error() == "ERR exec failed"); + assert(replies.size() == 1); + assert(replies[0].isError()); +} + +static void test_transaction_queue_error_fails() { + RedisBatchFixture fixture; + fixture.queue_error = true; + RedisClient client; + setupClient(&client, fixture.server.port()); + + RedisTransaction transaction = client.transaction(); + transaction.appendCommand(RedisCommand{"NOPE"}); + + std::vector replies; + RedisResult result = transaction.exec(&replies); + assert(result.code == 0); + assert(!result.ok()); + assert(result.reply.isError()); + assert(result.reply.error() == "ERR queue rejected"); + assert(replies.empty()); +} + +static void test_transaction_discard() { + RedisBatchFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + + RedisTransaction transaction = client.transaction(); + transaction.appendCommand(RedisCommand{"SET", "discarded", "8"}); + RedisResult discard_result = transaction.discard(); + assert(discard_result.code == 0); + assert(discard_result.ok()); + assert(discard_result.reply.asString() == "OK"); + + std::vector replies; + RedisResult exec_result = transaction.exec(&replies); + assert(exec_result.code == ERR_INVALID_PARAM); + assert(replies.empty()); + assert(fixture.kv.count("discarded") == 0); + for (size_t i = 0; i < fixture.commands.size(); ++i) { + assert(fixture.commands[i][0] != "MULTI"); + assert(fixture.commands[i][0] != "EXEC"); + } + + RedisResult second_discard = transaction.discard(); + assert(second_discard.code == ERR_INVALID_PARAM); +} + +static void test_redis_client_header_is_self_sufficient() { + RedisClient client; + RedisPipeline pipeline = client.pipeline(); + RedisTransaction transaction = client.transaction(); + (void)pipeline; + (void)transaction; +} + +int main() { + test_pipeline_sync_exec(); + test_pipeline_async_exec(); + test_pipeline_error_reply_marks_failure(); + test_transaction_exec(); + test_transaction_exec_nil_reply_fails(); + test_transaction_exec_error_reply_fails(); + test_transaction_queue_error_fails(); + test_transaction_discard(); + test_redis_client_header_is_self_sufficient(); + return 0; +} diff --git a/unittest/redis_client_test.cpp b/unittest/redis_client_test.cpp new file mode 100644 index 000000000..2988a2ff6 --- /dev/null +++ b/unittest/redis_client_test.cpp @@ -0,0 +1,358 @@ +#include "redis/RedisClient.h" +#include "redis_test_server.h" + +#include +#include +#include +#include +#include +#include +#include + +using namespace hv; + +struct RedisFixture { + std::map kv; + std::map > hashes; + std::vector published_messages; + std::vector commands; + FakeRedisServer server; + + RedisFixture() { + server.setCommandHandler([this](const RedisCommand& cmd) { + commands.push_back(cmd); + RedisReply reply; + if (cmd.empty()) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR empty command"; + return reply; + } + if (cmd[0] == "AUTH") { + if (cmd.size() != 2) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'AUTH'"; + return reply; + } + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + } + else if (cmd[0] == "SELECT") { + if (cmd.size() != 2) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'SELECT'"; + return reply; + } + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + } + else if (cmd[0] == "SET") { + if (cmd.size() != 3) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'SET'"; + return reply; + } + kv[cmd[1]] = cmd[2]; + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + } + else if (cmd[0] == "GET") { + if (cmd.size() != 2) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'GET'"; + return reply; + } + std::map::const_iterator it = kv.find(cmd[1]); + if (it == kv.end()) { + reply.type = REDIS_REPLY_NIL; + } + else { + reply.type = REDIS_REPLY_STRING; + reply.str = it->second; + reply.bulk = true; + } + } + else if (cmd[0] == "DEL") { + if (cmd.size() != 2) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'DEL'"; + return reply; + } + reply.type = REDIS_REPLY_INTEGER; + reply.integer = kv.erase(cmd[1]); + } + else if (cmd[0] == "EXISTS") { + if (cmd.size() != 2) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'EXISTS'"; + return reply; + } + reply.type = REDIS_REPLY_INTEGER; + reply.integer = kv.count(cmd[1]) ? 1 : 0; + } + else if (cmd[0] == "EXPIRE") { + if (cmd.size() != 3) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'EXPIRE'"; + return reply; + } + reply.type = REDIS_REPLY_INTEGER; + reply.integer = kv.count(cmd[1]) ? 1 : 0; + } + else if (cmd[0] == "HSET") { + if (cmd.size() != 4) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'HSET'"; + return reply; + } + bool inserted = hashes[cmd[1]].count(cmd[2]) == 0; + hashes[cmd[1]][cmd[2]] = cmd[3]; + reply.type = REDIS_REPLY_INTEGER; + reply.integer = inserted ? 1 : 0; + } + else if (cmd[0] == "HGET") { + if (cmd.size() != 3) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'HGET'"; + return reply; + } + std::map >::const_iterator hash_it = hashes.find(cmd[1]); + if (hash_it == hashes.end() || hash_it->second.count(cmd[2]) == 0) { + reply.type = REDIS_REPLY_NIL; + } + else { + reply.type = REDIS_REPLY_STRING; + reply.str = hash_it->second.find(cmd[2])->second; + reply.bulk = true; + } + } + else if (cmd[0] == "PUBLISH") { + if (cmd.size() != 3) { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR wrong number of arguments for 'PUBLISH'"; + return reply; + } + published_messages.push_back(cmd[1] + ":" + cmd[2]); + reply.type = REDIS_REPLY_INTEGER; + reply.integer = 1; + } + else { + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR unsupported"; + } + return reply; + }); + server.start(); + } + + ~RedisFixture() { + server.stop(); + } +}; + +static void setupClient(RedisClient* client, int port) { + client->setHost("127.0.0.1"); + client->setPort(port); + client->setAuth("secret"); + client->setDb(2); + client->setTimeout(3000); + client->setConnectTimeout(3000); +} + +static bool hasCommand(const std::vector& commands, const char* name) { + for (size_t i = 0; i < commands.size(); ++i) { + if (!commands[i].empty() && commands[i][0] == name) { + return true; + } + } + return false; +} + +static void test_sync_command_and_helpers() { + RedisFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + + RedisResult set_result = client.set("name", "libhv"); + assert(set_result.code == 0); + assert(set_result.ok()); + assert(set_result.reply.asString() == "OK"); + assert(hasCommand(fixture.commands, "AUTH")); + assert(hasCommand(fixture.commands, "SELECT")); + + RedisValueResult get_result = client.get("name"); + assert(get_result.ok()); + assert(get_result.value == "libhv"); + + RedisValueResult exists_result = client.exists("name"); + assert(exists_result.ok()); + assert(exists_result.value == 1); + + RedisValueResult expire_result = client.expire("name", 60); + assert(expire_result.ok()); + assert(expire_result.value == 1); + + RedisValueResult hset_result = client.hset("user:1", "lang", "cpp"); + assert(hset_result.ok()); + assert(hset_result.value == 1); + + RedisValueResult hget_result = client.hget("user:1", "lang"); + assert(hget_result.ok()); + assert(hget_result.value == "cpp"); + + RedisValueResult publish_result = client.publish("updates", "hello"); + assert(publish_result.ok()); + assert(publish_result.value == 1); + assert(fixture.published_messages.size() == 1); + assert(fixture.published_messages[0] == "updates:hello"); + + RedisResult commandf_result = client.commandf("SET %s %s", "language", "cxx"); + assert(commandf_result.ok()); + RedisValueResult commandf_get = client.get("language"); + assert(commandf_get.ok()); + assert(commandf_get.value == "cxx"); + + RedisValueResult del_result = client.del("name"); + assert(del_result.ok()); + assert(del_result.value == 1); + + RedisValueResult missing_result = client.get("name"); + assert(!missing_result.ok()); + assert(missing_result.isNil()); + + RedisResult unsupported = client.command(RedisCommand{"NOPE"}); + assert(unsupported.code == 0); + assert(!unsupported.ok()); + assert(unsupported.reply.isError()); +} + +static void test_async_helpers() { + RedisFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + + std::mutex mutex; + std::condition_variable cv; + int completed = 0; + int expected = 0; + bool all_ok = true; + + auto done = [&](bool ok) { + std::lock_guard lock(mutex); + all_ok = all_ok && ok; + ++completed; + cv.notify_one(); + }; + + int rc = client.setAsync("async:key", "value", [&](const RedisResult& result) { + done(result.ok() && result.reply.asString() == "OK"); + }); + assert(rc == 0); + ++expected; + + rc = client.getAsync("async:key", [&](const RedisValueResult& result) { + done(result.ok() && result.value == "value"); + }); + assert(rc == 0); + ++expected; + + rc = client.existsAsync("async:key", [&](const RedisValueResult& result) { + done(result.ok() && result.value == 1); + }); + assert(rc == 0); + ++expected; + + rc = client.expireAsync("async:key", 10, [&](const RedisValueResult& result) { + done(result.ok() && result.value == 1); + }); + assert(rc == 0); + ++expected; + + rc = client.hsetAsync("user:2", "role", "tester", [&](const RedisValueResult& result) { + done(result.ok() && result.value == 1); + }); + assert(rc == 0); + ++expected; + + rc = client.hgetAsync("user:2", "role", [&](const RedisValueResult& result) { + done(result.ok() && result.value == "tester"); + }); + assert(rc == 0); + ++expected; + + rc = client.commandfAsync("SET %s %s", [&](const RedisResult& result) { + done(result.ok() && result.reply.asString() == "OK"); + }, "fmt:key", "fmt-value"); + assert(rc == 0); + ++expected; + + rc = client.getAsync("missing:key", [&](const RedisValueResult& result) { + done(result.isNil()); + }); + assert(rc == 0); + ++expected; + + rc = client.commandAsync(RedisCommand{"NOPE"}, [&](const RedisResult& result) { + done(result.code == 0 && result.reply.isError()); + }); + assert(rc == 0); + ++expected; + + rc = client.delAsync("async:key", [&](const RedisValueResult& result) { + done(result.ok() && result.value == 1); + }); + assert(rc == 0); + ++expected; + + rc = client.publishAsync("events", "async", [&](const RedisValueResult& result) { + done(result.ok() && result.value == 1); + }); + assert(rc == 0); + ++expected; + + std::unique_lock lock(mutex); + bool ready = cv.wait_for(lock, std::chrono::seconds(3), [&completed, &expected]() { return completed == expected; }); + assert(ready); + assert(all_ok); + assert(hasCommand(fixture.commands, "AUTH")); + assert(hasCommand(fixture.commands, "SELECT")); + RedisValueResult fmt_result = client.get("fmt:key"); + assert(fmt_result.ok()); + assert(fmt_result.value == "fmt-value"); + assert(fixture.published_messages.size() == 1); + assert(fixture.published_messages[0] == "events:async"); +} + +static void test_sync_command_in_async_callback_is_rejected() { + RedisFixture fixture; + RedisClient client; + setupClient(&client, fixture.server.port()); + assert(client.set("nested:key", "nested-value").ok()); + + std::mutex mutex; + std::condition_variable cv; + bool done = false; + RedisValueResult nested_result; + + int rc = client.getAsync("nested:key", [&](const RedisValueResult& result) { + assert(result.ok()); + RedisValueResult local = client.get("nested:key"); + std::lock_guard lock(mutex); + nested_result = local; + done = true; + cv.notify_one(); + }); + assert(rc == 0); + + std::unique_lock lock(mutex); + bool ready = cv.wait_for(lock, std::chrono::seconds(3), [&done]() { return done; }); + assert(ready); + assert(nested_result.code == ERR_INVALID_HANDLE); + assert(!nested_result.ok()); +} + +int main() { + test_sync_command_and_helpers(); + test_async_helpers(); + test_sync_command_in_async_callback_is_rejected(); + return 0; +} diff --git a/unittest/redis_protocol_test.cpp b/unittest/redis_protocol_test.cpp new file mode 100644 index 000000000..05ce52e22 --- /dev/null +++ b/unittest/redis_protocol_test.cpp @@ -0,0 +1,167 @@ +#include + +#include + +#include "redis/RedisMessage.h" + +using namespace hv; + +static void test_encode_command() { + RedisCommand cmd; + cmd.push_back("SET"); + cmd.push_back("key"); + cmd.push_back("value"); + std::string wire = RedisEncodeCommand(cmd); + assert(wire == "*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n"); +} + +static void test_encode_reply() { + RedisReply simple; + simple.type = REDIS_REPLY_STRING; + simple.str = "OK"; + assert(RedisEncodeReply(simple) == "+OK\r\n"); + + RedisReply bulk; + bulk.type = REDIS_REPLY_STRING; + bulk.str = "value"; + bulk.bulk = true; + assert(RedisEncodeReply(bulk) == "$5\r\nvalue\r\n"); + + RedisReply nil; + nil.type = REDIS_REPLY_NIL; + assert(RedisEncodeReply(nil) == "$-1\r\n"); + + RedisReply integer; + integer.type = REDIS_REPLY_INTEGER; + integer.integer = 42; + assert(RedisEncodeReply(integer) == ":42\r\n"); + + RedisReply array; + array.type = REDIS_REPLY_ARRAY; + array.elements.push_back(simple); + array.elements.push_back(integer); + assert(RedisEncodeReply(array) == "*2\r\n+OK\r\n:42\r\n"); +} + +static void test_result_ok_semantics() { + RedisResult success; + success.code = 0; + assert(success.ok()); + + RedisResult client_error; + client_error.code = -1; + assert(!client_error.ok()); + + RedisResult redis_error; + redis_error.code = 0; + redis_error.reply.type = REDIS_REPLY_ERROR; + redis_error.reply.str = "ERR wrongtype"; + assert(!redis_error.ok()); +} + +static void test_parse_fragmented_replies() { + RedisParser parser; + std::string payload = "+OK\r\n$-1\r\n*2\r\n:1\r\n-ERR wrongtype\r\n"; + parser.Feed(payload.data(), 7); + assert(parser.HasReply()); + RedisReply ok = parser.NextReply(); + assert(ok.type == REDIS_REPLY_STRING); + assert(ok.str == "OK"); + assert(!ok.bulk); + + parser.Feed(payload.data() + 7, payload.size() - 7); + RedisReply nil = parser.NextReply(); + assert(nil.type == REDIS_REPLY_NIL); + RedisReply array = parser.NextReply(); + assert(array.type == REDIS_REPLY_ARRAY); + assert(array.elements.size() == 2); + assert(array.elements[0].type == REDIS_REPLY_INTEGER); + assert(array.elements[0].integer == 1); + assert(array.elements[1].type == REDIS_REPLY_ERROR); + assert(array.elements[1].str == "ERR wrongtype"); +} + +static void test_parse_nested_array_and_bulk_string() { + RedisParser parser; + std::string payload = "*3\r\n$5\r\nvalue\r\n$0\r\n\r\n*2\r\n:7\r\n*1\r\n$-1\r\n"; + parser.Feed(payload.data(), payload.size()); + assert(parser.HasReply()); + RedisReply reply = parser.NextReply(); + assert(reply.type == REDIS_REPLY_ARRAY); + assert(reply.elements.size() == 3); + assert(reply.elements[0].type == REDIS_REPLY_STRING); + assert(reply.elements[0].bulk); + assert(reply.elements[0].str == "value"); + assert(reply.elements[1].type == REDIS_REPLY_STRING); + assert(reply.elements[1].bulk); + assert(reply.elements[1].str.empty()); + assert(reply.elements[2].type == REDIS_REPLY_ARRAY); + assert(reply.elements[2].elements.size() == 2); + assert(reply.elements[2].elements[0].type == REDIS_REPLY_INTEGER); + assert(reply.elements[2].elements[0].integer == 7); + assert(reply.elements[2].elements[1].type == REDIS_REPLY_ARRAY); + assert(reply.elements[2].elements[1].elements.size() == 1); + assert(reply.elements[2].elements[1].elements[0].type == REDIS_REPLY_NIL); + assert(!reply.elements[2].elements[1].elements[0].null_array); +} + +static void test_null_array_round_trip() { + RedisParser parser; + std::string payload = "*-1\r\n"; + parser.Feed(payload.data(), payload.size()); + assert(parser.HasReply()); + RedisReply reply = parser.NextReply(); + assert(reply.type == REDIS_REPLY_NIL); + assert(reply.null_array); + assert(RedisEncodeReply(reply) == "*-1\r\n"); +} + +static void test_parse_hard_error_does_not_block_following_reply() { + RedisParser parser; + std::string payload = "?bad\r\n+OK\r\n"; + parser.Feed(payload.data(), payload.size()); + assert(parser.HasError()); + assert(parser.HasReply()); + RedisReply reply = parser.NextReply(); + assert(reply.type == REDIS_REPLY_STRING); + assert(reply.str == "OK"); +} + +static void test_parse_incomplete_does_not_set_error() { + RedisParser parser; + parser.Feed("$5\r\nva", 6); + assert(!parser.HasError()); + assert(!parser.HasReply()); + + parser.Feed("lue\r\n", 5); + assert(!parser.HasError()); + assert(parser.HasReply()); + RedisReply reply = parser.NextReply(); + assert(reply.type == REDIS_REPLY_STRING); + assert(reply.bulk); + assert(reply.str == "value"); +} + +static void test_parse_hard_error_in_nested_array_recovers() { + RedisParser parser; + std::string payload = "*2\r\n:1\r\n$-2\r\n+NEXT\r\n"; + parser.Feed(payload.data(), payload.size()); + assert(parser.HasError()); + assert(parser.HasReply()); + RedisReply reply = parser.NextReply(); + assert(reply.type == REDIS_REPLY_STRING); + assert(reply.str == "NEXT"); +} + +int main() { + test_encode_command(); + test_encode_reply(); + test_result_ok_semantics(); + test_parse_fragmented_replies(); + test_parse_nested_array_and_bulk_string(); + test_null_array_round_trip(); + test_parse_hard_error_does_not_block_following_reply(); + test_parse_incomplete_does_not_set_error(); + test_parse_hard_error_in_nested_array_recovers(); + return 0; +} diff --git a/unittest/redis_subscriber_test.cpp b/unittest/redis_subscriber_test.cpp new file mode 100644 index 000000000..67fc9fd5e --- /dev/null +++ b/unittest/redis_subscriber_test.cpp @@ -0,0 +1,128 @@ +#include "redis/RedisSubscriber.h" +#include "redis_test_server.h" + +#include +#include +#include +#include +#include +#include + +using namespace hv; + +namespace { + +struct SubscriberCapture { + std::mutex mutex; + std::condition_variable cv; + size_t subscribed = 0; + size_t unsubscribed = 0; + std::vector messages; +}; + +static RedisReply authSelectOnlyHandler(const RedisCommand& cmd) { + if (!cmd.empty() && (cmd[0] == "AUTH" || cmd[0] == "SELECT")) { + RedisReply reply; + reply.type = REDIS_REPLY_STRING; + reply.str = "OK"; + return reply; + } + RedisReply reply; + reply.type = REDIS_REPLY_ERROR; + reply.str = "ERR unsupported"; + return reply; +} + +bool waitForCount(std::mutex& mutex, std::condition_variable& cv, size_t* count, size_t expected) { + std::unique_lock lock(mutex); + return cv.wait_for(lock, std::chrono::seconds(3), [count, expected]() { return *count >= expected; }); +} + +void setupSubscriber(RedisSubscriber* subscriber, int port) { + subscriber->setHost("127.0.0.1"); + subscriber->setPort(port); + subscriber->setAuth("secret"); + subscriber->setDb(5); +} + +void bindCallbacks(RedisSubscriber* subscriber, SubscriberCapture* capture, const std::string& name) { + subscriber->onSubscribe = [capture, name](const std::string& subscribed_name) { + std::lock_guard lock(capture->mutex); + if (subscribed_name == name) { + ++capture->subscribed; + } + capture->cv.notify_one(); + }; + subscriber->onUnsubscribe = [capture, name](const std::string& unsubscribed_name) { + std::lock_guard lock(capture->mutex); + if (unsubscribed_name == name) { + ++capture->unsubscribed; + } + capture->cv.notify_one(); + }; + subscriber->onMessage = [capture](const std::string& channel, const std::string& message) { + std::lock_guard lock(capture->mutex); + capture->messages.push_back(channel + ":" + message); + capture->cv.notify_one(); + }; +} + +void expectPublishedMessage(SubscriberCapture* capture, const std::string& expected) { + std::unique_lock lock(capture->mutex); + bool received = capture->cv.wait_for(lock, std::chrono::seconds(3), [capture]() { return capture->messages.size() == 1; }); + assert(received); + assert(capture->messages[0] == expected); +} + +void runSubscriptionFlow(const std::string& subscription_name, + const std::string& publish_channel, + const std::string& expected_message, + int (RedisSubscriber::*subscribe_fn)(const std::string&), + int (RedisSubscriber::*unsubscribe_fn)(const std::string&)) { + FakeRedisServer server; + server.setCommandHandler(authSelectOnlyHandler); + server.start(); + + RedisSubscriber subscriber; + setupSubscriber(&subscriber, server.port()); + + SubscriberCapture capture; + bindCallbacks(&subscriber, &capture, subscription_name); + + subscriber.start(); + assert((subscriber.*subscribe_fn)(subscription_name) == 0); + assert(waitForCount(capture.mutex, capture.cv, &capture.subscribed, 1)); + + assert(server.publish(publish_channel, "hello") == 1); + expectPublishedMessage(&capture, expected_message); + + assert((subscriber.*unsubscribe_fn)(subscription_name) == 0); + assert(waitForCount(capture.mutex, capture.cv, &capture.unsubscribed, 1)); + assert(server.publish(publish_channel, "ignored") == 0); + + subscriber.stop(true); + server.stop(); +} + +} // namespace + +static void test_subscribe_and_message_callback() { + runSubscriptionFlow("news", "news", "news:hello", &RedisSubscriber::subscribe, &RedisSubscriber::unsubscribe); +} + +static void test_psubscribe_and_punsubscribe_callback() { + runSubscriptionFlow("room:*", "room:1", "room:1:hello", &RedisSubscriber::psubscribe, &RedisSubscriber::punsubscribe); +} + +static void test_subscribe_without_start_is_rejected() { + RedisSubscriber subscriber; + assert(subscriber.subscribe("late") == ERR_CONNECT); + assert(subscriber.psubscribe("late:*") == ERR_CONNECT); +} + +int main() { + test_subscribe_and_message_callback(); + test_psubscribe_and_punsubscribe_callback(); + test_subscribe_without_start_is_rejected(); + return 0; +} diff --git a/unittest/redis_test_server.cpp b/unittest/redis_test_server.cpp new file mode 100644 index 000000000..c6bbe3396 --- /dev/null +++ b/unittest/redis_test_server.cpp @@ -0,0 +1,227 @@ +#include "redis_test_server.h" + +#include +#include +#include + +#include "TcpServer.h" +#include "hbase.h" +#include "hsocket.h" + +struct FakeRedisServer::Impl { + struct ClientState { + hv::RedisParser parser; + std::set channels; + std::set patterns; + }; + + int port; + bool close_after_reply; + std::function handler; + std::shared_ptr server; + + Impl() + : port(0) + , close_after_reply(false) {} + + static std::shared_ptr getOrCreateState(const hv::SocketChannelPtr& channel) { + std::shared_ptr state = channel->getContextPtr(); + if (!state) { + state = channel->newContextPtr(); + } + return state; + } + + static hv::RedisReply makeStringReply(const std::string& value, bool bulk = false) { + hv::RedisReply reply; + reply.type = hv::REDIS_REPLY_STRING; + reply.str = value; + reply.bulk = bulk; + return reply; + } + + static hv::RedisReply makeIntegerReply(int64_t value) { + hv::RedisReply reply; + reply.type = hv::REDIS_REPLY_INTEGER; + reply.integer = value; + return reply; + } + + static hv::RedisReply makeErrorReply(const std::string& error) { + hv::RedisReply reply; + reply.type = hv::REDIS_REPLY_ERROR; + reply.str = error; + return reply; + } + + static hv::RedisReply makeArrayReply(const std::string& kind, const std::string& name, int64_t count) { + hv::RedisReply reply; + reply.type = hv::REDIS_REPLY_ARRAY; + reply.elements.push_back(makeStringReply(kind, true)); + reply.elements.push_back(makeStringReply(name, true)); + reply.elements.push_back(makeIntegerReply(count)); + return reply; + } + + static bool commandEquals(const hv::RedisCommand& command, const char* verb) { + return !command.empty() && command[0] == verb; + } + + hv::RedisReply makePubSubReply(const hv::RedisCommand& command, const std::shared_ptr& state) { + if (command.size() != 2) { + return makeErrorReply("ERR wrong number of arguments"); + } + const std::string& name = command[1]; + if (commandEquals(command, "SUBSCRIBE")) { + state->channels.insert(name); + return makeArrayReply("subscribe", name, (int64_t)(state->channels.size() + state->patterns.size())); + } + if (commandEquals(command, "PSUBSCRIBE")) { + state->patterns.insert(name); + return makeArrayReply("psubscribe", name, (int64_t)(state->channels.size() + state->patterns.size())); + } + if (commandEquals(command, "UNSUBSCRIBE")) { + state->channels.erase(name); + return makeArrayReply("unsubscribe", name, (int64_t)(state->channels.size() + state->patterns.size())); + } + state->patterns.erase(name); + return makeArrayReply("punsubscribe", name, (int64_t)(state->channels.size() + state->patterns.size())); + } + + hv::RedisReply makeReply(const hv::RedisReply& request, const std::shared_ptr& state) { + if (!request.isArray()) { + return makeErrorReply("ERR invalid command"); + } + + hv::RedisCommand command; + command.reserve(request.elements.size()); + for (size_t i = 0; i < request.elements.size(); ++i) { + const hv::RedisReply& arg = request.elements[i]; + if (!arg.isString()) { + return makeErrorReply("ERR invalid command arg"); + } + command.push_back(arg.str); + } + + if (commandEquals(command, "SUBSCRIBE") || commandEquals(command, "PSUBSCRIBE") || commandEquals(command, "UNSUBSCRIBE") || commandEquals(command, "PUNSUBSCRIBE")) { + return makePubSubReply(command, state); + } + + if (handler) { + return handler(command); + } + + return makeStringReply("OK"); + } + + int publish(const std::string& channel, const std::string& message) { + if (!server) { + return 0; + } + + int delivered = 0; + server->foreachChannel([&](const hv::SocketChannelPtr& client) { + std::shared_ptr state = client->getContextPtr(); + if (!state) { + return; + } + if (state->channels.count(channel) != 0) { + hv::RedisReply push; + push.type = hv::REDIS_REPLY_ARRAY; + push.elements.push_back(makeStringReply("message", true)); + push.elements.push_back(makeStringReply(channel, true)); + push.elements.push_back(makeStringReply(message, true)); + client->write(hv::RedisEncodeReply(push)); + ++delivered; + } + for (std::set::const_iterator it = state->patterns.begin(); it != state->patterns.end(); ++it) { + if (!hv_wildcard_match(channel.c_str(), it->c_str())) { + continue; + } + hv::RedisReply push; + push.type = hv::REDIS_REPLY_ARRAY; + push.elements.push_back(makeStringReply("pmessage", true)); + push.elements.push_back(makeStringReply(*it, true)); + push.elements.push_back(makeStringReply(channel, true)); + push.elements.push_back(makeStringReply(message, true)); + client->write(hv::RedisEncodeReply(push)); + ++delivered; + } + }); + return delivered; + } +}; + +FakeRedisServer::FakeRedisServer() + : impl_(std::make_shared()) {} + +FakeRedisServer::~FakeRedisServer() { + stop(); +} + +void FakeRedisServer::setPort(int port) { + impl_->port = port; +} + +int FakeRedisServer::port() const { + return impl_->port; +} + +void FakeRedisServer::setCommandHandler(const std::function& handler) { + impl_->handler = handler; +} + +void FakeRedisServer::closeClientAfterReply(bool enabled) { + impl_->close_after_reply = enabled; +} + +int FakeRedisServer::publish(const std::string& channel, const std::string& message) { + return impl_->publish(channel, message); +} + +void FakeRedisServer::start() { + stop(); + impl_->server = std::make_shared(); + impl_->server->createsocket(impl_->port, "127.0.0.1"); + if (impl_->server->listenfd >= 0) { + sockaddr_u addr; + socklen_t len = sizeof(addr); + memset(&addr, 0, sizeof(addr)); + if (getsockname(impl_->server->listenfd, &addr.sa, &len) == 0) { + impl_->port = sockaddr_port(&addr); + } + } + + impl_->server->onConnection = [](const hv::SocketChannelPtr& channel) { + if (channel->isConnected()) { + channel->newContextPtr(); + } + }; + + impl_->server->onMessage = [this](const hv::SocketChannelPtr& channel, hv::Buffer* buf) { + std::shared_ptr state = Impl::getOrCreateState(channel); + state->parser.Feed((const char*)buf->data(), buf->size()); + if (state->parser.HasError()) { + channel->close(); + return; + } + while (state->parser.HasReply()) { + hv::RedisReply request = state->parser.NextReply(); + hv::RedisReply reply = impl_->makeReply(request, state); + channel->write(hv::RedisEncodeReply(reply)); + if (impl_->close_after_reply) { + channel->close(); + return; + } + } + }; + + impl_->server->start(); +} + +void FakeRedisServer::stop() { + if (impl_->server) { + impl_->server->stop(true); + impl_->server.reset(); + } +} diff --git a/unittest/redis_test_server.h b/unittest/redis_test_server.h new file mode 100644 index 000000000..b3fea3053 --- /dev/null +++ b/unittest/redis_test_server.h @@ -0,0 +1,30 @@ +#ifndef UNITTEST_REDIS_TEST_SERVER_H_ +#define UNITTEST_REDIS_TEST_SERVER_H_ + +#include +#include +#include + +#include "redis/RedisMessage.h" + +class FakeRedisServer { +public: + FakeRedisServer(); + ~FakeRedisServer(); + + void setPort(int port); + int port() const; + + void setCommandHandler(const std::function& handler); + void closeClientAfterReply(bool enabled); + int publish(const std::string& channel, const std::string& message); + + void start(); + void stop(); + +private: + struct Impl; + std::shared_ptr impl_; +}; + +#endif // UNITTEST_REDIS_TEST_SERVER_H_ From bed3de6f74ba049b080062896d859294583557aa Mon Sep 17 00:00:00 2001 From: ithewei Date: Sat, 11 Jul 2026 15:41:03 +0800 Subject: [PATCH 2/4] fix: address redis client review findings Fix issues found during code review of the redis module: - RedisMessage: cap array reserve hint (P1). The '*' array parser used the untrusted element count directly in vector::reserve, so a header like '*9999999999999\r\n' could throw length_error/bad_alloc and terminate the process before any element arrived. Reserve a bounded hint and let push_back grow with the actually-received elements. - AsyncRedisClient: disable reconnect on handshake rejection (P1). An AUTH/ SELECT error reply means the password or db is wrong; reconnecting with the same settings produced an endless reconnect + re-auth storm because TcpClient resets its retry counter on every TCP connect. Now setReconnect(NULL) before closing so the failure is reported once. - RedisClient::tokenize: quote-aware splitting for commandf (P1). Previously a value containing whitespace was split into multiple tokens and a wrong command was sent silently. Now supports double/single quotes (redis-cli sdssplitargs style); an unterminated quote yields ERR_INVALID_PARAM. - RedisClient::commandAsync/commandBatchAsync: guard start() with isStarted() (P2), matching the sync path. Avoids a redundant startConnect and a spurious onError(-1) on every async command after the connection is established. --- redis/AsyncRedisClient.cpp | 6 ++++ redis/RedisClient.cpp | 72 +++++++++++++++++++++++++++++++------- redis/RedisMessage.cpp | 9 ++++- 3 files changed, 73 insertions(+), 14 deletions(-) diff --git a/redis/AsyncRedisClient.cpp b/redis/AsyncRedisClient.cpp index b4a0fcacd..6d21b2049 100644 --- a/redis/AsyncRedisClient.cpp +++ b/redis/AsyncRedisClient.cpp @@ -382,6 +382,12 @@ struct AsyncRedisClient::Impl { void handleHandshakeReply(const RedisReply& reply) { if (reply.isError()) { + // Authentication / SELECT rejection is fatal: the password or db is + // wrong and reconnecting with the same settings only produces an + // endless reconnect + re-auth storm (TcpClient resets its retry + // counter on every successful TCP connect). Disable reconnect before + // closing so the failure is reported once and the client stays down. + tcp_client.setReconnect(NULL); handleClientError(ERR_RESPONSE); return; } diff --git a/redis/RedisClient.cpp b/redis/RedisClient.cpp index 213f25595..1d0624db5 100644 --- a/redis/RedisClient.cpp +++ b/redis/RedisClient.cpp @@ -119,7 +119,9 @@ RedisResult RedisClient::command(const RedisCommand& command) { } int RedisClient::commandAsync(const RedisCommand& command, RedisCallback cb) { - async_.start(); + if (!async_.isStarted()) { + async_.start(); + } return async_.command(command, std::move(cb)); } @@ -269,25 +271,69 @@ RedisResult RedisClient::commandBatch(const std::vector& commands, } int RedisClient::commandBatchAsync(const std::vector& commands, RedisRepliesCallback cb) { - async_.start(); + if (!async_.isStarted()) { + async_.start(); + } return async_.commandBatch(commands, std::move(cb)); } RedisCommand RedisClient::tokenize(const std::string& line) { + // Quote-aware splitting (same spirit as redis-cli's sdssplitargs) so that + // arguments containing whitespace can be passed via commandf, e.g. + // commandf("SET %s %s", "key", "\"hello world\"") + // Without this, a value with spaces would be split into multiple tokens and + // a wrong command would be sent silently. On an unterminated quote we return + // an empty command; the caller maps that to ERR_INVALID_PARAM. RedisCommand command; - std::string token; - for (size_t i = 0; i < line.size(); ++i) { - unsigned char ch = (unsigned char)line[i]; - if (std::isspace(ch)) { - if (!token.empty()) { - command.push_back(token); - token.clear(); + size_t i = 0; + size_t n = line.size(); + while (i < n) { + // skip leading whitespace between tokens + while (i < n && std::isspace((unsigned char)line[i])) { + ++i; + } + if (i >= n) { + break; + } + std::string token; + bool in_token = true; + while (in_token && i < n) { + char ch = line[i]; + if (ch == '"') { + // double-quoted: honor backslash escapes, must be closed + ++i; + while (i < n && line[i] != '"') { + if (line[i] == '\\' && i + 1 < n) { + ++i; + } + token.push_back(line[i]); + ++i; + } + if (i >= n) { + return RedisCommand(); // unterminated quote + } + ++i; // consume closing quote + } + else if (ch == '\'') { + // single-quoted: literal, '' inside means a single quote char + ++i; + while (i < n && line[i] != '\'') { + token.push_back(line[i]); + ++i; + } + if (i >= n) { + return RedisCommand(); // unterminated quote + } + ++i; // consume closing quote + } + else if (std::isspace((unsigned char)ch)) { + in_token = false; + } + else { + token.push_back(ch); + ++i; } - continue; } - token.push_back((char)ch); - } - if (!token.empty()) { command.push_back(token); } return command; diff --git a/redis/RedisMessage.cpp b/redis/RedisMessage.cpp index 2ad6d3a9a..0a28719fe 100644 --- a/redis/RedisMessage.cpp +++ b/redis/RedisMessage.cpp @@ -195,7 +195,14 @@ struct RedisParser::Impl { } size_t cursor = next; std::vector elements; - elements.reserve((size_t)number); + // Do not reserve based on the untrusted element count: a huge value + // (e.g. "*9999999999999\r\n") would trigger std::length_error / + // std::bad_alloc before any element data arrives. Reserve a bounded + // hint and let push_back grow as real elements are parsed; every + // element must already exist in the buffer, so memory stays bounded + // by the actually-received bytes. + const int64_t kReserveHint = 1024; + elements.reserve((size_t)(number < kReserveHint ? number : kReserveHint)); for (int64_t i = 0; i < number; ++i) { RedisReply element; ParseStatus status = ParseOne(cursor, element); From 33ed61b412f6c39b111e0a8032975a90dc429ec4 Mon Sep 17 00:00:00 2001 From: ithewei Date: Sat, 11 Jul 2026 16:05:48 +0800 Subject: [PATCH 3/4] feat: --with-redis in CI --- .github/workflows/CI.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 921abb4d4..1024b02cd 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -20,7 +20,7 @@ jobs: run: | sudo apt update sudo apt install libssl-dev libnghttp2-dev - ./configure --with-openssl --with-nghttp2 --with-kcp --with-mqtt + ./configure --with-openssl --with-nghttp2 --with-kcp --with-mqtt --with-redis make libhv evpp - name: test From f1e20b8594d518e46f085ff7dc7a69f1292abef4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 11 Jul 2026 08:56:29 +0000 Subject: [PATCH 4/4] Fix redis review comments: callback on invalid param, reconnect storm, single-quote tokenizer --- config.mk | 21 ++++----------------- redis/AsyncRedisClient.cpp | 9 +++++++++ redis/RedisClient.cpp | 23 ++++++++++++++++++----- redis/RedisSubscriber.cpp | 1 + 4 files changed, 32 insertions(+), 22 deletions(-) diff --git a/config.mk b/config.mk index ad7852caa..4283f5a71 100644 --- a/config.mk +++ b/config.mk @@ -1,37 +1,24 @@ + PREFIX=/usr/local INSTALL_INCDIR=$(PREFIX)/include/hv INSTALL_LIBDIR=$(PREFIX)/lib - BUILD_SHARED=yes BUILD_STATIC=yes - -# modules -# include icmp dns ftp smtp WITH_PROTOCOL=no - WITH_EVPP=yes WITH_HTTP=yes WITH_HTTP_SERVER=yes WITH_HTTP_CLIENT=yes WITH_MQTT=no - -# features -# base/hsocket.h: Unix Domain Socket +WITH_REDIS=yes ENABLE_UDS=no -# base/RAII.cpp: Windows MiniDumpWriteDump ENABLE_WINDUMP=no -# http/http_content.h: KeyValue,QueryParams,MultiPart USE_MULTIMAP=no - -# dependencies -# for http/client WITH_CURL=no -# for http2 WITH_NGHTTP2=no -# for SSL/TLS WITH_OPENSSL=no WITH_GNUTLS=no WITH_MBEDTLS=no - -# rudp WITH_KCP=no +WITH_IO_URING=no +CONFIG_DATE=20260711 \ No newline at end of file diff --git a/redis/AsyncRedisClient.cpp b/redis/AsyncRedisClient.cpp index 6d21b2049..8626c10b6 100644 --- a/redis/AsyncRedisClient.cpp +++ b/redis/AsyncRedisClient.cpp @@ -552,6 +552,9 @@ bool AsyncRedisClient::isInLoopThread() { int AsyncRedisClient::command(const RedisCommand& command, RedisCallback cb) { if (command.empty()) { + if (cb) { + cb(RedisResult(ERR_INVALID_PARAM)); + } return ERR_INVALID_PARAM; } auto request = std::make_shared(); @@ -566,6 +569,9 @@ int AsyncRedisClient::command(const RedisCommand& command, RedisCallback cb) { int AsyncRedisClient::commandBatch(const std::vector& commands, RedisRepliesCallback cb) { if (commands.empty()) { + if (cb) { + cb(ERR_INVALID_PARAM, std::vector()); + } return ERR_INVALID_PARAM; } auto request = std::make_shared(); @@ -573,6 +579,9 @@ int AsyncRedisClient::commandBatch(const std::vector& commands, Re request->batch_callback = std::move(cb); for (size_t i = 0; i < commands.size(); ++i) { if (commands[i].empty()) { + if (request->batch_callback) { + request->batch_callback(ERR_INVALID_PARAM, std::vector()); + } return ERR_INVALID_PARAM; } request->payload += RedisEncodeCommand(commands[i]); diff --git a/redis/RedisClient.cpp b/redis/RedisClient.cpp index 1d0624db5..63489ba8b 100644 --- a/redis/RedisClient.cpp +++ b/redis/RedisClient.cpp @@ -317,14 +317,27 @@ RedisCommand RedisClient::tokenize(const std::string& line) { else if (ch == '\'') { // single-quoted: literal, '' inside means a single quote char ++i; - while (i < n && line[i] != '\'') { - token.push_back(line[i]); - ++i; + bool closed = false; + while (i < n) { + if (line[i] == '\'') { + ++i; + if (i < n && line[i] == '\'') { + token.push_back('\''); + ++i; + } + else { + closed = true; + break; + } + } + else { + token.push_back(line[i]); + ++i; + } } - if (i >= n) { + if (!closed) { return RedisCommand(); // unterminated quote } - ++i; // consume closing quote } else if (std::isspace((unsigned char)ch)) { in_token = false; diff --git a/redis/RedisSubscriber.cpp b/redis/RedisSubscriber.cpp index 8eb284991..df52c4234 100644 --- a/redis/RedisSubscriber.cpp +++ b/redis/RedisSubscriber.cpp @@ -361,6 +361,7 @@ struct RedisSubscriber::Impl { void handleHandshakeReply(const RedisReply& reply) { if (reply.isError()) { + tcp_client.setReconnect(NULL); handleClientError(ERR_RESPONSE); return; }