diff --git a/.github/wordlist.txt b/.github/wordlist.txt
index c62a2332..f59b8316 100644
--- a/.github/wordlist.txt
+++ b/.github/wordlist.txt
@@ -17,6 +17,7 @@ Autoloading
backend
backends
behaviour
+bitwise
boolean
bugfix
CAS
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5c07f2e7..4501a9d9 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -161,7 +161,6 @@ IF(ENABLE_TLS)
ENDIF()
FIND_PACKAGE(OpenSSL REQUIRED)
SET(valkey_tls_sources
- src/cluster_tls.c
src/tls.c)
ADD_LIBRARY(valkey_tls ${valkey_tls_sources})
ADD_LIBRARY(valkey::valkey_tls ALIAS valkey_tls)
diff --git a/Makefile b/Makefile
index 6884607c..59391d2d 100644
--- a/Makefile
+++ b/Makefile
@@ -15,8 +15,8 @@ INCLUDE_DIR = include/valkey
TEST_SRCS = $(TEST_DIR)/client_test.c $(TEST_DIR)/ut_parse_cmd.c $(TEST_DIR)/ut_slotmap_update.c
TEST_BINS = $(patsubst $(TEST_DIR)/%.c,$(TEST_DIR)/%,$(TEST_SRCS))
-SOURCES = $(filter-out $(wildcard $(SRC_DIR)/*tls.c) $(SRC_DIR)/rdma.c, $(wildcard $(SRC_DIR)/*.c))
-HEADERS = $(filter-out $(wildcard $(INCLUDE_DIR)/*tls.h) $(INCLUDE_DIR)/rdma.h, $(wildcard $(INCLUDE_DIR)/*.h))
+SOURCES = $(filter-out $(SRC_DIR)/tls.c $(SRC_DIR)/rdma.c, $(wildcard $(SRC_DIR)/*.c))
+HEADERS = $(filter-out $(INCLUDE_DIR)/tls.h $(INCLUDE_DIR)/rdma.h, $(wildcard $(INCLUDE_DIR)/*.h))
OBJS = $(patsubst $(SRC_DIR)/%.c,$(OBJ_DIR)/%.o,$(SOURCES))
@@ -95,7 +95,7 @@ TLS_DYLIB_MAKE_CMD=$(CC) $(OPTIMIZATION) $(PLATFORM_FLAGS) -shared -Wl,-soname,$
USE_TLS?=0
ifeq ($(USE_TLS),1)
- TLS_SOURCES = $(wildcard $(SRC_DIR)/*tls.c)
+ TLS_SOURCES = $(SRC_DIR)/tls.c
TLS_OBJS = $(patsubst $(SRC_DIR)/%.c,$(OBJ_DIR)/%.o,$(TLS_SOURCES))
# This is required for test.c only
@@ -293,7 +293,6 @@ install: $(DYLIBNAME) $(STLIBNAME) $(PKGCONFNAME) $(TLS_INSTALL)
install-tls: $(TLS_DYLIBNAME) $(TLS_STLIBNAME) $(TLS_PKGCONFNAME)
mkdir -p $(INSTALL_INCLUDE_PATH) $(INSTALL_LIBRARY_PATH)
$(INSTALL) $(INCLUDE_DIR)/tls.h $(INSTALL_INCLUDE_PATH)
- $(INSTALL) $(INCLUDE_DIR)/cluster_tls.h $(INSTALL_INCLUDE_PATH)
$(INSTALL) $(TLS_DYLIBNAME) $(INSTALL_LIBRARY_PATH)/$(TLS_DYLIB_MINOR_NAME)
ln -sf $(TLS_DYLIB_MINOR_NAME) $(INSTALL_LIBRARY_PATH)/$(TLS_ROOT_DYLIB_NAME)
ln -sf $(TLS_DYLIB_MINOR_NAME) $(INSTALL_LIBRARY_PATH)/$(TLS_DYLIB_MAJOR_NAME)
diff --git a/docs/cluster.md b/docs/cluster.md
index 320476cf..a08ef367 100644
--- a/docs/cluster.md
+++ b/docs/cluster.md
@@ -7,6 +7,7 @@ It is not intended as a complete reference. For that it's always best to refer t
- [Synchronous API](#synchronous-api)
- [Connecting](#connecting)
+ - [Connection options](#connection-options)
- [Executing commands](#executing-commands)
- [Executing commands on a specific node](#executing-commands-on-a-specific-node)
- [Disconnecting/cleanup](#disconnecting-cleanup)
@@ -14,11 +15,13 @@ It is not intended as a complete reference. For that it's always best to refer t
- [Events](#events)
- [Asynchronous API](#asynchronous-api)
- [Connecting](#connecting-1)
+ - [Connection options](#connection-options-1)
- [Executing commands](#executing-commands-1)
- [Executing commands on a specific node](#executing-commands-on-a-specific-node-1)
- [Disconnecting/cleanup](#disconnecting-cleanup-1)
- [Events](#events-1)
- [Miscellaneous](#miscellaneous)
+ - [TLS support](#tls-support)
- [Cluster node iterator](#cluster-node-iterator)
- [Extend the list of supported commands](#extend-the-list-of-supported-commands)
- [Random number generator](#random-number-generator)
@@ -27,37 +30,58 @@ It is not intended as a complete reference. For that it's always best to refer t
### Connecting
-There are a variety of ways to setup and connect to a cluster.
-The most basic alternative lacks many options, but can be enough for some use cases.
+There are a few alternative ways to setup and connect to a cluster.
+The basic alternatives lacks most options, but can be enough for some use cases.
```c
-valkeyClusterContext *cc = valkeyClusterConnect("127.0.0.1:6379", VALKEYCLUSTER_FLAG_NULL);
+valkeyClusterContext *valkeyClusterConnect(const char *addrs);
+valkeyClusterContext *valkeyClusterConnectWithTimeout(const char *addrs,
+ const struct timeval tv);
+```
+
+There is also a convenience struct to specify various options.
+
+```c
+valkeyClusterContext *valkeyClusterConnectWithOptions(const valkeyClusterOptions *options);
+```
+
+When connecting to a cluster, `NULL` is returned when the context can't be allocated, or `err` and `errstr` are set in the returned allocated context when there are issues.
+So when connecting it's simple to handle error states.
+
+```c
+valkeyClusterContext *cc = valkeyClusterConnect("127.0.0.1:6379,127.0.0.1:6380");
if (cc == NULL || cc->err) {
fprintf(stderr, "Error: %s\n", cc ? cc->errstr : "OOM");
}
```
-When connecting to a cluster, libvalkey will return `NULL` in the event that we can't allocate the context, and set the `err` member if we can connect but there are issues.
-So when connecting it's simple to handle error states.
-
-To be able to set options before any connection attempt is performed a `valkeyClusterContext` can be created using `valkeyClusterContextInit`.
-The context is where the known cluster state and the options are kept, which can be configured using functions like
-`valkeyClusterSetOptionAddNodes` to add one or many cluster node addresses,
-or `valkeyClusterSetOptionUsername` together with `valkeyClusterSetOptionPassword` to configure authentication and so on.
-See [`include/valkey/cluster.h`](../include/valkey/cluster.h) for more details.
+### Connection options
-The function `valkeyClusterConnect2` is provided to connect using the prepared `valkeyClusterContext`.
+There are a variety of options you can specify using the `valkeyClusterOptions` struct when connecting to a cluster.
+This includes information about how to connect to the cluster and defining optional callbacks and other options.
+See [include/valkey/cluster.h](../include/valkey/cluster.h) for more details.
```c
-valkeyClusterContext *cc = valkeyClusterContextInit();
-valkeyClusterSetOptionAddNodes(cc, "127.0.0.1:6379,127.0.0.1:6380");
-int status = valkeyClusterConnect2(cc);
-if (status == VALKEY_ERR) {
- printf("Error: %s\n", cc->errstr);
- // handle error
+valkeyClusterOptions opt = {
+ .initial_nodes = "127.0.0.1:6379,127.0.0.1:6380"; // Addresses to initially connect to.
+ .options = VALKEY_OPT_USE_CLUSTER_NODES; // See available option flags below.
+ .password = "password"; // Authenticate connections using the `AUTH` command.
+};
+
+valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&opt);
+if (cc == NULL || cc->err) {
+ fprintf(stderr, "Error: %s\n", cc ? cc->errstr : "OOM");
}
```
+There are also several flags you can specify in `valkeyClusterOptions.options`. It's a bitwise OR of the following flags:
+
+| Flag | Description |
+| --- | --- |
+| `VALKEY_OPT_USE_CLUSTER_NODES` | Tells libvalkey to use the command `CLUSTER NODES` when updating its slot map (cluster topology).
Libvalkey uses `CLUSTER SLOTS` by default. |
+| `VALKEY_OPT_USE_REPLICAS` | Tells libvalkey to keep parsed information of replica nodes. |
+| `VALKEY_OPT_BLOCKING_INITIAL_UPDATE` | **ASYNC**: Tells libvalkey to perform the initial slot map update in a blocking fashion. The function call will wait for a slot map update before returning so that the returned context is immediately ready to accept commands. |
+
### Executing commands
The primary command interface is a `printf`-like function that takes a format string along with a variable number of arguments.
@@ -92,7 +116,7 @@ When there is a need to send commands to a specific node, the following low-leve
valkeyReply *reply = valkeyClusterCommandToNode(cc, node, "DBSIZE");
```
-This function handles `printf`-like arguments similar to `valkeyClusterCommand()`, but will only attempt to send the command to the given node and will not perform redirects or retries.
+This function handles `printf`-like arguments similar to `valkeyClusterCommand`, but will only attempt to send the command to the given node and will not perform redirects or retries.
If the command times out or the connection to the node fails, a slot map update is scheduled to be performed when the next command is sent.
`valkeyClusterCommandToNode` also performs a slot map update if it has previously been scheduled.
@@ -121,6 +145,7 @@ if (valkeyClusterAppendCommand(cc, "SET foo bar") != VALKEY_OK) {
fprintf(stderr, "Error appending command: %s\n", cc->errstr);
exit(1);
}
+
if (valkeyClusterAppendCommand(cc, "GET foo") != VALKEY_OK) {
fprintf(stderr, "Error appending command: %s\n", cc->errstr);
exit(1);
@@ -128,15 +153,15 @@ if (valkeyClusterAppendCommand(cc, "GET foo") != VALKEY_OK) {
valkeyReply *reply;
if (valkeyClusterGetReply(cc,&reply) != VALKEY_OK) {
- fprintf(stderr, "Error reading reply %zu: %s\n", i, c->errstr);
- exit(1);
+ fprintf(stderr, "Error reading reply %zu: %s\n", i, c->errstr);
+ exit(1);
}
// Handle the reply for SET here.
freeReplyObject(reply);
if (valkeyClusterGetReply(cc,&reply) != VALKEY_OK) {
- fprintf(stderr, "Error reading reply %zu: %s\n", i, c->errstr);
- exit(1);
+ fprintf(stderr, "Error reading reply %zu: %s\n", i, c->errstr);
+ exit(1);
}
// Handle the reply for GET here.
freeReplyObject(reply);
@@ -149,10 +174,18 @@ freeReplyObject(reply);
There is a hook to get notified when certain events occur.
```c
-int valkeyClusterSetEventCallback(valkeyClusterContext *cc,
- void(fn)(const valkeyClusterContext *cc, int event,
- void *privdata),
- void *privdata);
+/* Function to be called when events occur. */
+void event_cb(const valkeyClusterContext *cc, int event, void *privdata) {
+ switch (event) {
+ // Handle event
+ }
+}
+
+valkeyClusterOptions opt = {
+ .event_callback = event_cb;
+ .event_privdata = my_privdata; // User defined data can be provided to the callback.
+};
+valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&opt);
```
The callback is called with `event` set to one of the following values:
@@ -160,7 +193,7 @@ The callback is called with `event` set to one of the following values:
* `VALKEYCLUSTER_EVENT_SLOTMAP_UPDATED` when the slot mapping has been updated;
* `VALKEYCLUSTER_EVENT_READY` when the slot mapping has been fetched for the first
time and the client is ready to accept commands, useful when initiating the
- client with `valkeyClusterAsyncConnect2()` where a client is not immediately
+ client with `valkeyClusterAsyncConnect` where a client is not immediately
ready after a successful call;
* `VALKEYCLUSTER_EVENT_FREE_CONTEXT` when the cluster context is being freed, so
that the user can free the event `privdata`.
@@ -169,11 +202,17 @@ The callback is called with `event` set to one of the following values:
There is a hook to get notified about connect and reconnect attempts.
This is useful for applying socket options or access endpoint information for a connection to a particular node.
-The callback is registered using the following function:
+The callback is registered using an option.
```c
-int valkeyClusterSetConnectCallback(valkeyClusterContext *cc,
- void(fn)(const valkeyContext *c, int status));
+void connect_cb(const valkeyContext *c, int status) {
+ // Perform desired action
+}
+
+valkeyClusterOptions opt = {
+ .connect_callback = connect_cb;
+};
+valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&opt);
```
The callback is called just after connect, before TLS handshake and authentication.
@@ -186,51 +225,65 @@ The `err` field in the `valkeyContext` can be used to find out the cause of the
## Asynchronous API
-### Connecting
+The asynchronous API supports a wide range of event libraries and uses [adapters](../include/valkey/adapters/) to attach to a specific event library.
+Each adapter provide a convenience function that configures which event loop instance the created context will be attached to.
-There are two alternative ways to initiate a cluster client, which also determines how the client behaves during the initial connect.
+### Connecting
-The first alternative is to use the function `valkeyClusterAsyncConnect`, which initially connects to the cluster in a blocking fashion and waits for the slot map before returning.
-Any command sent by the user thereafter will create a new non-blocking connection, unless a non-blocking connection already exists to the destination.
-The function returns a pointer to a newly created `valkeyClusterAsyncContext` struct and its `err` field should be checked to make sure the initial slot map update was successful.
+To asynchronously connect to a cluster a `valkeyClusterOptions` should first be initiated with initial nodes and more,
+but it's also important to configure which event library to use before calling `valkeyClusterAsyncConnectWithOptions`.
```c
-// Insufficient error handling for brevity.
-valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnect("127.0.0.1:6379", VALKEYCLUSTER_FLAG_NULL);
-if (acc->err) {
- printf("error: %s\n", acc->errstr);
- exit(1);
-}
+valkeyClusterOptions options = {
+ .initial_nodes = "127.0.0.1:7000";
+};
-// Attach an event engine. In this example we use libevent.
-struct event_base *base = event_base_new();
-valkeyClusterLibeventAttach(acc, base);
-```
+// Use convenience function to set which event library to use.
+valkeyClusterOptionsUseLibev(&options, EV_DEFAULT);
-The second alternative is to use `valkeyClusterAsyncContextInit` and `valkeyClusterAsyncConnect2` which avoids the initial blocking connect.
-This connection alternative requires an attached event engine when `valkeyClusterAsyncConnect2` is called, but the connect and the initial slot map update is done in a non-blocking fashion.
+// Initiate the context and start connecting to the initial nodes.
+valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+```
-This means that commands sent directly after `valkeyClusterAsyncConnect2` may fail because the initial slot map has not yet been retrieved and the client doesn't know which cluster node to send the command to.
+Since an initial slot map update is performed asynchronously any command sent directly after `valkeyClusterAsyncConnectWithOptions` may fail
+because the initial slot map has not yet been retrieved and the client doesn't know which cluster node to send the command to.
You may use the [`eventCallback`](#events-per-cluster-context-1) to be notified when the slot map is updated and the client is ready to accept commands.
A crude example of using the `eventCallback` can be found in [this test case](../tests/ct_async.c).
+Another option is to enable blocking initial slot map updates using the option `VALKEY_OPT_BLOCKING_INITIAL_UPDATE`.
+When enabled `valkeyClusterAsyncConnectWithOptions` will initially connect to the cluster in a blocking fashion and wait for the slot map before returning.
+Any command sent by the user thereafter will create a new non-blocking connection, unless a non-blocking connection already exists to the destination.
+The function returns a pointer to a newly created `valkeyClusterAsyncContext` struct and its `err` field should be checked to make sure the initial slot map update was successful.
+
+There is also a separate API to perform the context initiation and initial connect in separate steps.
+This is useful when there is a need to provide an event callback with the current `valkeyClusterAsyncContext`.
+The context is first initiated using `valkeyClusterAsyncContextInit` and then `valkeyClusterAsyncConnect` will initiate connection attempts.
+
```c
-// Insufficient error handling for brevity.
-valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
+valkeyClusterOptions options = {
+ .initial_nodes = "127.0.0.1:7000";
+};
+valkeyClusterOptionsUseLibev(&options, EV_DEFAULT);
-// Add a cluster node address for the initial connect.
-valkeyClusterSetOptionAddNodes(acc->cc, "127.0.0.1:6379");
+// Initiate the context.
+valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit(&options);
-// Attach an event engine. In this example we use libevent.
-struct event_base *base = event_base_new();
-valkeyClusterLibeventAttach(acc, base);
+// Set the event callback using the context as privdata.
+valkeyClusterAsyncSetEventCallback(acc, eventCallback, acc);
-if (valkeyClusterAsyncConnect2(acc) != VALKEY_OK) {
- printf("error: %s\n", acc->errstr);
- exit(1);
-}
+// Start connecting to the initial nodes.
+valkeyClusterAsyncConnect(acc)
```
+### Connection options
+
+There is a variety of options you can specify using the `valkeyClusterOptions` struct when connecting to a cluster.
+
+One asynchronous API specific option is `VALKEY_OPT_BLOCKING_INITIAL_UPDATE` which enables the initial slot map update to be performed in a blocking fashion.
+The connect function will wait for a slot map update before returning so that the returned context is immediately ready to accept commands.
+
+See previous [Connection options](#connection-options) section for common options.
+
### Executing commands
Executing commands in an asynchronous context work similarly to the synchronous context, except that you can pass a callback that will be invoked when the reply is received.
@@ -284,18 +337,24 @@ After this, the disconnection callback is executed with the `VALKEY_OK` status a
#### Events per cluster context
-Use [`valkeyClusterSetEventCallback`](#events-per-cluster-context) with `acc->cc` as the context to get notified when certain events occur.
+Use [`event_callback` in `valkeyClusterOptions`](#events-per-cluster-context) to get notified when certain events occur.
+
+When the callback function requires the current `valkeyClusterAsyncContext` it can be provided in the `privdata`.
+In this case initiate the context using `valkeyClusterAsyncContextInit`, set the callback and `privdata` using `valkeyClusterAsyncSetEventCallback`,
+and initiate connection attempts using `valkeyClusterAsyncConnect` as described under the [Connecting](#connecting-1) section.
#### Events per connection
Because the connections that will be created are non-blocking, the kernel is not able to instantly return if the specified host and port is able to accept a connection.
Instead, use a connect callback to be notified when a connection is established or failed.
Similarly, a disconnect callback can be used to be notified about a disconnected connection (either because of an error or per user request).
-The callbacks are installed using the following functions:
+The callbacks can be enabled using the following options when calling `valkeyClusterAsyncConnectWithOptions`:
```c
-status = valkeyClusterAsyncSetConnectCallback(acc, callbackFn);
-status = valkeyClusterAsyncSetDisconnectCallback(acc, callbackFn);
+valkeyClusterOptions opt = {
+ .async_connect_callback = connect_cb;
+ .async_disconnect_callback = disconnect_cb;
+}
```
The connect callback function should have the following prototype, aliased to `valkeyConnectCallback`:
@@ -305,17 +364,43 @@ void(valkeyAsyncContext *ac, int status);
On a connection attempt, the `status` argument is set to `VALKEY_OK` when the connection was successful.
The file description of the connection socket can be retrieved from a `valkeyAsyncContext` as `ac->c->fd`.
-On a disconnect, the `status` argument is set to `VALKEY_OK` when disconnection was initiated by the user, or `VALKEY_ERR` when the disconnection was caused by an error.
-When it is `VALKEY_ERR`, the `err` field in the context can be accessed to find out the cause of the error.
-You don't need to reconnect in the disconnect callback.
-libvalkey will reconnect by itself when the next command is handled.
+The disconnect callback function should have the following prototype, aliased to `valkeyDisconnectCallback`:
+```c
+void(const valkeyAsyncContext *ac, int status);
+```
-Setting the connect and disconnect callbacks can only be done once per context.
-For subsequent calls it will return `VALKEY_ERR`.
+On a disconnection the `status` argument is set to `VALKEY_OK` if it was initiated by the user, or to `VALKEY_ERR` when it was caused by an error.
+When caused by an error the `err` field in the context can be accessed to get the error cause.
+You don't need to reconnect in the disconnect callback since libvalkey will reconnect by itself when the next command is handled.
## Miscellaneous
+### TLS support
+
+TLS support is not enabled by default and requires an explicit build flag as described in [`README.md`](../README.md#building).
+
+When support is enabled, TLS can be enabled on a cluster context using a prepared `valkeyTLSContext` and the options `valkeyClusterOptions.tls` and `valkeyClusterOptions.tls_init_fn`.
+
+```c
+// Initialize the OpenSSL library.
+valkeyInitOpenSSL();
+
+// Initialize a valkeyTLSContext, which holds an OpenSSL context.
+valkeyTLSContext *tls = valkeyCreateTLSContext("ca.crt", NULL, "client.crt",
+ "client.key", NULL, NULL);
+// Set options to enable TLS on context.
+valkeyClusterOptions opt = {
+ .tls = tls;
+ .tls_init_fn = &valkeyInitiateTLSWithContext;
+};
+
+valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&opt);
+if (cc == NULL || cc->err) {
+ fprintf(stderr, "Error: %s\n", cc ? cc->errstr : "OOM");
+}
+```
+
### Cluster node iterator
A `valkeyClusterNodeIterator` can be used to iterate on all known master nodes in a cluster context.
diff --git a/docs/migration-guide.md b/docs/migration-guide.md
index 9ec1ced0..afa1c346 100644
--- a/docs/migration-guide.md
+++ b/docs/migration-guide.md
@@ -27,11 +27,20 @@ The type `sds` is removed from the public API.
## Migrating from `hiredis-cluster` 0.14.0
+* The cluster client initiation procedure is changed and `valkeyClusterOptions`
+ should be used to specify options when creating a context.
+ See documentation for configuration examples when using the
+ [Synchronous API](cluster.md#synchronous-api) or the
+ [Asynchronous API](cluster.md#asynchronous-api).
+ The [examples](../examples/) directory also contains some common client
+ initiation examples that might be helpful.
+* The default command to update the internal slot map is changed to `CLUSTER SLOTS`.
+ `CLUSTER NODES` can be re-enabled through options using `VALKEY_OPT_USE_CLUSTER_NODES`.
+
### Renamed API functions
* `ctx_get_by_node` is renamed to `valkeyClusterGetValkeyContext`.
* `actx_get_by_node` is renamed to `valkeyClusterGetValkeyAsyncContext`.
-* `redisClusterAsyncSetConnectCallbackNC` is renamed to `valkeyClusterAsyncSetConnectCallback`.
### Renamed API defines
@@ -41,14 +50,32 @@ The type `sds` is removed from the public API.
### Removed API functions
-* `redisClusterSetMaxRedirect` removed and replaced with `valkeyClusterSetOptionMaxRetry`.
-* `redisClusterSetOptionAddNode` removed and replaced with `valkeyClusterSetOptionAddNodes`.
- (Note the "s" in the end of the function name.)
+* `redisClusterConnect2` removed, use `valkeyClusterConnectWithOptions`.
+* `redisClusterContextInit` removed, use `valkeyClusterConnectWithOptions`.
+* `redisClusterSetConnectCallback` removed, use `valkeyClusterOptions.connect_callback`.
+* `redisClusterSetEventCallback` removed, use `valkeyClusterOptions.event_callback`.
+* `redisClusterSetMaxRedirect` removed, use `valkeyClusterOptions.max_retry`.
+* `redisClusterSetOptionAddNode` removed, use `valkeyClusterOptions.initial_nodes`.
+* `redisClusterSetOptionAddNodes` removed, use `valkeyClusterOptions.initial_nodes`.
* `redisClusterSetOptionConnectBlock` removed since it was deprecated.
* `redisClusterSetOptionConnectNonBlock` removed since it was deprecated.
+* `redisClusterSetOptionConnectTimeout` removed, use `valkeyClusterOptions.connect_timeout`.
+* `redisClusterSetOptionMaxRetry` removed, use `valkeyClusterOptions.max_retry`.
+* `redisClusterSetOptionParseSlaves` removed, use `valkeyClusterOptions.flags` and `VALKEY_OPT_USE_REPLICAS`.
+* `redisClusterSetOptionPassword` removed, use `valkeyClusterOptions.password`.
+* `redisClusterSetOptionRouteUseSlots` removed, the use of `CLUSTER SLOTS` is enabled by default.
+* `redisClusterSetOptionUsername` removed, use `valkeyClusterOptions.username`.
+* `redisClusterAsyncSetConnectCallback` removed, but `valkeyClusterOptions.async_connect_callback` can be used which accepts a non-const callback function prototype.
+* `redisClusterAsyncSetConnectCallbackNC` removed, use `valkeyClusterOptions.async_connect_callback`.
+* `redisClusterAsyncSetDisconnectCallback` removed, use `valkeyClusterOptions.async_disconnect_callback`.
* `parse_cluster_nodes` removed from API, for internal use only.
* `parse_cluster_slots` removed from API, for internal use only.
-* `redisClusterAsyncSetConnectCallback` is removed, but can be replaced with `valkeyClusterAsyncSetConnectCallback` which accepts the non-const callback function prototype.
+
+### Removed API defines
+
+* `HIRCLUSTER_FLAG_NULL` removed.
+* `HIRCLUSTER_FLAG_ADD_SLAVE` removed, flag can be replaced with an option, see `VALKEY_OPT_USE_REPLICAS`.
+* `HIRCLUSTER_FLAG_ROUTE_USE_SLOTS` removed, the use of `CLUSTER SLOTS` is enabled by default.
### Removed support for splitting multi-key commands per slot
diff --git a/docs/standalone.md b/docs/standalone.md
index c6e87b8c..a204036f 100644
--- a/docs/standalone.md
+++ b/docs/standalone.md
@@ -72,12 +72,12 @@ There are also several flags you can specify when using the `valkeyOptions` help
| Flag | Description |
| --- | --- |
-| VALKEY\_OPT\_NONBLOCK | Tells libvalkey to make a non-blocking connection. |
-| VALKEY\_OPT\_REUSEADDR | Tells libvalkey to set the [SO_REUSEADDR](https://man7.org/linux/man-pages/man7/socket.7.html) socket option |
-| VALKEY\_OPT\_PREFER\_IPV4
VALKEY\_OPT\_PREFER_IPV6
VALKEY\_OPT\_PREFER\_IP\_UNSPEC | Informs libvalkey to either prefer IPv4 or IPv6 when invoking [getaddrinfo](https://man7.org/linux/man-pages/man3/gai_strerror.3.html). `VALKEY_OPT_PREFER_IP_UNSPEC` will cause libvalkey to specify `AF_UNSPEC` in the getaddrinfo call, which means both IPv4 and IPv6 addresses will be searched simultaneously.
Libvalkey prefers IPv4 by default. |
-| VALKEY\_OPT\_NO\_PUSH\_AUTOFREE | Tells libvalkey to not install the default RESP3 PUSH handler (which just intercepts and frees the replies). This is useful in situations where you want to process these messages in-band. |
-| VALKEY\_OPT\_NOAUTOFREEREPLIES | **ASYNC**: tells libvalkey not to automatically invoke `freeReplyObject` after executing the reply callback. |
-| VALKEY\_OPT\_NOAUTOFREE | **ASYNC**: Tells libvalkey not to automatically free the `valkeyAsyncContext` on connection/communication failure, but only if the user makes an explicit call to `valkeyAsyncDisconnect` or `valkeyAsyncFree` |
+| `VALKEY_OPT_NONBLOCK` | Tells libvalkey to make a non-blocking connection. |
+| `VALKEY_OPT_REUSEADDR` | Tells libvalkey to set the [SO_REUSEADDR](https://man7.org/linux/man-pages/man7/socket.7.html) socket option |
+| `VALKEY_OPT_PREFER_IPV4`
`VALKEY_OPT_PREFER_IPV6`
`VALKEY_OPT_PREFER_IP_UNSPEC` | Informs libvalkey to either prefer IPv4 or IPv6 when invoking [getaddrinfo](https://man7.org/linux/man-pages/man3/gai_strerror.3.html). `VALKEY_OPT_PREFER_IP_UNSPEC` will cause libvalkey to specify `AF_UNSPEC` in the getaddrinfo call, which means both IPv4 and IPv6 addresses will be searched simultaneously.
Libvalkey prefers IPv4 by default. |
+| `VALKEY_OPT_NO_PUSH_AUTOFREE` | Tells libvalkey to not install the default RESP3 PUSH handler (which just intercepts and frees the replies). This is useful in situations where you want to process these messages in-band. |
+| `VALKEY_OPT_NOAUTOFREEREPLIES` | **ASYNC**: tells libvalkey not to automatically invoke `freeReplyObject` after executing the reply callback. |
+| `VALKEY_OPT_NOAUTOFREE` | **ASYNC**: Tells libvalkey not to automatically free the `valkeyAsyncContext` on connection/communication failure, but only if the user makes an explicit call to `valkeyAsyncDisconnect` or `valkeyAsyncFree` |
### Executing commands
@@ -298,7 +298,7 @@ Libvalkey also has an asynchronous API which supports a great many different eve
Libvalkey provides an `valkeyAsyncContext` to manage asynchronous connections which works similarly to the synchronous context.
```c
-valkeyAsyncContext *ac = valkeyAsyncConnect("loalhost", 6379);
+valkeyAsyncContext *ac = valkeyAsyncConnect("localhost", 6379);
if (ac == NULL) {
fprintf(stderr, "Error: Out of memory trying to allocate valkeyAsyncContext\n");
exit(1);
diff --git a/examples/cluster-async-tls.c b/examples/cluster-async-tls.c
index 87a346db..1425bf49 100644
--- a/examples/cluster-async-tls.c
+++ b/examples/cluster-async-tls.c
@@ -1,5 +1,5 @@
#include
-#include
+#include
#include
@@ -66,23 +66,22 @@ int main(int argc, char **argv) {
exit(1);
}
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE_TLS);
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
- valkeyClusterSetOptionParseSlaves(acc->cc);
- valkeyClusterSetOptionEnableTLS(acc->cc, tls);
-
- if (valkeyClusterConnect2(acc->cc) != VALKEY_OK) {
- printf("Error: %s\n", acc->cc->errstr);
+ struct event_base *base = event_base_new();
+
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_TLS;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ options.tls = tls;
+ options.tls_init_fn = &valkeyInitiateTLSWithContext;
+ valkeyClusterOptionsUseLibevent(&options, base);
+
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ if (acc == NULL || acc->err != 0) {
+ printf("Error: %s\n", acc ? acc->errstr : "OOM");
exit(-1);
}
- struct event_base *base = event_base_new();
- valkeyClusterLibeventAttach(acc, base);
-
int status;
status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"THE_ID",
"SET %s %s", "key", "value");
diff --git a/examples/cluster-async.c b/examples/cluster-async.c
index 9f52ed81..0c3bdcb2 100644
--- a/examples/cluster-async.c
+++ b/examples/cluster-async.c
@@ -49,53 +49,55 @@ void disconnectCallback(const valkeyAsyncContext *ac, int status) {
int main(int argc, char **argv) {
(void)argc;
(void)argv;
+ struct event_base *base = event_base_new();
+
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = "127.0.0.1:7000";
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
+
printf("Connecting...\n");
- valkeyClusterAsyncContext *cc =
- valkeyClusterAsyncConnect("127.0.0.1:7000", VALKEYCLUSTER_FLAG_NULL);
- if (!cc) {
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ if (!acc) {
printf("Error: Allocation failure\n");
exit(-1);
- } else if (cc->err) {
- printf("Error: %s\n", cc->errstr);
+ } else if (acc->err) {
+ printf("Error: %s\n", acc->errstr);
// handle error
exit(-1);
}
- struct event_base *base = event_base_new();
- valkeyClusterLibeventAttach(cc, base);
- valkeyClusterAsyncSetConnectCallback(cc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(cc, disconnectCallback);
-
int status;
- status = valkeyClusterAsyncCommand(cc, setCallback, (char *)"THE_ID",
+ status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"THE_ID",
"SET %s %s", "key", "value");
if (status != VALKEY_OK) {
- printf("error: err=%d errstr=%s\n", cc->err, cc->errstr);
+ printf("error: err=%d errstr=%s\n", acc->err, acc->errstr);
}
- status = valkeyClusterAsyncCommand(cc, getCallback, (char *)"THE_ID",
+ status = valkeyClusterAsyncCommand(acc, getCallback, (char *)"THE_ID",
"GET %s", "key");
if (status != VALKEY_OK) {
- printf("error: err=%d errstr=%s\n", cc->err, cc->errstr);
+ printf("error: err=%d errstr=%s\n", acc->err, acc->errstr);
}
- status = valkeyClusterAsyncCommand(cc, setCallback, (char *)"THE_ID",
+ status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"THE_ID",
"SET %s %s", "key2", "value2");
if (status != VALKEY_OK) {
- printf("error: err=%d errstr=%s\n", cc->err, cc->errstr);
+ printf("error: err=%d errstr=%s\n", acc->err, acc->errstr);
}
- status = valkeyClusterAsyncCommand(cc, getCallback, (char *)"THE_ID",
+ status = valkeyClusterAsyncCommand(acc, getCallback, (char *)"THE_ID",
"GET %s", "key2");
if (status != VALKEY_OK) {
- printf("error: err=%d errstr=%s\n", cc->err, cc->errstr);
+ printf("error: err=%d errstr=%s\n", acc->err, acc->errstr);
}
printf("Dispatch..\n");
event_base_dispatch(base);
printf("Done..\n");
- valkeyClusterAsyncFree(cc);
+ valkeyClusterAsyncFree(acc);
event_base_free(base);
return 0;
}
diff --git a/examples/cluster-clientside-caching-async.c b/examples/cluster-clientside-caching-async.c
index 4b8052fc..c153ef58 100644
--- a/examples/cluster-clientside-caching-async.c
+++ b/examples/cluster-clientside-caching-async.c
@@ -126,11 +126,8 @@ void disconnectCallback(const valkeyAsyncContext *ac, int status) {
/* Helper to modify keys using a separate client. */
void modifyKey(const char *key, const char *value) {
printf("Modify key: '%s'\n", key);
- valkeyClusterContext *cc = valkeyClusterContextInit();
- int status = valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
- assert(status == VALKEY_OK);
- status = valkeyClusterConnect2(cc);
- assert(status == VALKEY_OK);
+ valkeyClusterContext *cc = valkeyClusterConnect(CLUSTER_NODE);
+ assert(cc);
valkeyReply *reply = valkeyClusterCommand(cc, "SET %s %s", key, value);
assert(reply != NULL);
@@ -142,24 +139,22 @@ void modifyKey(const char *key, const char *value) {
int main(int argc, char **argv) {
(void)argc;
(void)argv;
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
+ struct event_base *base = event_base_new();
+
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
+
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit(&options);
assert(acc);
int status;
- status = valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- assert(status == VALKEY_OK);
- status = valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- assert(status == VALKEY_OK);
- status = valkeyClusterSetEventCallback(acc->cc, eventCallback, acc);
- assert(status == VALKEY_OK);
- status = valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- assert(status == VALKEY_OK);
-
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
+ status = valkeyClusterAsyncSetEventCallback(acc, eventCallback, acc);
assert(status == VALKEY_OK);
- status = valkeyClusterAsyncConnect2(acc);
+ status = valkeyClusterAsyncConnect(acc);
assert(status == VALKEY_OK);
event_base_dispatch(base);
diff --git a/examples/cluster-simple.c b/examples/cluster-simple.c
index 25bfd361..431ca13d 100644
--- a/examples/cluster-simple.c
+++ b/examples/cluster-simple.c
@@ -8,11 +8,11 @@ int main(int argc, char **argv) {
UNUSED(argv);
struct timeval timeout = {1, 500000}; // 1.5s
- valkeyClusterContext *cc = valkeyClusterContextInit();
- valkeyClusterSetOptionAddNodes(cc, "127.0.0.1:7000");
- valkeyClusterSetOptionConnectTimeout(cc, timeout);
- valkeyClusterSetOptionRouteUseSlots(cc);
- valkeyClusterConnect2(cc);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = "127.0.0.1:7000";
+ options.connect_timeout = &timeout;
+
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
if (!cc) {
printf("Error: Allocation failure\n");
exit(-1);
diff --git a/examples/cluster-tls.c b/examples/cluster-tls.c
index 526dadfe..be7ac9ff 100644
--- a/examples/cluster-tls.c
+++ b/examples/cluster-tls.c
@@ -1,7 +1,5 @@
#include
-#include
#include
-#include
#include
#include
@@ -25,13 +23,13 @@ int main(int argc, char **argv) {
struct timeval timeout = {1, 500000}; // 1.5s
- valkeyClusterContext *cc = valkeyClusterContextInit();
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE_TLS);
- valkeyClusterSetOptionConnectTimeout(cc, timeout);
- valkeyClusterSetOptionRouteUseSlots(cc);
- valkeyClusterSetOptionParseSlaves(cc);
- valkeyClusterSetOptionEnableTLS(cc, tls);
- valkeyClusterConnect2(cc);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_TLS;
+ options.connect_timeout = &timeout;
+ options.tls = tls;
+ options.tls_init_fn = &valkeyInitiateTLSWithContext;
+
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
if (!cc) {
printf("Error: Allocation failure\n");
exit(-1);
diff --git a/include/valkey/adapters/ae.h b/include/valkey/adapters/ae.h
index 5d8849d9..7c3df00c 100644
--- a/include/valkey/adapters/ae.h
+++ b/include/valkey/adapters/ae.h
@@ -140,14 +140,15 @@ static int valkeyAeAttachAdapter(valkeyAsyncContext *ac, void *loop) {
}
VALKEY_UNUSED
-static int valkeyClusterAeAttach(valkeyClusterAsyncContext *acc,
- aeEventLoop *loop) {
- if (acc == NULL || loop == NULL) {
+static int valkeyClusterOptionsUseAe(valkeyClusterOptions *options,
+ aeEventLoop *loop) {
+ if (options == NULL || loop == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyAeAttachAdapter;
- acc->attach_data = loop;
+ options->attach_fn = valkeyAeAttachAdapter;
+ options->attach_data = loop;
return VALKEY_OK;
}
+
#endif /* VALKEY_ADAPTERS_AE_H */
diff --git a/include/valkey/adapters/glib.h b/include/valkey/adapters/glib.h
index acd04120..b471b65b 100644
--- a/include/valkey/adapters/glib.h
+++ b/include/valkey/adapters/glib.h
@@ -153,14 +153,14 @@ static int valkeyGlibAttachAdapter(valkeyAsyncContext *ac, void *context) {
}
VALKEY_UNUSED
-static int valkeyClusterGlibAttach(valkeyClusterAsyncContext *acc,
- GMainContext *context) {
- if (acc == NULL) { // A NULL context is accepted.
+static int valkeyClusterOptionsUseGlib(valkeyClusterOptions *options,
+ GMainContext *context) {
+ if (options == NULL) { // A NULL context is accepted.
return VALKEY_ERR;
}
- acc->attach_fn = valkeyGlibAttachAdapter;
- acc->attach_data = context;
+ options->attach_fn = valkeyGlibAttachAdapter;
+ options->attach_data = context;
return VALKEY_OK;
}
diff --git a/include/valkey/adapters/ivykis.h b/include/valkey/adapters/ivykis.h
index cccdb228..a0bf0b82 100644
--- a/include/valkey/adapters/ivykis.h
+++ b/include/valkey/adapters/ivykis.h
@@ -85,17 +85,17 @@ static int valkeyIvykisAttach(valkeyAsyncContext *ac) {
}
/* Internal adapter function with correct function signature. */
-static int valkeyClusterIvykisAttachAdapter(valkeyAsyncContext *ac, VALKEY_UNUSED void *) {
+static int valkeyIvykisAttachAdapter(valkeyAsyncContext *ac, VALKEY_UNUSED void *) {
return valkeyIvykisAttach(ac);
}
VALKEY_UNUSED
-static int valkeyClusterIvykisAttach(valkeyClusterAsyncContext *acc) {
- if (acc == NULL) {
+static int valkeyClusterOptionsUseIvykis(valkeyClusterOptions *options) {
+ if (options == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyClusterIvykisAttachAdapter;
+ options->attach_fn = valkeyIvykisAttachAdapter;
return VALKEY_OK;
}
diff --git a/include/valkey/adapters/libev.h b/include/valkey/adapters/libev.h
index 1a6ee282..cc96d8b6 100644
--- a/include/valkey/adapters/libev.h
+++ b/include/valkey/adapters/libev.h
@@ -193,14 +193,14 @@ static int valkeyLibevAttachAdapter(valkeyAsyncContext *ac, void *loop) {
}
VALKEY_UNUSED
-static int valkeyClusterLibevAttach(valkeyClusterAsyncContext *acc,
- struct ev_loop *loop) {
- if (acc == NULL || loop == NULL) {
+static int valkeyClusterOptionsUseLibev(valkeyClusterOptions *options,
+ struct ev_loop *loop) {
+ if (options == NULL || loop == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyLibevAttachAdapter;
- acc->attach_data = loop;
+ options->attach_fn = valkeyLibevAttachAdapter;
+ options->attach_data = loop;
return VALKEY_OK;
}
diff --git a/include/valkey/adapters/libevent.h b/include/valkey/adapters/libevent.h
index d2ab4fdf..c063486c 100644
--- a/include/valkey/adapters/libevent.h
+++ b/include/valkey/adapters/libevent.h
@@ -182,14 +182,15 @@ static int valkeyLibeventAttachAdapter(valkeyAsyncContext *ac, void *base) {
}
VALKEY_UNUSED
-static int valkeyClusterLibeventAttach(valkeyClusterAsyncContext *acc,
- struct event_base *base) {
- if (acc == NULL || base == NULL) {
+static int valkeyClusterOptionsUseLibevent(valkeyClusterOptions *options,
+ struct event_base *base) {
+ if (options == NULL || base == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyLibeventAttachAdapter;
- acc->attach_data = base;
+ options->attach_fn = valkeyLibeventAttachAdapter;
+ options->attach_data = base;
return VALKEY_OK;
}
+
#endif /* VALKEY_ADAPTERS_LIBEVENT_H */
diff --git a/include/valkey/adapters/libhv.h b/include/valkey/adapters/libhv.h
index 5a13aab3..d73ec1a2 100644
--- a/include/valkey/adapters/libhv.h
+++ b/include/valkey/adapters/libhv.h
@@ -129,14 +129,14 @@ static int valkeyLibhvAttachAdapter(valkeyAsyncContext *ac, void *loop) {
}
VALKEY_UNUSED
-static int valkeyClusterLibhvAttach(valkeyClusterAsyncContext *acc,
- hloop_t *loop) {
- if (acc == NULL || loop == NULL) {
+static int valkeyClusterOptionsUseLibhv(valkeyClusterOptions *options,
+ hloop_t *loop) {
+ if (options == NULL || loop == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyLibhvAttachAdapter;
- acc->attach_data = loop;
+ options->attach_fn = valkeyLibhvAttachAdapter;
+ options->attach_data = loop;
return VALKEY_OK;
}
diff --git a/include/valkey/adapters/libsdevent.h b/include/valkey/adapters/libsdevent.h
index 25fb29db..51151861 100644
--- a/include/valkey/adapters/libsdevent.h
+++ b/include/valkey/adapters/libsdevent.h
@@ -184,14 +184,14 @@ static int valkeyLibsdeventAttachAdapter(valkeyAsyncContext *ac, void *event) {
}
VALKEY_UNUSED
-static int valkeyClusterLibsdeventAttach(valkeyClusterAsyncContext *acc,
- struct sd_event *event) {
- if (acc == NULL || event == NULL) {
+static int valkeyClusterOptionsUseLibsdevent(valkeyClusterOptions *options,
+ struct sd_event *event) {
+ if (options == NULL || event == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyLibsdeventAttachAdapter;
- acc->attach_data = event;
+ options->attach_fn = valkeyLibsdeventAttachAdapter;
+ options->attach_data = event;
return VALKEY_OK;
}
diff --git a/include/valkey/adapters/libuv.h b/include/valkey/adapters/libuv.h
index d443e19c..716e7e66 100644
--- a/include/valkey/adapters/libuv.h
+++ b/include/valkey/adapters/libuv.h
@@ -203,14 +203,15 @@ static int valkeyLibuvAttachAdapter(valkeyAsyncContext *ac, void *loop) {
}
VALKEY_UNUSED
-static int valkeyClusterLibuvAttach(valkeyClusterAsyncContext *acc,
- uv_loop_t *loop) {
- if (acc == NULL || loop == NULL) {
+static int valkeyClusterOptionsUseLibuv(valkeyClusterOptions *options,
+ uv_loop_t *loop) {
+ if (options == NULL || loop == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyLibuvAttachAdapter;
- acc->attach_data = loop;
+ options->attach_fn = valkeyLibuvAttachAdapter;
+ options->attach_data = loop;
return VALKEY_OK;
}
+
#endif /* VALKEY_ADAPTERS_LIBUV_H */
diff --git a/include/valkey/adapters/macosx.h b/include/valkey/adapters/macosx.h
index 962e2b71..78164413 100644
--- a/include/valkey/adapters/macosx.h
+++ b/include/valkey/adapters/macosx.h
@@ -149,14 +149,14 @@ static int valkeyMacOSAttachAdapter(valkeyAsyncContext *ac, void *loop) {
}
VALKEY_UNUSED
-static int valkeyClusterMacOSAttach(valkeyClusterAsyncContext *acc,
- CFRunLoopRef loop) {
- if (acc == NULL || loop == NULL) {
+static int valkeyClusterOptionsUseMacOS(valkeyClusterOptions *options,
+ CFRunLoopRef loop) {
+ if (options == NULL || loop == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyMacOSAttachAdapter;
- acc->attach_data = loop;
+ options->attach_fn = valkeyMacOSAttachAdapter;
+ options->attach_data = loop;
return VALKEY_OK;
}
diff --git a/include/valkey/adapters/poll.h b/include/valkey/adapters/poll.h
index 544a2532..5035e079 100644
--- a/include/valkey/adapters/poll.h
+++ b/include/valkey/adapters/poll.h
@@ -202,12 +202,12 @@ static int valkeyPollAttachAdapter(valkeyAsyncContext *ac, VALKEY_UNUSED void *u
}
VALKEY_UNUSED
-static int valkeyClusterPollAttach(valkeyClusterAsyncContext *acc) {
- if (acc == NULL) {
+static int valkeyClusterOptionsUsePoll(valkeyClusterOptions *options) {
+ if (options == NULL) {
return VALKEY_ERR;
}
- acc->attach_fn = valkeyPollAttachAdapter;
+ options->attach_fn = valkeyPollAttachAdapter;
return VALKEY_OK;
}
diff --git a/include/valkey/cluster.h b/include/valkey/cluster.h
index 36f28751..56a5eef5 100644
--- a/include/valkey/cluster.h
+++ b/include/valkey/cluster.h
@@ -44,19 +44,7 @@
#define VALKEY_ROLE_PRIMARY 1
#define VALKEY_ROLE_REPLICA 2
-/* Configuration flags */
-#define VALKEYCLUSTER_FLAG_NULL 0x0
-/* Flag to enable parsing of slave nodes. Currently not used, but the
- information is added to its master node structure. */
-#define VALKEYCLUSTER_FLAG_ADD_SLAVE 0x1000
-/* Flag to enable routing table updates using the command 'cluster slots'.
- * Default is the 'cluster nodes' command. */
-#define VALKEYCLUSTER_FLAG_ROUTE_USE_SLOTS 0x4000
-/* Flag specific to the async API which means that the user requested a
- * client disconnect or free. */
-#define VALKEYCLUSTER_FLAG_DISCONNECTING 0x8000
-
-/* Events, for valkeyClusterSetEventCallback() */
+/* Events, for event_callback in valkeyClusterOptions. */
#define VALKEYCLUSTER_EVENT_SLOTMAP_UPDATED 1
#define VALKEYCLUSTER_EVENT_READY 2
#define VALKEYCLUSTER_EVENT_FREE_CONTEXT 3
@@ -156,52 +144,83 @@ typedef struct valkeyClusterNodeIterator {
char opaque_data[VALKEY_NODE_ITERATOR_SIZE];
} valkeyClusterNodeIterator;
-/*
- * Synchronous API
- */
+/* --- Configuration options --- */
+
+/* Enable slotmap updates using the command CLUSTER NODES.
+ * Default is the CLUSTER SLOTS command. */
+#define VALKEY_OPT_USE_CLUSTER_NODES 0x1000
+/* Enable parsing of replica nodes. Currently not used, but the
+ * information is added to its primary node structure. */
+#define VALKEY_OPT_USE_REPLICAS 0x2000
+/* Use a blocking slotmap update after an initial async connect. */
+#define VALKEY_OPT_BLOCKING_INITIAL_UPDATE 0x4000
+
+typedef struct {
+ const char *initial_nodes; /* Initial cluster node address(es). */
+ int options; /* Bit field of VALKEY_OPT_xxx */
+ const struct timeval *connect_timeout; /* Timeout value for connect, no timeout if NULL. */
+ const struct timeval *command_timeout; /* Timeout value for commands, no timeout if NULL. */
+ const char *username; /* Authentication username. */
+ const char *password; /* Authentication password. */
+ int max_retry; /* Allowed retry attempts. */
+
+ /* Common callbacks. */
+
+ /* A hook to get notified when certain events occur. The `event` is set to
+ * VALKEYCLUSTER_EVENT_SLOTMAP_UPDATED when the slot mapping has been updated;
+ * VALKEYCLUSTER_EVENT_READY when the slot mapping has been fetched for the first
+ * time and the client is ready to accept commands;
+ * VALKEYCLUSTER_EVENT_FREE_CONTEXT when the cluster context is being freed. */
+ void (*event_callback)(const struct valkeyClusterContext *cc, int event, void *privdata);
+ void *event_privdata;
+
+ /* Synchronous API callbacks. */
+
+ /* A hook for connect and reconnect attempts, e.g. for applying additional
+ * socket options. This is called just after connect, before TLS handshake and
+ * Valkey authentication.
+ *
+ * On successful connection, `status` is set to `VALKEY_OK` and the file
+ * descriptor can be accessed as `c->fd` to apply socket options.
+ *
+ * On failed connection attempt, this callback is called with `status` set to
+ * `VALKEY_ERR`. The `err` field in the `valkeyContext` can be used to find out
+ * the cause of the error. */
+ void (*connect_callback)(const valkeyContext *c, int status);
+
+ /* Asynchronous API callbacks. */
+
+ /* A hook for asynchronous connect or reconnect attempts.
+ *
+ * On successful connection, `status` is set to `VALKEY_OK`.
+ * On failed connection attempt, this callback is called with `status` set to
+ * `VALKEY_ERR`. The `err` field in the `valkeyAsyncContext` can be used to
+ * find out the cause of the error. */
+ void (*async_connect_callback)(struct valkeyAsyncContext *, int status);
+
+ /* A hook for asynchronous disconnections.
+ * Called when either a connection is terminated due to an error or per
+ * user request. The status is set accordingly (VALKEY_OK, VALKEY_ERR). */
+ void (*async_disconnect_callback)(const struct valkeyAsyncContext *, int status);
+
+ /* Event engine attach function, initiated using a engine specific helper. */
+ int (*attach_fn)(valkeyAsyncContext *ac, void *attach_data);
+ void *attach_data;
-valkeyClusterContext *valkeyClusterConnect(const char *addrs, int flags);
-valkeyClusterContext *valkeyClusterConnectWithTimeout(const char *addrs,
- const struct timeval tv,
- int flags);
-int valkeyClusterConnect2(valkeyClusterContext *cc);
+ /* TLS context, initiated using valkeyCreateTLSContext. */
+ void *tls;
+ int (*tls_init_fn)(struct valkeyContext *, struct valkeyTLSContext *);
+} valkeyClusterOptions;
+
+/* --- Synchronous API --- */
-valkeyClusterContext *valkeyClusterContextInit(void);
+valkeyClusterContext *valkeyClusterConnectWithOptions(const valkeyClusterOptions *options);
+valkeyClusterContext *valkeyClusterConnect(const char *addrs);
+valkeyClusterContext *valkeyClusterConnectWithTimeout(const char *addrs, const struct timeval tv);
void valkeyClusterFree(valkeyClusterContext *cc);
-/* Configuration options */
-int valkeyClusterSetOptionAddNodes(valkeyClusterContext *cc, const char *addrs);
-int valkeyClusterSetOptionUsername(valkeyClusterContext *cc,
- const char *username);
-int valkeyClusterSetOptionPassword(valkeyClusterContext *cc,
- const char *password);
-int valkeyClusterSetOptionParseSlaves(valkeyClusterContext *cc);
-int valkeyClusterSetOptionRouteUseSlots(valkeyClusterContext *cc);
-int valkeyClusterSetOptionConnectTimeout(valkeyClusterContext *cc,
- const struct timeval tv);
-int valkeyClusterSetOptionTimeout(valkeyClusterContext *cc,
- const struct timeval tv);
-int valkeyClusterSetOptionMaxRetry(valkeyClusterContext *cc,
- int max_retry_count);
-/* A hook for connect and reconnect attempts, e.g. for applying additional
- * socket options. This is called just after connect, before TLS handshake and
- * Valkey authentication.
- *
- * On successful connection, `status` is set to `VALKEY_OK` and the file
- * descriptor can be accessed as `c->fd` to apply socket options.
- *
- * On failed connection attempt, this callback is called with `status` set to
- * `VALKEY_ERR`. The `err` field in the `valkeyContext` can be used to find out
- * the cause of the error. */
-int valkeyClusterSetConnectCallback(valkeyClusterContext *cc,
- void(fn)(const valkeyContext *c,
- int status));
-
-/* A hook for events. */
-int valkeyClusterSetEventCallback(valkeyClusterContext *cc,
- void(fn)(const valkeyClusterContext *cc,
- int event, void *privdata),
- void *privdata);
+/* Options configurable in runtime. */
+int valkeyClusterSetOptionTimeout(valkeyClusterContext *cc, const struct timeval tv);
/* Blocking
* The following functions will block for a reply, or return NULL if there was
@@ -264,24 +283,22 @@ int valkeyClusterUpdateSlotmap(valkeyClusterContext *cc);
valkeyContext *valkeyClusterGetValkeyContext(valkeyClusterContext *cc,
valkeyClusterNode *node);
-/*
- * Asynchronous API
- */
+/* --- Asynchronous API --- */
-valkeyClusterAsyncContext *valkeyClusterAsyncContextInit(void);
+valkeyClusterAsyncContext *valkeyClusterAsyncConnectWithOptions(const valkeyClusterOptions *options);
+void valkeyClusterAsyncDisconnect(valkeyClusterAsyncContext *acc);
void valkeyClusterAsyncFree(valkeyClusterAsyncContext *acc);
-int valkeyClusterAsyncSetConnectCallback(valkeyClusterAsyncContext *acc,
- valkeyConnectCallback *fn);
-int valkeyClusterAsyncSetDisconnectCallback(valkeyClusterAsyncContext *acc,
- valkeyDisconnectCallback *fn);
+/* Initiate and connect as separate steps. */
+valkeyClusterAsyncContext *valkeyClusterAsyncContextInit(const valkeyClusterOptions *options);
+int valkeyClusterAsyncConnect(valkeyClusterAsyncContext *acc);
-/* Connect and update slotmap, will block until complete. */
-valkeyClusterAsyncContext *valkeyClusterAsyncConnect(const char *addrs,
- int flags);
-/* Connect and update slotmap asynchronously using configured event engine. */
-int valkeyClusterAsyncConnect2(valkeyClusterAsyncContext *acc);
-void valkeyClusterAsyncDisconnect(valkeyClusterAsyncContext *acc);
+/* Callback option configurable after a context initiation, enabling that the
+ * valkeyClusterAsyncContext pointer can be given as privdata in the callback. */
+int valkeyClusterAsyncSetEventCallback(valkeyClusterAsyncContext *acc,
+ void(fn)(const valkeyClusterContext *cc,
+ int event, void *privdata),
+ void *privdata);
/* Commands */
int valkeyClusterAsyncCommand(valkeyClusterAsyncContext *acc,
diff --git a/include/valkey/cluster_tls.h b/include/valkey/cluster_tls.h
deleted file mode 100644
index a081f98f..00000000
--- a/include/valkey/cluster_tls.h
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Copyright (c) 2022, Bjorn Svensson
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are met:
- *
- * * Redistributions of source code must retain the above copyright notice,
- * this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above copyright
- * notice, this list of conditions and the following disclaimer in the
- * documentation and/or other materials provided with the distribution.
- * * Neither the name of Redis nor the names of its contributors may be used
- * to endorse or promote products derived from this software without
- * specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
- * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
- * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
- * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
- * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
- * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
- * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
- * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
- * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
- * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
- * POSSIBILITY OF SUCH DAMAGE.
- */
-#ifndef VALKEY_CLUSTER_TLS_H
-#define VALKEY_CLUSTER_TLS_H
-
-#include "cluster.h"
-#include "tls.h"
-#include "valkey.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-/**
- * Configuration option to enable TLS negotiation on a context.
- */
-int valkeyClusterSetOptionEnableTLS(valkeyClusterContext *cc,
- valkeyTLSContext *tls);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /* VALKEY_CLUSTER_TLS_H */
diff --git a/include/valkey/valkey.h b/include/valkey/valkey.h
index d6e4fc9c..e9e0e00b 100644
--- a/include/valkey/valkey.h
+++ b/include/valkey/valkey.h
@@ -169,6 +169,7 @@ enum valkeyConnectionType {
#define VALKEY_OPT_PREFER_IPV4 0x20 /* Prefer IPv4 in DNS lookups. */
#define VALKEY_OPT_PREFER_IPV6 0x40 /* Prefer IPv6 in DNS lookups. */
#define VALKEY_OPT_PREFER_IP_UNSPEC (VALKEY_OPT_PREFER_IPV4 | VALKEY_OPT_PREFER_IPV6)
+#define VALKEY_OPT_LAST_SA_OPTION 0x40 /* Last defined standalone option. */
/* In Unix systems a file descriptor is a regular signed int, with -1
* representing an invalid descriptor. In Windows it is a SOCKET
diff --git a/src/cluster.c b/src/cluster.c
index 6f277420..6878ec7d 100644
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -49,6 +49,15 @@
#include
#include
+/* Make sure standalone and cluster options don't overlap. */
+vk_static_assert(VALKEY_OPT_USE_CLUSTER_NODES > VALKEY_OPT_LAST_SA_OPTION);
+
+/* Internal option flags. */
+#define VALKEY_FLAG_USE_CLUSTER_NODES 0x1
+#define VALKEY_FLAG_PARSE_REPLICAS 0x2
+#define VALKEY_FLAG_DISCONNECTING 0x4
+#define VALKEY_FLAG_BLOCKING_INITIAL_UPDATE 0x8
+
// Cluster errors are offset by 100 to be sufficiently out of range of
// standard Valkey errors
#define VALKEY_ERR_CLUSTER_TOO_MANY_RETRIES 100
@@ -101,6 +110,10 @@ static void cluster_slot_destroy(cluster_slot *slot);
static int updateNodesAndSlotmap(valkeyClusterContext *cc, dict *nodes);
static int updateSlotMapAsync(valkeyClusterAsyncContext *acc,
valkeyAsyncContext *ac);
+static int valkeyClusterSetOptionAddNodes(valkeyClusterContext *cc, const char *addrs);
+static int valkeyClusterSetOptionConnectTimeout(valkeyClusterContext *cc, const struct timeval tv);
+static int valkeyClusterSetOptionPassword(valkeyClusterContext *cc, const char *password);
+static int valkeyClusterSetOptionUsername(valkeyClusterContext *cc, const char *username);
void listClusterNodeDestructor(void *val) { freeValkeyClusterNode(val); }
@@ -639,7 +652,7 @@ static dict *parse_cluster_slots(valkeyClusterContext *cc, valkeyContext *c,
}
slot = NULL;
- } else if (cc->flags & VALKEYCLUSTER_FLAG_ADD_SLAVE) {
+ } else if (cc->flags & VALKEY_FLAG_PARSE_REPLICAS) {
replica = node_get_with_slots(cc, host, port,
VALKEY_ROLE_REPLICA);
if (replica == NULL) {
@@ -926,7 +939,7 @@ static int parse_cluster_nodes_line(valkeyClusterContext *cc, valkeyContext *c,
static dict *parse_cluster_nodes(valkeyClusterContext *cc, valkeyContext *c, valkeyReply *reply) {
dict *nodes = NULL;
int slot_ranges_found = 0;
- int add_replicas = cc->flags & VALKEYCLUSTER_FLAG_ADD_SLAVE;
+ int add_replicas = cc->flags & VALKEY_FLAG_PARSE_REPLICAS;
dict *replicas = NULL;
if (reply->type != VALKEY_REPLY_STRING) {
@@ -1014,9 +1027,9 @@ static dict *parse_cluster_nodes(valkeyClusterContext *cc, valkeyContext *c, val
/* Sends CLUSTER SLOTS or CLUSTER NODES to the node with context c. */
static int clusterUpdateRouteSendCommand(valkeyClusterContext *cc,
valkeyContext *c) {
- const char *cmd = (cc->flags & VALKEYCLUSTER_FLAG_ROUTE_USE_SLOTS ?
- VALKEY_COMMAND_CLUSTER_SLOTS :
- VALKEY_COMMAND_CLUSTER_NODES);
+ const char *cmd = (cc->flags & VALKEY_FLAG_USE_CLUSTER_NODES ?
+ VALKEY_COMMAND_CLUSTER_NODES :
+ VALKEY_COMMAND_CLUSTER_SLOTS);
if (valkeyAppendCommand(c, cmd) != VALKEY_OK) {
valkeyClusterSetError(cc, c->err, c->errstr);
return VALKEY_ERR;
@@ -1046,10 +1059,10 @@ static int clusterUpdateRouteHandleReply(valkeyClusterContext *cc,
}
dict *nodes;
- if (cc->flags & VALKEYCLUSTER_FLAG_ROUTE_USE_SLOTS) {
- nodes = parse_cluster_slots(cc, c, reply);
- } else {
+ if (cc->flags & VALKEY_FLAG_USE_CLUSTER_NODES) {
nodes = parse_cluster_nodes(cc, c, reply);
+ } else {
+ nodes = parse_cluster_slots(cc, c, reply);
}
freeReplyObject(reply);
return updateNodesAndSlotmap(cc, nodes);
@@ -1243,7 +1256,7 @@ int valkeyClusterUpdateSlotmap(valkeyClusterContext *cc) {
return VALKEY_ERR;
}
-valkeyClusterContext *valkeyClusterContextInit(void) {
+static valkeyClusterContext *valkeyClusterContextInit(const valkeyClusterOptions *options) {
valkeyClusterContext *cc;
cc = vk_calloc(1, sizeof(valkeyClusterContext));
@@ -1262,7 +1275,52 @@ valkeyClusterContext *valkeyClusterContextInit(void) {
}
cc->requests->free = listCommandFree;
- cc->max_retry_count = CLUSTER_DEFAULT_MAX_RETRY_COUNT;
+ if (options->options & VALKEY_OPT_USE_CLUSTER_NODES) {
+ cc->flags |= VALKEY_FLAG_USE_CLUSTER_NODES;
+ }
+ if (options->options & VALKEY_OPT_USE_REPLICAS) {
+ cc->flags |= VALKEY_FLAG_PARSE_REPLICAS;
+ }
+ if (options->options & VALKEY_OPT_BLOCKING_INITIAL_UPDATE) {
+ cc->flags |= VALKEY_FLAG_BLOCKING_INITIAL_UPDATE;
+ }
+ if (options->max_retry > 0) {
+ cc->max_retry_count = options->max_retry;
+ } else {
+ cc->max_retry_count = CLUSTER_DEFAULT_MAX_RETRY_COUNT;
+ }
+ if (options->initial_nodes != NULL &&
+ valkeyClusterSetOptionAddNodes(cc, options->initial_nodes) != VALKEY_OK) {
+ return cc; /* err and errstr already set. */
+ }
+ if (options->connect_timeout != NULL &&
+ valkeyClusterSetOptionConnectTimeout(cc, *options->connect_timeout) != VALKEY_OK) {
+ return cc; /* err and errstr already set. */
+ }
+ if (options->command_timeout != NULL &&
+ valkeyClusterSetOptionTimeout(cc, *options->command_timeout) != VALKEY_OK) {
+ return cc; /* err and errstr already set. */
+ }
+ if (options->username != NULL &&
+ valkeyClusterSetOptionUsername(cc, options->username) != VALKEY_OK) {
+ return cc; /* err and errstr already set. */
+ }
+ if (options->password != NULL &&
+ valkeyClusterSetOptionPassword(cc, options->password) != VALKEY_OK) {
+ return cc; /* err and errstr already set. */
+ }
+ if (options->connect_callback) {
+ cc->on_connect = options->connect_callback;
+ }
+ if (options->event_callback) {
+ cc->event_callback = options->event_callback;
+ cc->event_privdata = options->event_privdata;
+ }
+ if (options->tls) {
+ cc->tls = options->tls;
+ cc->tls_init_fn = options->tls_init_fn;
+ }
+
return cc;
}
@@ -1287,52 +1345,32 @@ void valkeyClusterFree(valkeyClusterContext *cc) {
vk_free(cc);
}
-static valkeyClusterContext *
-valkeyClusterConnectInternal(valkeyClusterContext *cc, const char *addrs) {
- if (valkeyClusterSetOptionAddNodes(cc, addrs) != VALKEY_OK) {
- return cc;
- }
- valkeyClusterUpdateSlotmap(cc);
- return cc;
-}
-
-valkeyClusterContext *valkeyClusterConnect(const char *addrs, int flags) {
- valkeyClusterContext *cc;
-
- cc = valkeyClusterContextInit();
-
+valkeyClusterContext *valkeyClusterConnectWithOptions(const valkeyClusterOptions *options) {
+ valkeyClusterContext *cc = valkeyClusterContextInit(options);
if (cc == NULL) {
return NULL;
}
+ /* Only connect if options are ok. */
+ if (cc->err == 0) {
+ valkeyClusterUpdateSlotmap(cc);
+ }
+ return cc;
+}
- cc->flags = flags;
+valkeyClusterContext *valkeyClusterConnect(const char *addrs) {
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = addrs;
- return valkeyClusterConnectInternal(cc, addrs);
+ return valkeyClusterConnectWithOptions(&options);
}
valkeyClusterContext *valkeyClusterConnectWithTimeout(const char *addrs,
- const struct timeval tv,
- int flags) {
- valkeyClusterContext *cc;
-
- cc = valkeyClusterContextInit();
-
- if (cc == NULL) {
- return NULL;
- }
+ const struct timeval tv) {
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = addrs;
+ options.connect_timeout = &tv;
- cc->flags = flags;
-
- if (cc->connect_timeout == NULL) {
- cc->connect_timeout = vk_malloc(sizeof(struct timeval));
- if (cc->connect_timeout == NULL) {
- return NULL;
- }
- }
-
- memcpy(cc->connect_timeout, &tv, sizeof(struct timeval));
-
- return valkeyClusterConnectInternal(cc, addrs);
+ return valkeyClusterConnectWithOptions(&options);
}
static int valkeyClusterSetOptionAddNode(valkeyClusterContext *cc, const char *addr) {
@@ -1428,8 +1466,8 @@ static int valkeyClusterSetOptionAddNode(valkeyClusterContext *cc, const char *a
return VALKEY_ERR;
}
-int valkeyClusterSetOptionAddNodes(valkeyClusterContext *cc,
- const char *addrs) {
+static int valkeyClusterSetOptionAddNodes(valkeyClusterContext *cc,
+ const char *addrs) {
int ret;
sds *address = NULL;
int address_count = 0;
@@ -1473,8 +1511,8 @@ int valkeyClusterSetOptionAddNodes(valkeyClusterContext *cc,
* Disabled by default. Can be disabled again by providing an
* empty string or a null pointer.
*/
-int valkeyClusterSetOptionUsername(valkeyClusterContext *cc,
- const char *username) {
+static int valkeyClusterSetOptionUsername(valkeyClusterContext *cc,
+ const char *username) {
if (cc == NULL) {
return VALKEY_ERR;
}
@@ -1499,8 +1537,8 @@ int valkeyClusterSetOptionUsername(valkeyClusterContext *cc,
* Configure a password used when connecting to password-protected
* Valkey instances. (See Valkey AUTH command)
*/
-int valkeyClusterSetOptionPassword(valkeyClusterContext *cc,
- const char *password) {
+static int valkeyClusterSetOptionPassword(valkeyClusterContext *cc,
+ const char *password) {
if (cc == NULL) {
return VALKEY_ERR;
@@ -1522,30 +1560,8 @@ int valkeyClusterSetOptionPassword(valkeyClusterContext *cc,
return VALKEY_OK;
}
-int valkeyClusterSetOptionParseSlaves(valkeyClusterContext *cc) {
-
- if (cc == NULL) {
- return VALKEY_ERR;
- }
-
- cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
-
- return VALKEY_OK;
-}
-
-int valkeyClusterSetOptionRouteUseSlots(valkeyClusterContext *cc) {
-
- if (cc == NULL) {
- return VALKEY_ERR;
- }
-
- cc->flags |= VALKEYCLUSTER_FLAG_ROUTE_USE_SLOTS;
-
- return VALKEY_OK;
-}
-
-int valkeyClusterSetOptionConnectTimeout(valkeyClusterContext *cc,
- const struct timeval tv) {
+static int valkeyClusterSetOptionConnectTimeout(valkeyClusterContext *cc,
+ const struct timeval tv) {
if (cc == NULL) {
return VALKEY_ERR;
@@ -1625,35 +1641,6 @@ int valkeyClusterSetOptionTimeout(valkeyClusterContext *cc,
return VALKEY_OK;
}
-int valkeyClusterSetOptionMaxRetry(valkeyClusterContext *cc,
- int max_retry_count) {
- if (cc == NULL || max_retry_count <= 0) {
- return VALKEY_ERR;
- }
-
- cc->max_retry_count = max_retry_count;
-
- return VALKEY_OK;
-}
-
-int valkeyClusterConnect2(valkeyClusterContext *cc) {
-
- if (cc == NULL) {
- return VALKEY_ERR;
- }
-
- if (dictSize(cc->nodes) == 0) {
- valkeyClusterSetError(cc, VALKEY_ERR_OTHER,
- "server address not configured");
- return VALKEY_ERR;
- }
- /* Clear a previously set shutdown flag since we allow a
- * reconnection of an async context using this API (legacy). */
- cc->flags &= ~VALKEYCLUSTER_FLAG_DISCONNECTING;
-
- return valkeyClusterUpdateSlotmap(cc);
-}
-
valkeyContext *valkeyClusterGetValkeyContext(valkeyClusterContext *cc,
valkeyClusterNode *node) {
valkeyContext *c = NULL;
@@ -2123,28 +2110,6 @@ static int prepareCommand(valkeyClusterContext *cc, struct cmd *command) {
return VALKEY_OK;
}
-int valkeyClusterSetConnectCallback(valkeyClusterContext *cc,
- void(fn)(const valkeyContext *c,
- int status)) {
- if (cc->on_connect == NULL) {
- cc->on_connect = fn;
- return VALKEY_OK;
- }
- return VALKEY_ERR;
-}
-
-int valkeyClusterSetEventCallback(valkeyClusterContext *cc,
- void(fn)(const valkeyClusterContext *cc,
- int event, void *privdata),
- void *privdata) {
- if (cc->event_callback == NULL) {
- cc->event_callback = fn;
- cc->event_privdata = privdata;
- return VALKEY_OK;
- }
- return VALKEY_ERR;
-}
-
void *valkeyClusterFormattedCommand(valkeyClusterContext *cc, char *cmd,
int len) {
valkeyReply *reply = NULL;
@@ -2666,24 +2631,6 @@ static inline void valkeyClusterAsyncClearError(valkeyClusterAsyncContext *acc)
acc->errstr[0] = '\0';
}
-static valkeyClusterAsyncContext *
-valkeyClusterAsyncInitialize(valkeyClusterContext *cc) {
- valkeyClusterAsyncContext *acc;
-
- if (cc == NULL) {
- return NULL;
- }
-
- acc = vk_calloc(1, sizeof(valkeyClusterAsyncContext));
- if (acc == NULL)
- return NULL;
-
- acc->cc = cc;
- valkeyClusterAsyncSetError(acc, cc->err, cc->errstr);
-
- return acc;
-}
-
static cluster_async_data *cluster_async_data_create(void) {
/* use calloc to guarantee all fields are zeroed */
return vk_calloc(1, sizeof(cluster_async_data));
@@ -2809,64 +2756,76 @@ valkeyClusterGetValkeyAsyncContext(valkeyClusterAsyncContext *acc,
return ac;
}
-valkeyClusterAsyncContext *valkeyClusterAsyncContextInit(void) {
+valkeyClusterAsyncContext *valkeyClusterAsyncContextInit(const valkeyClusterOptions *options) {
valkeyClusterContext *cc;
valkeyClusterAsyncContext *acc;
- cc = valkeyClusterContextInit();
+ cc = valkeyClusterContextInit(options);
if (cc == NULL) {
return NULL;
}
- acc = valkeyClusterAsyncInitialize(cc);
+ acc = vk_calloc(1, sizeof(valkeyClusterAsyncContext));
if (acc == NULL) {
valkeyClusterFree(cc);
return NULL;
}
+ acc->cc = cc;
+ valkeyClusterAsyncSetError(acc, cc->err, cc->errstr);
+
+ if (options->async_connect_callback != NULL) {
+ acc->onConnect = options->async_connect_callback;
+ }
+ if (options->async_disconnect_callback != NULL) {
+ acc->onDisconnect = options->async_disconnect_callback;
+ }
+ if (options->attach_fn != NULL) {
+ acc->attach_fn = options->attach_fn;
+ acc->attach_data = options->attach_data;
+ }
return acc;
}
-valkeyClusterAsyncContext *valkeyClusterAsyncConnect(const char *addrs,
- int flags) {
-
- valkeyClusterContext *cc;
- valkeyClusterAsyncContext *acc;
-
- cc = valkeyClusterConnect(addrs, flags);
- if (cc == NULL) {
- return NULL;
- }
-
- acc = valkeyClusterAsyncInitialize(cc);
+valkeyClusterAsyncContext *valkeyClusterAsyncConnectWithOptions(const valkeyClusterOptions *options) {
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit(options);
if (acc == NULL) {
- valkeyClusterFree(cc);
return NULL;
}
+ valkeyClusterAsyncConnect(acc);
return acc;
}
-int valkeyClusterAsyncConnect2(valkeyClusterAsyncContext *acc) {
+int valkeyClusterAsyncConnect(valkeyClusterAsyncContext *acc) {
/* An attach function for an async event library is required. */
if (acc->attach_fn == NULL) {
return VALKEY_ERR;
}
- return updateSlotMapAsync(acc, NULL /*any node*/);
-}
-int valkeyClusterAsyncSetConnectCallback(valkeyClusterAsyncContext *acc,
- valkeyConnectCallback *fn) {
- if (acc->onConnect != NULL)
- return VALKEY_ERR;
- acc->onConnect = fn;
- return VALKEY_OK;
+ /* Clear a previously set shutdown flag to allow a
+ * reconnection of an async context using this API. */
+ acc->cc->flags &= ~VALKEY_FLAG_DISCONNECTING;
+
+ /* Use blocking initial slotmap update when configured. */
+ if (acc->cc->flags & VALKEY_FLAG_BLOCKING_INITIAL_UPDATE) {
+ if (valkeyClusterUpdateSlotmap(acc->cc) != VALKEY_OK) {
+ valkeyClusterAsyncSetError(acc, acc->cc->err, acc->cc->errstr);
+ return VALKEY_ERR;
+ }
+ return VALKEY_OK;
+ }
+ /* Use non-blocking initial slotmap update. */
+ return updateSlotMapAsync(acc, NULL /*any node*/);
}
-int valkeyClusterAsyncSetDisconnectCallback(valkeyClusterAsyncContext *acc,
- valkeyDisconnectCallback *fn) {
- if (acc->onDisconnect == NULL) {
- acc->onDisconnect = fn;
+int valkeyClusterAsyncSetEventCallback(valkeyClusterAsyncContext *acc,
+ void(fn)(const valkeyClusterContext *cc,
+ int event, void *privdata),
+ void *privdata) {
+ if (acc->cc->event_callback == NULL) {
+ acc->cc->event_callback = fn;
+ acc->cc->event_privdata = privdata;
return VALKEY_OK;
}
return VALKEY_ERR;
@@ -2967,7 +2926,7 @@ static int updateSlotMapAsync(valkeyClusterAsyncContext *acc,
/* Don't allow concurrent slot map updates. */
return VALKEY_ERR;
}
- if (acc->cc->flags & VALKEYCLUSTER_FLAG_DISCONNECTING) {
+ if (acc->cc->flags & VALKEY_FLAG_DISCONNECTING) {
/* No slot map updates during a cluster client disconnect. */
return VALKEY_ERR;
}
@@ -2986,12 +2945,12 @@ static int updateSlotMapAsync(valkeyClusterAsyncContext *acc,
/* Send a command depending of config */
int status;
- if (acc->cc->flags & VALKEYCLUSTER_FLAG_ROUTE_USE_SLOTS) {
- status = valkeyAsyncCommand(ac, clusterSlotsReplyCallback, acc,
- VALKEY_COMMAND_CLUSTER_SLOTS);
- } else {
+ if (acc->cc->flags & VALKEY_FLAG_USE_CLUSTER_NODES) {
status = valkeyAsyncCommand(ac, clusterNodesReplyCallback, acc,
VALKEY_COMMAND_CLUSTER_NODES);
+ } else {
+ status = valkeyAsyncCommand(ac, clusterSlotsReplyCallback, acc,
+ VALKEY_COMMAND_CLUSTER_SLOTS);
}
if (status == VALKEY_OK) {
@@ -3059,7 +3018,7 @@ static void valkeyClusterAsyncCallback(valkeyAsyncContext *ac, void *r,
}
/* Skip retry handling when not expected, or during a client disconnect. */
- if (cad->retry_count == NO_RETRY || cc->flags & VALKEYCLUSTER_FLAG_DISCONNECTING)
+ if (cad->retry_count == NO_RETRY || cc->flags & VALKEY_FLAG_DISCONNECTING)
goto done;
error_type = cluster_reply_error_type(reply);
@@ -3172,7 +3131,7 @@ int valkeyClusterAsyncFormattedCommand(valkeyClusterAsyncContext *acc,
cc = acc->cc;
/* Don't accept new commands when the client is about to disconnect. */
- if (cc->flags & VALKEYCLUSTER_FLAG_DISCONNECTING) {
+ if (cc->flags & VALKEY_FLAG_DISCONNECTING) {
valkeyClusterAsyncSetError(acc, VALKEY_ERR_OTHER, "disconnecting");
return VALKEY_ERR;
}
@@ -3253,7 +3212,7 @@ int valkeyClusterAsyncFormattedCommandToNode(valkeyClusterAsyncContext *acc,
struct cmd *command = NULL;
/* Don't accept new commands when the client is about to disconnect. */
- if (cc->flags & VALKEYCLUSTER_FLAG_DISCONNECTING) {
+ if (cc->flags & VALKEY_FLAG_DISCONNECTING) {
valkeyClusterAsyncSetError(acc, VALKEY_ERR_OTHER, "disconnecting");
return VALKEY_ERR;
}
@@ -3435,7 +3394,7 @@ void valkeyClusterAsyncDisconnect(valkeyClusterAsyncContext *acc) {
}
cc = acc->cc;
- cc->flags |= VALKEYCLUSTER_FLAG_DISCONNECTING;
+ cc->flags |= VALKEY_FLAG_DISCONNECTING;
dictIterator di;
dictInitIterator(&di, cc->nodes);
@@ -3458,7 +3417,7 @@ void valkeyClusterAsyncFree(valkeyClusterAsyncContext *acc) {
return;
valkeyClusterContext *cc = acc->cc;
- cc->flags |= VALKEYCLUSTER_FLAG_DISCONNECTING;
+ cc->flags |= VALKEY_FLAG_DISCONNECTING;
valkeyClusterFree(cc);
vk_free(acc);
diff --git a/src/cluster_tls.c b/src/cluster_tls.c
deleted file mode 100644
index 7a0df24b..00000000
--- a/src/cluster_tls.c
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Copyright (c) 2022, Bjorn Svensson
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are met:
- *
- * * Redistributions of source code must retain the above copyright notice,
- * this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above copyright
- * notice, this list of conditions and the following disclaimer in the
- * documentation and/or other materials provided with the distribution.
- * * Neither the name of Redis nor the names of its contributors may be used
- * to endorse or promote products derived from this software without
- * specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
- * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
- * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
- * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
- * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
- * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
- * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
- * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
- * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
- * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
- * POSSIBILITY OF SUCH DAMAGE.
- */
-#include "cluster_tls.h"
-
-int valkeyClusterSetOptionEnableTLS(valkeyClusterContext *cc,
- valkeyTLSContext *tls) {
- if (cc == NULL || tls == NULL) {
- return VALKEY_ERR;
- }
-
- cc->tls = tls;
- cc->tls_init_fn = &valkeyInitiateTLSWithContext;
-
- return VALKEY_OK;
-}
diff --git a/tests/clusterclient.c b/tests/clusterclient.c
index 29b9bbf8..97517db2 100644
--- a/tests/clusterclient.c
+++ b/tests/clusterclient.c
@@ -63,7 +63,7 @@ void eventCallback(const valkeyClusterContext *cc, int event, void *privdata) {
int main(int argc, char **argv) {
int show_events = 0;
- int use_cluster_slots = 1;
+ int use_cluster_nodes = 0;
int send_to_all = 0;
int argindex;
@@ -72,7 +72,7 @@ int main(int argc, char **argv) {
if (strcmp(argv[argindex], "--events") == 0) {
show_events = 1;
} else if (strcmp(argv[argindex], "--use-cluster-nodes") == 0) {
- use_cluster_slots = 0;
+ use_cluster_nodes = 1;
} else {
fprintf(stderr, "Unknown argument: '%s'\n", argv[argindex]);
exit(1);
@@ -88,18 +88,19 @@ int main(int argc, char **argv) {
struct timeval timeout = {1, 500000}; // 1.5s
- valkeyClusterContext *cc = valkeyClusterContextInit();
- valkeyClusterSetOptionAddNodes(cc, initnode);
- valkeyClusterSetOptionConnectTimeout(cc, timeout);
- if (use_cluster_slots) {
- valkeyClusterSetOptionRouteUseSlots(cc);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = initnode;
+ options.connect_timeout = &timeout;
+ if (use_cluster_nodes) {
+ options.options = VALKEY_OPT_USE_CLUSTER_NODES;
}
if (show_events) {
- valkeyClusterSetEventCallback(cc, eventCallback, NULL);
+ options.event_callback = eventCallback;
}
- if (valkeyClusterConnect2(cc) != VALKEY_OK) {
- printf("Connect error: %s\n", cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ if (cc == NULL || cc->err) {
+ printf("Connect error: %s\n", cc ? cc->errstr : "OOM");
exit(2);
}
diff --git a/tests/clusterclient_async.c b/tests/clusterclient_async.c
index e81c5fee..84c4c322 100644
--- a/tests/clusterclient_async.c
+++ b/tests/clusterclient_async.c
@@ -237,13 +237,13 @@ void disconnectCallback(const valkeyAsyncContext *ac, int status) {
}
int main(int argc, char **argv) {
- int use_cluster_slots = 1; // Get topology via CLUSTER SLOTS
+ int use_cluster_nodes = 0;
int show_connection_events = 0;
int optind;
for (optind = 1; optind < argc && argv[optind][0] == '-'; optind++) {
if (strcmp(argv[optind], "--use-cluster-nodes") == 0) {
- use_cluster_slots = 0; // Use the default CLUSTER NODES instead
+ use_cluster_nodes = 1;
} else if (strcmp(argv[optind], "--events") == 0) {
show_events = 1;
} else if (strcmp(argv[optind], "--connection-events") == 0) {
@@ -262,36 +262,34 @@ int main(int argc, char **argv) {
}
const char *initnode = argv[optind];
struct timeval timeout = {0, 500000};
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterSetOptionAddNodes(acc->cc, initnode);
- valkeyClusterSetOptionTimeout(acc->cc, timeout);
- valkeyClusterSetOptionConnectTimeout(acc->cc, timeout);
- valkeyClusterSetOptionMaxRetry(acc->cc, 1);
- valkeyClusterSetEventCallback(acc->cc, eventCallback, acc);
- if (use_cluster_slots) {
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = initnode;
+ options.connect_timeout = &timeout;
+ options.command_timeout = &timeout;
+ options.max_retry = 1;
+ if (!async_initial_update) {
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ }
+ if (use_cluster_nodes) {
+ options.options |= VALKEY_OPT_USE_CLUSTER_NODES;
}
if (show_connection_events) {
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
}
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- int status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
-
- if (async_initial_update) {
- if (valkeyClusterAsyncConnect2(acc) != VALKEY_OK) {
- printf("Connect error: %s\n", acc->errstr);
- exit(2);
- }
- } else {
- if (valkeyClusterConnect2(acc->cc) != VALKEY_OK) {
- printf("Connect error: %s\n", acc->cc->errstr);
- exit(2);
- }
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit(&options);
+ if (acc == NULL || acc->err != 0) {
+ printf("Initiation failure: %s\n", acc ? acc->errstr : "OOM");
+ exit(2);
+ }
+ valkeyClusterAsyncSetEventCallback(acc, eventCallback, acc);
+ if (valkeyClusterAsyncConnect(acc) != VALKEY_OK) {
+ printf("Connect error: %s\n", acc->errstr);
+ exit(2);
}
event_base_dispatch(base);
diff --git a/tests/clusterclient_reconnect_async.c b/tests/clusterclient_reconnect_async.c
index 05e973bf..6af5451f 100644
--- a/tests/clusterclient_reconnect_async.c
+++ b/tests/clusterclient_reconnect_async.c
@@ -27,16 +27,15 @@ void connectToValkey(valkeyClusterAsyncContext *acc) {
/* reset context in case of reconnect */
valkeyClusterAsyncDisconnect(acc);
- int status = valkeyClusterConnect2(acc->cc);
- if (status == VALKEY_OK) {
+ if (valkeyClusterAsyncConnect(acc) == VALKEY_OK) {
// cluster mode
- } else if (acc->cc->err &&
- strcmp(acc->cc->errstr, VALKEY_ENOCLUSTER) == 0) {
+ } else if (acc->err &&
+ strcmp(acc->errstr, VALKEY_ENOCLUSTER) == 0) {
printf("[no cluster]\n");
- acc->cc->err = 0;
- memset(acc->cc->errstr, '\0', strlen(acc->cc->errstr));
+ acc->err = 0;
+ memset(acc->errstr, '\0', strlen(acc->errstr));
} else {
- printf("Connect error: %s\n", acc->cc->errstr);
+ printf("Connect error: %s\n", acc->errstr);
exit(-1);
}
}
@@ -95,15 +94,15 @@ int main(int argc, char **argv) {
exit(1);
}
const char *initnode = argv[1];
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterSetOptionAddNodes(acc->cc, initnode);
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = initnode;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- int status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit(&options);
+ assert(acc);
connectToValkey(acc);
// schedule reading from stdin and sending next command
diff --git a/tests/ct_async.c b/tests/ct_async.c
index 6148fdb3..4aff38c5 100644
--- a/tests/ct_async.c
+++ b/tests/ct_async.c
@@ -60,32 +60,22 @@ void eventCallback(const valkeyClusterContext *cc, int event, void *privdata) {
}
int main(void) {
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
-
- int status;
- status = valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- assert(status == VALKEY_OK);
- status = valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- assert(status == VALKEY_ERR); /* Re-registration not accepted */
-
- status = valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- assert(status == VALKEY_OK);
- status = valkeyClusterSetEventCallback(acc->cc, eventCallback, acc);
- assert(status == VALKEY_OK);
- status = valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- assert(status == VALKEY_OK);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- /* Expect error when connecting without an attached event library. */
- status = valkeyClusterAsyncConnect2(acc);
- assert(status == VALKEY_ERR);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit(&options);
+ assert(acc);
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
+ /* Set an event callback that uses acc as privdata */
+ int status = valkeyClusterAsyncSetEventCallback(acc, eventCallback, acc);
assert(status == VALKEY_OK);
- status = valkeyClusterAsyncConnect2(acc);
+ status = valkeyClusterAsyncConnect(acc);
assert(status == VALKEY_OK);
event_base_dispatch(base);
diff --git a/tests/ct_async_glib.c b/tests/ct_async_glib.c
index e6d305f3..34c5c89b 100644
--- a/tests/ct_async_glib.c
+++ b/tests/ct_async_glib.c
@@ -41,17 +41,18 @@ int main(int argc, char **argv) {
GMainContext *context = NULL;
mainloop = g_main_loop_new(context, FALSE);
- valkeyClusterAsyncContext *acc =
- valkeyClusterAsyncConnect(CLUSTER_NODE, VALKEYCLUSTER_FLAG_NULL);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseGlib(&options, context);
+
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
assert(acc);
ASSERT_MSG(acc->err == 0, acc->errstr);
- int status = valkeyClusterGlibAttach(acc, context);
- assert(status == VALKEY_OK);
-
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
-
+ int status;
status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"id", "SET key value");
ASSERT_MSG(status == VALKEY_OK, acc->errstr);
diff --git a/tests/ct_async_libev.c b/tests/ct_async_libev.c
index 292df507..0a89562f 100644
--- a/tests/ct_async_libev.c
+++ b/tests/ct_async_libev.c
@@ -35,18 +35,18 @@ int main(int argc, char **argv) {
UNUSED(argc);
UNUSED(argv);
- valkeyClusterAsyncContext *acc =
- valkeyClusterAsyncConnect(CLUSTER_NODE, VALKEYCLUSTER_FLAG_NULL);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibev(&options, EV_DEFAULT);
+
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
assert(acc);
ASSERT_MSG(acc->err == 0, acc->errstr);
int status;
- status = valkeyClusterLibevAttach(acc, EV_DEFAULT);
- assert(status == VALKEY_OK);
-
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
-
status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"ID",
"SET key value");
ASSERT_MSG(status == VALKEY_OK, acc->errstr);
diff --git a/tests/ct_async_libuv.c b/tests/ct_async_libuv.c
index b2771a1c..8bb22d7c 100644
--- a/tests/ct_async_libuv.c
+++ b/tests/ct_async_libuv.c
@@ -36,19 +36,20 @@ int main(int argc, char **argv) {
UNUSED(argc);
UNUSED(argv);
- valkeyClusterAsyncContext *acc =
- valkeyClusterAsyncConnect(CLUSTER_NODE, VALKEYCLUSTER_FLAG_NULL);
+ uv_loop_t *loop = uv_default_loop();
+
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibuv(&options, loop);
+
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
assert(acc);
ASSERT_MSG(acc->err == 0, acc->errstr);
int status;
- uv_loop_t *loop = uv_default_loop();
- status = valkeyClusterLibuvAttach(acc, loop);
- assert(status == VALKEY_OK);
-
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
-
status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"ID",
"SET key value");
ASSERT_MSG(status == VALKEY_OK, acc->errstr);
diff --git a/tests/ct_commands.c b/tests/ct_commands.c
index 2d462bf7..cf3e5e35 100644
--- a/tests/ct_commands.c
+++ b/tests/ct_commands.c
@@ -462,15 +462,12 @@ void test_multi(valkeyClusterContext *cc) {
int main(void) {
struct timeval timeout = {0, 500000};
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.connect_timeout = &timeout;
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
- valkeyClusterSetOptionConnectTimeout(cc, timeout);
-
- int status;
- status = valkeyClusterConnect2(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
load_valkey_version(cc);
test_bitfield(cc);
diff --git a/tests/ct_connection.c b/tests/ct_connection.c
index 7dff73e7..b257203c 100644
--- a/tests/ct_connection.c
+++ b/tests/ct_connection.c
@@ -29,16 +29,14 @@ void reset_counters(void) {
// Connecting to a password protected cluster and
// providing a correct password.
void test_password_ok(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionPassword(cc, CLUSTER_PASSWORD);
- valkeyClusterSetConnectCallback(cc, connect_callback);
-
- int status;
- status = valkeyClusterConnect2(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
- assert(connect_success_counter == 1); // for CLUSTER NODES
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.password = CLUSTER_PASSWORD;
+ options.connect_callback = connect_callback;
+
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
+ assert(connect_success_counter == 1); // for CLUSTER SLOTS
load_valkey_version(cc);
assert(connect_success_counter == 2); // for checking valkey version
@@ -58,14 +56,12 @@ void test_password_ok(void) {
// Connecting to a password protected cluster and
// providing wrong password.
void test_password_wrong(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionPassword(cc, "faultypass");
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.password = "faultypass";
- int status;
- status = valkeyClusterConnect2(cc);
- assert(status == VALKEY_ERR);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ assert(cc);
assert(cc->err == VALKEY_ERR_OTHER);
if (valkey_version_less_than(6, 0))
@@ -79,14 +75,12 @@ void test_password_wrong(void) {
// Connecting to a password protected cluster and
// not providing any password.
void test_password_missing(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE_WITH_PASSWORD);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ // No password set.
- // A password is not configured..
- int status;
- status = valkeyClusterConnect2(cc);
- assert(status == VALKEY_ERR);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ assert(cc);
assert(cc->err == VALKEY_ERR_OTHER);
assert(strncmp(cc->errstr, "NOAUTH", 6) == 0);
@@ -101,50 +95,13 @@ void test_username_ok(void) {
return;
// Connect to the cluster using username and password
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionUsername(cc, CLUSTER_USERNAME);
- valkeyClusterSetOptionPassword(cc, CLUSTER_PASSWORD);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.username = CLUSTER_USERNAME;
+ options.password = CLUSTER_PASSWORD;
- int ret = valkeyClusterConnect2(cc);
- ASSERT_MSG(ret == VALKEY_OK, cc->errstr);
-
- // Test connection
- valkeyReply *reply = valkeyClusterCommand(cc, "SET key1 Hello");
- CHECK_REPLY_OK(cc, reply);
- freeReplyObject(reply);
-
- valkeyClusterFree(cc);
-}
-
-// Test of disabling the use of username after it was enabled.
-void test_username_disabled(void) {
- if (valkey_version_less_than(6, 0))
- return;
-
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionUsername(cc, "missing-user");
- valkeyClusterSetOptionPassword(cc, CLUSTER_PASSWORD);
-
- // Connect using 'AUTH ' should fail
- int ret = valkeyClusterConnect2(cc);
- assert(ret == VALKEY_ERR);
- assert(cc->err == VALKEY_ERR_OTHER);
- assert(strncmp(cc->errstr, "WRONGPASS invalid username-password pair",
- 40) == 0);
-
- // Disable use of username (2 alternatives)
- ret = valkeyClusterSetOptionUsername(cc, NULL);
- ASSERT_MSG(ret == VALKEY_OK, cc->errstr);
- ret = valkeyClusterSetOptionUsername(cc, "");
- ASSERT_MSG(ret == VALKEY_OK, cc->errstr);
-
- // Connect using 'AUTH ' should pass
- ret = valkeyClusterConnect2(cc);
- ASSERT_MSG(ret == VALKEY_OK, cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
// Test connection
valkeyReply *reply = valkeyClusterCommand(cc, "SET key1 Hello");
@@ -156,23 +113,20 @@ void test_username_disabled(void) {
// Connect and handle two clusters simultaneously
void test_multicluster(void) {
- int ret;
valkeyReply *reply;
// Connect to first cluster
- valkeyClusterContext *cc1 = valkeyClusterContextInit();
+ valkeyClusterContext *cc1 = valkeyClusterConnect(CLUSTER_NODE);
assert(cc1);
- valkeyClusterSetOptionAddNodes(cc1, CLUSTER_NODE);
- ret = valkeyClusterConnect2(cc1);
- ASSERT_MSG(ret == VALKEY_OK, cc1->errstr);
+ ASSERT_MSG(cc1->err == 0, cc1->errstr);
// Connect to second cluster
- valkeyClusterContext *cc2 = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.password = CLUSTER_PASSWORD;
+ valkeyClusterContext *cc2 = valkeyClusterConnectWithOptions(&options);
assert(cc2);
- valkeyClusterSetOptionAddNodes(cc2, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionPassword(cc2, CLUSTER_PASSWORD);
- ret = valkeyClusterConnect2(cc2);
- ASSERT_MSG(ret == VALKEY_OK, cc2->errstr);
+ ASSERT_MSG(cc2->err == 0, cc2->errstr);
// Set keys differently in clusters
reply = valkeyClusterCommand(cc1, "SET key Hello1");
@@ -207,16 +161,14 @@ void test_multicluster(void) {
void test_connect_timeout(void) {
struct timeval timeout = {0, 200000};
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
-
/* Configure a non-routable IP address and a timeout */
- valkeyClusterSetOptionAddNodes(cc, "192.168.0.0:7000");
- valkeyClusterSetOptionConnectTimeout(cc, timeout);
- valkeyClusterSetConnectCallback(cc, connect_callback);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = "192.168.0.0:7000";
+ options.connect_timeout = &timeout;
+ options.connect_callback = connect_callback;
- int status = valkeyClusterConnect2(cc);
- assert(status == VALKEY_ERR);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ assert(cc);
assert(cc->err == VALKEY_ERR_IO);
assert(strcmp(cc->errstr, "Connection timed out") == 0);
assert(connect_success_counter == 0);
@@ -230,13 +182,12 @@ void test_connect_timeout(void) {
void test_command_timeout(void) {
struct timeval timeout = {0, 10000};
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
- valkeyClusterSetOptionTimeout(cc, timeout);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.command_timeout = &timeout;
- int status = valkeyClusterConnect2(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
valkeyClusterNodeIterator ni;
valkeyClusterInitNodeIterator(&ni, cc);
@@ -263,14 +214,8 @@ void test_command_timeout(void) {
/* Connect and configure a command timeout while connected. */
void test_command_timeout_set_while_connected(void) {
- struct timeval timeout = {0, 10000};
-
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
-
- int status = valkeyClusterConnect2(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnect(CLUSTER_NODE);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
valkeyClusterNodeIterator ni;
valkeyClusterInitNodeIterator(&ni, cc);
@@ -283,6 +228,7 @@ void test_command_timeout_set_while_connected(void) {
freeReplyObject(reply);
/* Set command timeout while connected */
+ struct timeval timeout = {0, 10000};
valkeyClusterSetOptionTimeout(cc, timeout);
reply = valkeyClusterCommandToNode(cc, node, "DEBUG SLEEP 0.2");
@@ -347,26 +293,23 @@ void commandCallback(valkeyClusterAsyncContext *cc, void *r, void *privdata) {
// Connecting to a password protected cluster using
// the async API, providing correct password.
void test_async_password_ok(void) {
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionPassword(acc->cc, CLUSTER_PASSWORD);
-
struct event_base *base = event_base_new();
- valkeyClusterLibeventAttach(acc, base);
- int ret;
- ret = valkeyClusterConnect2(acc->cc);
- assert(ret == VALKEY_OK);
- assert(acc->err == 0);
- assert(acc->cc->err == 0);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.password = CLUSTER_PASSWORD;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
+
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
// Test connection
ExpectedResult r = {
.type = VALKEY_REPLY_STATUS, .str = "OK", .disconnect = true};
- ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
+ int ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
assert(ret == VALKEY_OK);
event_base_dispatch(base);
@@ -378,16 +321,17 @@ void test_async_password_ok(void) {
/* Connect to a password protected cluster using the wrong password.
An eventloop is not attached since it is not needed is this case. */
void test_async_password_wrong(void) {
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionPassword(acc->cc, "faultypass");
+ struct event_base *base = event_base_new();
- int ret;
- ret = valkeyClusterConnect2(acc->cc);
- assert(ret == VALKEY_ERR);
- assert(acc->err == VALKEY_OK); // TODO: This must be wrong!
- assert(acc->cc->err == VALKEY_ERR_OTHER);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.password = "faultypass";
+ valkeyClusterOptionsUseLibevent(&options, base);
+
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ assert(acc);
+ assert(acc->err == VALKEY_ERR_OTHER);
if (valkey_version_less_than(6, 0))
assert(strcmp(acc->cc->errstr, "ERR invalid password") == 0);
else
@@ -395,79 +339,80 @@ void test_async_password_wrong(void) {
// No connection
ExpectedResult r;
- ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
+ int ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
assert(ret == VALKEY_ERR);
assert(acc->err == VALKEY_ERR_OTHER);
assert(strcmp(acc->errstr, "slotmap not available") == 0);
valkeyClusterAsyncFree(acc);
+ event_base_free(base);
}
/* Connect to a password protected cluster without providing a password.
An eventloop is not attached since it is not needed is this case. */
void test_async_password_missing(void) {
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE_WITH_PASSWORD);
+ struct event_base *base = event_base_new();
+
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
// Password not configured
- int ret;
- ret = valkeyClusterConnect2(acc->cc);
- assert(ret == VALKEY_ERR);
- assert(acc->err == VALKEY_OK); // TODO: This must be wrong!
- assert(acc->cc->err == VALKEY_ERR_OTHER);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ assert(acc);
+ assert(acc->err == VALKEY_ERR_OTHER);
assert(strncmp(acc->cc->errstr, "NOAUTH", 6) == 0);
// No connection
ExpectedResult r;
- ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
+ int ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
assert(ret == VALKEY_ERR);
assert(acc->err == VALKEY_ERR_OTHER);
assert(strcmp(acc->errstr, "slotmap not available") == 0);
valkeyClusterAsyncFree(acc);
+ event_base_free(base);
}
// Connect to a cluster and authenticate using username and password
void test_async_username_ok(void) {
if (valkey_version_less_than(6, 0))
return;
+ struct event_base *base = event_base_new();
// Connect to the cluster using username and password
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionUsername(acc->cc, "missing-user");
- valkeyClusterSetOptionPassword(acc->cc, CLUSTER_PASSWORD);
-
- struct event_base *base = event_base_new();
- valkeyClusterLibeventAttach(acc, base);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ options.username = "missing-user";
+ options.password = CLUSTER_PASSWORD;
+ valkeyClusterOptionsUseLibevent(&options, base);
// Connect using wrong username should fail
- int ret = valkeyClusterConnect2(acc->cc);
- assert(ret == VALKEY_ERR);
- assert(acc->cc->err == VALKEY_ERR_OTHER);
- assert(strncmp(acc->cc->errstr, "WRONGPASS invalid username-password pair",
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ assert(acc);
+ assert(acc->err == VALKEY_ERR_OTHER);
+ assert(strncmp(acc->errstr, "WRONGPASS invalid username-password pair",
40) == 0);
+ valkeyClusterAsyncFree(acc);
// Set correct username
- ret = valkeyClusterSetOptionUsername(acc->cc, CLUSTER_USERNAME);
- ASSERT_MSG(ret == VALKEY_OK, acc->cc->errstr);
+ options.username = CLUSTER_USERNAME;
// Connect using correct username should pass
- ret = valkeyClusterConnect2(acc->cc);
- assert(ret == VALKEY_OK);
+ acc = valkeyClusterAsyncConnectWithOptions(&options);
+ assert(acc);
assert(acc->err == 0);
- assert(acc->cc->err == 0);
// Test connection
ExpectedResult r = {
.type = VALKEY_REPLY_STATUS, .str = "OK", .disconnect = true};
- ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
+ int ret = valkeyClusterAsyncCommand(acc, commandCallback, &r, "SET key1 Hello");
assert(ret == VALKEY_OK);
event_base_dispatch(base);
@@ -478,40 +423,34 @@ void test_async_username_ok(void) {
// Connect and handle two clusters simultaneously using the async API
void test_async_multicluster(void) {
- int ret;
-
- valkeyClusterAsyncContext *acc1 = valkeyClusterAsyncContextInit();
- assert(acc1);
- valkeyClusterAsyncSetConnectCallback(acc1, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc1, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc1->cc, CLUSTER_NODE);
-
- valkeyClusterAsyncContext *acc2 = valkeyClusterAsyncContextInit();
- assert(acc2);
- valkeyClusterAsyncSetConnectCallback(acc2, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc2, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc2->cc, CLUSTER_NODE_WITH_PASSWORD);
- valkeyClusterSetOptionPassword(acc2->cc, CLUSTER_PASSWORD);
-
struct event_base *base = event_base_new();
- valkeyClusterLibeventAttach(acc1, base);
- valkeyClusterLibeventAttach(acc2, base);
+
+ valkeyClusterOptions options1 = {0};
+ options1.initial_nodes = CLUSTER_NODE;
+ options1.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options1.async_connect_callback = connectCallback;
+ options1.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options1, base);
// Connect to first cluster
- ret = valkeyClusterConnect2(acc1->cc);
- assert(ret == VALKEY_OK);
- assert(acc1->err == 0);
- assert(acc1->cc->err == 0);
+ valkeyClusterAsyncContext *acc1 = valkeyClusterAsyncConnectWithOptions(&options1);
+ ASSERT_MSG(acc1 && acc1->err == 0, acc1 ? acc1->errstr : "OOM");
+
+ valkeyClusterOptions options2 = {0};
+ options2.initial_nodes = CLUSTER_NODE_WITH_PASSWORD;
+ options2.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options2.password = CLUSTER_PASSWORD;
+ options2.async_connect_callback = connectCallback;
+ options2.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options2, base);
// Connect to second cluster
- ret = valkeyClusterConnect2(acc2->cc);
- assert(ret == VALKEY_OK);
- assert(acc2->err == 0);
- assert(acc2->cc->err == 0);
+ valkeyClusterAsyncContext *acc2 = valkeyClusterAsyncConnectWithOptions(&options2);
+ ASSERT_MSG(acc2 && acc2->err == 0, acc2 ? acc2->errstr : "OOM");
// Set keys differently in clusters
ExpectedResult r1 = {.type = VALKEY_REPLY_STATUS, .str = "OK"};
- ret = valkeyClusterAsyncCommand(acc1, commandCallback, &r1, "SET key A");
+ int ret = valkeyClusterAsyncCommand(acc1, commandCallback, &r1, "SET key A");
assert(ret == VALKEY_OK);
ExpectedResult r2 = {.type = VALKEY_REPLY_STATUS, .str = "OK"};
@@ -544,22 +483,20 @@ void test_async_multicluster(void) {
/* Connect to a non-routable address which results in a connection timeout. */
void test_async_connect_timeout(void) {
+ struct event_base *base = event_base_new();
struct timeval timeout = {0, 200000};
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
-
+ valkeyClusterOptions options = {0};
/* Configure a non-routable IP address and a timeout */
- valkeyClusterSetOptionAddNodes(acc->cc, "192.168.0.0:7000");
- valkeyClusterSetOptionConnectTimeout(acc->cc, timeout);
+ options.initial_nodes = "192.168.0.0:7000";
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.connect_timeout = &timeout;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- valkeyClusterLibeventAttach(acc, base);
-
- int status = valkeyClusterConnect2(acc->cc);
- assert(status == VALKEY_ERR);
- assert(acc->cc->err == VALKEY_ERR_IO);
- assert(strcmp(acc->cc->errstr, "Connection timed out") == 0);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ assert(acc);
+ assert(acc->err == VALKEY_ERR_IO);
+ assert(strcmp(acc->errstr, "Connection timed out") == 0);
event_base_dispatch(base);
@@ -569,19 +506,17 @@ void test_async_connect_timeout(void) {
/* Connect using a pre-configured command timeout */
void test_async_command_timeout(void) {
+ struct event_base *base = event_base_new();
struct timeval timeout = {0, 10000};
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- valkeyClusterSetOptionTimeout(acc->cc, timeout);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.command_timeout = &timeout;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- valkeyClusterLibeventAttach(acc, base);
-
- int status = valkeyClusterConnect2(acc->cc);
- assert(status == VALKEY_OK);
- assert(acc->cc->err == 0);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
valkeyClusterNodeIterator ni;
valkeyClusterInitNodeIterator(&ni, acc->cc);
@@ -591,8 +526,8 @@ void test_async_command_timeout(void) {
/* Simulate a command timeout and expect a timeout error */
ExpectedResult r = {
.noreply = true, .errstr = "Timeout", .disconnect = true};
- status = valkeyClusterAsyncCommandToNode(acc, node, commandCallback, &r,
- "DEBUG SLEEP 0.2");
+ int status = valkeyClusterAsyncCommandToNode(acc, node, commandCallback, &r,
+ "DEBUG SLEEP 0.2");
assert(status == VALKEY_OK);
event_base_dispatch(base);
@@ -607,7 +542,6 @@ int main(void) {
test_password_wrong();
test_password_missing();
test_username_ok();
- test_username_disabled();
test_multicluster();
test_connect_timeout();
test_command_timeout();
diff --git a/tests/ct_connection_ipv6.c b/tests/ct_connection_ipv6.c
index aeee635b..02723c1f 100644
--- a/tests/ct_connection_ipv6.c
+++ b/tests/ct_connection_ipv6.c
@@ -10,23 +10,14 @@
// Successful connection an IPv6 cluster
void test_successful_ipv6_connection(void) {
-
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
-
- int status;
struct timeval timeout = {0, 500000}; // 0.5s
- status = valkeyClusterSetOptionConnectTimeout(cc, timeout);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
-
- status = valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE_IPV6);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
- status = valkeyClusterSetOptionRouteUseSlots(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE_IPV6;
+ options.connect_timeout = &timeout;
- status = valkeyClusterConnect2(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
valkeyReply *reply;
reply = (valkeyReply *)valkeyClusterCommand(cc, "SET key_ipv6 value");
diff --git a/tests/ct_out_of_memory_handling.c b/tests/ct_out_of_memory_handling.c
index 3d7de08c..ad5992f7 100644
--- a/tests/ct_out_of_memory_handling.c
+++ b/tests/ct_out_of_memory_handling.c
@@ -114,75 +114,35 @@ void test_alloc_failure_handling(void) {
// Override allocators
valkeySetAllocators(&ha);
- // Context init
+ struct timeval timeout = {0, 500000};
+
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_USE_REPLICAS;
+ options.connect_timeout = &timeout;
+ options.command_timeout = &timeout;
+
+ // Connect
valkeyClusterContext *cc;
{
for (int i = 0; i < 3; ++i) {
successfulAllocations = i;
- cc = valkeyClusterContextInit();
+ cc = valkeyClusterConnectWithOptions(&options);
assert(cc == NULL);
}
- successfulAllocations = 3;
- cc = valkeyClusterContextInit();
- assert(cc);
- }
- cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
-
- // Add nodes
- {
- for (int i = 0; i < 9; ++i) {
- prepare_allocation_test(cc, i);
- result = valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
- assert(result == VALKEY_ERR);
- ASSERT_STR_EQ(cc->errstr, "Out of memory");
- }
-
- prepare_allocation_test(cc, 9);
- result = valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
- assert(result == VALKEY_OK);
- }
-
- // Set connect timeout
- {
- struct timeval timeout = {0, 500000};
-
- prepare_allocation_test(cc, 0);
- result = valkeyClusterSetOptionConnectTimeout(cc, timeout);
- assert(result == VALKEY_ERR);
- ASSERT_STR_EQ(cc->errstr, "Out of memory");
-
- prepare_allocation_test(cc, 1);
- result = valkeyClusterSetOptionConnectTimeout(cc, timeout);
- assert(result == VALKEY_OK);
- }
-
- // Set request timeout
- {
- struct timeval timeout = {0, 500000};
-
- prepare_allocation_test(cc, 0);
- result = valkeyClusterSetOptionTimeout(cc, timeout);
- assert(result == VALKEY_ERR);
- ASSERT_STR_EQ(cc->errstr, "Out of memory");
-
- prepare_allocation_test(cc, 1);
- result = valkeyClusterSetOptionTimeout(cc, timeout);
- assert(result == VALKEY_OK);
- }
-
- // Connect
- {
- for (int i = 0; i < 88; ++i) {
- prepare_allocation_test(cc, i);
- result = valkeyClusterConnect2(cc);
- assert(result == VALKEY_ERR);
+ for (int i = 3; i < 100; ++i) {
+ successfulAllocations = i;
+ cc = valkeyClusterConnectWithOptions(&options);
+ assert(cc);
ASSERT_STR_EQ(cc->errstr, "Out of memory");
+ valkeyClusterFree(cc);
}
+ // Skip iteration 100 to 159 since sdscatfmt give leak warnings during OOM.
- prepare_allocation_test(cc, 88);
- result = valkeyClusterConnect2(cc);
- assert(result == VALKEY_OK);
+ successfulAllocations = 160;
+ cc = valkeyClusterConnectWithOptions(&options);
+ assert(cc && cc->err == 0);
}
// Command
@@ -488,64 +448,40 @@ void test_alloc_failure_handling_async(void) {
// Override allocators
valkeySetAllocators(&ha);
- // Context init
- valkeyClusterAsyncContext *acc;
- {
- for (int i = 0; i < 4; ++i) {
- successfulAllocations = 0;
- acc = valkeyClusterAsyncContextInit();
- assert(acc == NULL);
- }
- successfulAllocations = 4;
- acc = valkeyClusterAsyncContextInit();
- assert(acc);
- }
- acc->cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
+ struct event_base *base = event_base_new();
+ assert(base);
- // Set callbacks
- {
- prepare_allocation_test_async(acc, 0);
- result = valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- assert(result == VALKEY_OK);
- result = valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- assert(result == VALKEY_OK);
- }
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_USE_REPLICAS |
+ VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- // Add nodes
+ // Connect
+ valkeyClusterAsyncContext *acc;
{
- for (int i = 0; i < 9; ++i) {
- prepare_allocation_test(acc->cc, i);
- result = valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- assert(result == VALKEY_ERR);
- ASSERT_STR_EQ(acc->cc->errstr, "Out of memory");
+ for (int i = 0; i < 13; ++i) {
+ successfulAllocations = i;
+ acc = valkeyClusterAsyncConnectWithOptions(&options);
+ assert(acc == NULL);
}
- prepare_allocation_test(acc->cc, 9);
- result = valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- assert(result == VALKEY_OK);
- }
-
- // Connect
- {
- for (int i = 0; i < 86; ++i) {
- prepare_allocation_test(acc->cc, i);
- result = valkeyClusterConnect2(acc->cc);
- assert(result == VALKEY_ERR);
- ASSERT_STR_EQ(acc->cc->errstr, "Out of memory");
+ for (int i = 13; i < 100; ++i) {
+ successfulAllocations = i;
+ acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_STR_EQ(acc->errstr, "Out of memory");
+ assert(acc != NULL);
+ valkeyClusterAsyncFree(acc);
}
+ // Skip iteration 100 to 156 since sdscatfmt give leak warnings during OOM.
- prepare_allocation_test(acc->cc, 86);
- result = valkeyClusterConnect2(acc->cc);
- assert(result == VALKEY_OK);
+ successfulAllocations = 157;
+ acc = valkeyClusterAsyncConnectWithOptions(&options);
+ assert(acc && acc->err == 0);
}
- struct event_base *base = event_base_new();
- assert(base);
-
- successfulAllocations = 0;
- result = valkeyClusterLibeventAttach(acc, base);
- assert(result == VALKEY_OK);
-
// Async command 1
ExpectedResult r1 = {.type = VALKEY_REPLY_STATUS, .str = "OK"};
{
diff --git a/tests/ct_pipeline.c b/tests/ct_pipeline.c
index 5fe8cc54..7d55a7c8 100644
--- a/tests/ct_pipeline.c
+++ b/tests/ct_pipeline.c
@@ -12,16 +12,13 @@
// Test of two pipelines using sync API
void test_pipeline(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
- int status;
- status = valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
-
- status = valkeyClusterConnect2(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
+ int status;
status = valkeyClusterAppendCommand(cc, "SET foo one");
ASSERT_MSG(status == VALKEY_OK, cc->errstr);
status = valkeyClusterAppendCommand(cc, "SET bar two");
@@ -95,20 +92,19 @@ void commandCallback(valkeyClusterAsyncContext *cc, void *r, void *privdata) {
// nature of an event loop. Therefore, unlike the synchronous API, there is only
// a single way to send commands.
void test_async_pipeline(void) {
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
+ struct event_base *base = event_base_new();
- int status;
- status = valkeyClusterConnect2(acc->cc);
- ASSERT_MSG(status == VALKEY_OK, acc->errstr);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
+ int status;
ExpectedResult r1 = {.type = VALKEY_REPLY_STATUS, .str = "OK"};
status =
valkeyClusterAsyncCommand(acc, commandCallback, &r1, "SET foo six");
diff --git a/tests/ct_specific_nodes.c b/tests/ct_specific_nodes.c
index 8793cb41..8e1063c9 100644
--- a/tests/ct_specific_nodes.c
+++ b/tests/ct_specific_nodes.c
@@ -318,27 +318,25 @@ void commandCallback(valkeyClusterAsyncContext *cc, void *r, void *privdata) {
}
void test_async_to_single_node(void) {
- int status;
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- valkeyClusterSetOptionMaxRetry(acc->cc, 1);
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
- status = valkeyClusterConnect2(acc->cc);
- ASSERT_MSG(status == VALKEY_OK, acc->errstr);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.max_retry = 1;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
valkeyClusterNodeIterator ni;
valkeyClusterInitNodeIterator(&ni, acc->cc);
valkeyClusterNode *node = valkeyClusterNodeNext(&ni);
assert(node);
+ int status;
ExpectedResult r1 = {.type = VALKEY_REPLY_INTEGER, .disconnect = true};
status = valkeyClusterAsyncCommandToNode(acc, node, commandCallback, &r1,
"DBSIZE");
@@ -351,27 +349,25 @@ void test_async_to_single_node(void) {
}
void test_async_formatted_to_single_node(void) {
- int status;
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- valkeyClusterSetOptionMaxRetry(acc->cc, 1);
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
- status = valkeyClusterConnect2(acc->cc);
- ASSERT_MSG(status == VALKEY_OK, acc->errstr);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.max_retry = 1;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
valkeyClusterNodeIterator ni;
valkeyClusterInitNodeIterator(&ni, acc->cc);
valkeyClusterNode *node = valkeyClusterNodeNext(&ni);
assert(node);
+ int status;
ExpectedResult r1 = {.type = VALKEY_REPLY_INTEGER, .disconnect = true};
char command[] = "*1\r\n$6\r\nDBSIZE\r\n";
status = valkeyClusterAsyncFormattedCommandToNode(
@@ -385,27 +381,25 @@ void test_async_formatted_to_single_node(void) {
}
void test_async_command_argv_to_single_node(void) {
- int status;
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- valkeyClusterSetOptionMaxRetry(acc->cc, 1);
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
- status = valkeyClusterConnect2(acc->cc);
- ASSERT_MSG(status == VALKEY_OK, acc->errstr);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.max_retry = 1;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
valkeyClusterNodeIterator ni;
valkeyClusterInitNodeIterator(&ni, acc->cc);
valkeyClusterNode *node = valkeyClusterNodeNext(&ni);
assert(node);
+ int status;
ExpectedResult r1 = {.type = VALKEY_REPLY_INTEGER, .disconnect = true};
status = valkeyClusterAsyncCommandArgvToNode(
acc, node, commandCallback, &r1, 1, (const char *[]){"DBSIZE"},
@@ -419,25 +413,23 @@ void test_async_command_argv_to_single_node(void) {
}
void test_async_to_all_nodes(void) {
- int status;
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- valkeyClusterSetOptionMaxRetry(acc->cc, 1);
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
- status = valkeyClusterConnect2(acc->cc);
- ASSERT_MSG(status == VALKEY_OK, acc->errstr);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.max_retry = 1;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
valkeyClusterNodeIterator ni;
valkeyClusterInitNodeIterator(&ni, acc->cc);
+ int status;
ExpectedResult r1 = {.type = VALKEY_REPLY_INTEGER};
valkeyClusterNode *node;
@@ -461,25 +453,23 @@ void test_async_to_all_nodes(void) {
}
void test_async_transaction(void) {
- int status;
+ struct event_base *base = event_base_new();
- valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit();
- assert(acc);
- valkeyClusterAsyncSetConnectCallback(acc, connectCallback);
- valkeyClusterAsyncSetDisconnectCallback(acc, disconnectCallback);
- valkeyClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
- valkeyClusterSetOptionMaxRetry(acc->cc, 1);
- valkeyClusterSetOptionRouteUseSlots(acc->cc);
- status = valkeyClusterConnect2(acc->cc);
- ASSERT_MSG(status == VALKEY_OK, acc->errstr);
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.options = VALKEY_OPT_BLOCKING_INITIAL_UPDATE;
+ options.max_retry = 1;
+ options.async_connect_callback = connectCallback;
+ options.async_disconnect_callback = disconnectCallback;
+ valkeyClusterOptionsUseLibevent(&options, base);
- struct event_base *base = event_base_new();
- status = valkeyClusterLibeventAttach(acc, base);
- assert(status == VALKEY_OK);
+ valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
+ ASSERT_MSG(acc && acc->err == 0, acc ? acc->errstr : "OOM");
valkeyClusterNode *node = valkeyClusterGetNodeByKey(acc->cc, (char *)"foo");
assert(node);
+ int status;
ExpectedResult r1 = {.type = VALKEY_REPLY_STATUS, .str = "OK"};
status = valkeyClusterAsyncCommandToNode(acc, node, commandCallback, &r1,
"MULTI");
@@ -509,15 +499,12 @@ void test_async_transaction(void) {
}
int main(void) {
- int status;
+ valkeyClusterOptions options = {0};
+ options.initial_nodes = CLUSTER_NODE;
+ options.max_retry = 1;
- valkeyClusterContext *cc = valkeyClusterContextInit();
- assert(cc);
- valkeyClusterSetOptionAddNodes(cc, CLUSTER_NODE);
- valkeyClusterSetOptionRouteUseSlots(cc);
- valkeyClusterSetOptionMaxRetry(cc, 1);
- status = valkeyClusterConnect2(cc);
- ASSERT_MSG(status == VALKEY_OK, cc->errstr);
+ valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
+ ASSERT_MSG(cc && cc->err == 0, cc ? cc->errstr : "OOM");
load_valkey_version(cc);
// Synchronous API
diff --git a/tests/ut_slotmap_update.c b/tests/ut_slotmap_update.c
index ad6931c5..f0b05556 100644
--- a/tests/ut_slotmap_update.c
+++ b/tests/ut_slotmap_update.c
@@ -123,15 +123,16 @@ char *resp_encode_array(char *p, sds *resp) {
/* Parse a cluster nodes reply from a basic deployment. */
void test_parse_cluster_nodes(bool parse_replicas) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ if (parse_replicas)
+ options.options |= VALKEY_OPT_USE_REPLICAS;
+
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
cluster_slot *slot;
dictIterator di;
- if (parse_replicas)
- cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
-
valkeyReply *reply = create_cluster_nodes_reply(
"07c37dfeb235213a872192d90877d0cd55635b91 127.0.0.1:30004@31004,hostname4 slave e7d1eecce10fd6bb5eb35b9f99a514335d9ba9ca 0 1426238317239 4 connected\n"
"67ed2db8d677e59ec4a4cefb06858cf2a1a89fa1 127.0.0.1:30002@31002,hostname2 master - 0 1426238316232 2 connected 5461-10922\n"
@@ -209,7 +210,8 @@ void test_parse_cluster_nodes(bool parse_replicas) {
}
void test_parse_cluster_nodes_during_failover(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
cluster_slot *slot;
@@ -274,7 +276,8 @@ void test_parse_cluster_nodes_during_failover(void) {
/* Skip nodes with the `noaddr` flag. */
void test_parse_cluster_nodes_with_noaddr(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
dictIterator di;
@@ -302,7 +305,8 @@ void test_parse_cluster_nodes_with_noaddr(void) {
}
void test_parse_cluster_nodes_with_empty_ip(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyClusterNode *node;
dictIterator di;
@@ -337,7 +341,8 @@ void test_parse_cluster_nodes_with_empty_ip(void) {
/* Parse replies with additional importing and migrating information. */
void test_parse_cluster_nodes_with_special_slot_entries(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
cluster_slot *slot;
@@ -377,15 +382,16 @@ void test_parse_cluster_nodes_with_special_slot_entries(void) {
/* Parse a cluster nodes reply containing a primary with multiple replicas. */
void test_parse_cluster_nodes_with_multiple_replicas(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ options.options |= VALKEY_OPT_USE_REPLICAS;
+
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
cluster_slot *slot;
dictIterator di;
listIter li;
- cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
-
valkeyReply *reply = create_cluster_nodes_reply(
"07c37dfeb235213a872192d90877d0cd55635b91 127.0.0.1:30004@31004,hostname4 slave e7d1eecce10fd6bb5eb35b9f99a514335d9ba9ca 0 1426238317239 4 connected\n"
"6ec23923021cf3ffec47632106199cb7f496ce01 127.0.0.1:30005@31005,hostname5 slave e7d1eecce10fd6bb5eb35b9f99a514335d9ba9ca 0 1426238316232 5 connected\n"
@@ -442,7 +448,8 @@ void test_parse_cluster_nodes_with_multiple_replicas(void) {
/* Give error when parsing erroneous data. */
void test_parse_cluster_nodes_with_parse_error(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyReply *reply;
dict *nodes;
@@ -490,7 +497,8 @@ void test_parse_cluster_nodes_with_parse_error(void) {
/* Redis pre-v4.0 returned node addresses without the clusterbus port,
* i.e. `ip:port` instead of `ip:port@cport` */
void test_parse_cluster_nodes_with_legacy_format(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
dictIterator di;
@@ -514,15 +522,16 @@ void test_parse_cluster_nodes_with_legacy_format(void) {
/* Parse a cluster slots reply from a basic deployment. */
void test_parse_cluster_slots(bool parse_replicas) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ if (parse_replicas)
+ options.options |= VALKEY_OPT_USE_REPLICAS;
+
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
cluster_slot *slot;
dictIterator di;
- if (parse_replicas)
- cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
-
valkeyReply *reply = create_cluster_slots_reply(
"[[0, 5460, ['127.0.0.1', 30001, 'e7d1eecce10fd6bb5eb35b9f99a514335d9ba9ca', ['hostname', 'localhost']],"
" ['127.0.0.1', 30004, '07c37dfeb235213a872192d90877d0cd55635b91', ['hostname', 'localhost']]],"
@@ -598,7 +607,8 @@ void test_parse_cluster_slots(bool parse_replicas) {
}
void test_parse_cluster_slots_with_empty_ip(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyClusterNode *node;
dictIterator di;
@@ -633,7 +643,8 @@ void test_parse_cluster_slots_with_empty_ip(void) {
}
void test_parse_cluster_slots_with_null_ip(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyClusterNode *node;
dictIterator di;
@@ -669,15 +680,16 @@ void test_parse_cluster_slots_with_null_ip(void) {
/* Parse a cluster slots reply containing a primary with multiple replicas. */
void test_parse_cluster_slots_with_multiple_replicas(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ options.options |= VALKEY_OPT_USE_REPLICAS;
+
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
cluster_slot *slot;
dictIterator di;
listIter li;
- cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
-
valkeyReply *reply = create_cluster_slots_reply(
"[[0, 16383, ['127.0.0.1', 30001, 'e7d1eecce10fd6bb5eb35b9f99a514335d9ba9ca'],"
" ['127.0.0.1', 30004, '07c37dfeb235213a872192d90877d0cd55635b91'],"
@@ -728,15 +740,16 @@ void test_parse_cluster_slots_with_multiple_replicas(void) {
}
void test_parse_cluster_slots_with_noncontiguous_slots(void) {
- valkeyClusterContext *cc = valkeyClusterContextInit();
+ valkeyClusterOptions options = {0};
+ options.options |= VALKEY_OPT_USE_REPLICAS;
+
+ valkeyClusterContext *cc = valkeyClusterContextInit(&options);
valkeyContext *c = valkeyContextInit();
valkeyClusterNode *node;
cluster_slot *slot;
dictIterator di;
listIter li;
- cc->flags |= VALKEYCLUSTER_FLAG_ADD_SLAVE;
-
valkeyReply *reply = create_cluster_slots_reply(
"[[0, 0, ['127.0.0.1', 30001, 'e7d1eecce10fd6bb5eb35b9f99a514335d9ba9ca'],"
" ['127.0.0.1', 30004, '07c37dfeb235213a872192d90877d0cd55635b91']],"