Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Introduce configuration options in the cluster API #137

Draft
wants to merge 5 commits into
base: main
Choose a base branch
from
Draft

Introduce configuration options in the cluster API #137

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1bbc930
Remove flags from cluster connect APIs
bjosv Sep 13, 2024
b15117f
Add cluster connect with options
bjosv Oct 16, 2024
8d75d64
Add async cluster connect with options
bjosv Oct 16, 2024
9007335
Remove config functions that are replaced with options
bjosv Oct 20, 2024
8f0e516
New option: Blocking slotmap updates after initial async connect
bjosv Dec 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 15 additions & 17 deletions examples/cluster-async-tls.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,26 +66,24 @@ 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);
exit(-1);
}
valkeyClusterOptions options = {0};
options.initial_nodes = CLUSTER_NODE_TLS;
options.onConnect = connectCallback;
options.onDisconnect = disconnectCallback;
options.options = VALKEY_OPT_USE_CLUSTER_SLOTS;
VALKEY_CLUSTER_OPTIONS_SET_SSL(&options, tls);

struct event_base *base = event_base_new();
valkeyClusterLibeventAttach(acc, base);
VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBEVENT(&options, base);

valkeyClusterAsyncContext *acc = valkeyClusterAsyncConnectWithOptions(&options);
if (acc == NULL || acc->err != 0) {
printf("Error: %s\n", acc ? acc->errstr : "OOM");
exit(-1);
}

int status;
status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"THE_ID",
"SET %s %s", "key", "value");
int status = valkeyClusterAsyncCommand(acc, setCallback, (char *)"THE_ID",
"SET %s %s", "key", "value");
if (status != VALKEY_OK) {
printf("error: err=%d errstr=%s\n", acc->err, acc->errstr);
}
Expand Down
40 changes: 21 additions & 19 deletions examples/cluster-async.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.onConnect = connectCallback;
options.onDisconnect = disconnectCallback;
VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBEVENT(&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;
}
29 changes: 12 additions & 17 deletions examples/cluster-clientside-caching-async.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,11 +127,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);
Expand All @@ -143,24 +140,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.onConnectNC = connectCallbackNC;
options.onDisconnect = disconnectCallback;
VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBEVENT(&options, base);

valkeyClusterAsyncContext *acc = valkeyClusterAsyncContextInit(&options);
assert(acc);

int status;
status = valkeyClusterAsyncSetConnectCallbackNC(acc, connectCallbackNC);
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);
assert(status == VALKEY_OK);

status = valkeyClusterAsyncConnect2(acc);
status = valkeyClusterAsyncConnect(acc);
assert(status == VALKEY_OK);

event_base_dispatch(base);
Expand Down
11 changes: 6 additions & 5 deletions examples/cluster-simple.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,12 @@ 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;
options.options = VALKEY_OPT_USE_CLUSTER_SLOTS;

valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
if (!cc) {
printf("Error: Allocation failure\n");
exit(-1);
Expand Down
14 changes: 7 additions & 7 deletions examples/cluster-tls.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,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.options = VALKEY_OPT_USE_CLUSTER_SLOTS;
VALKEY_CLUSTER_OPTIONS_SET_SSL(&options, tls);

valkeyClusterContext *cc = valkeyClusterConnectWithOptions(&options);
if (!cc) {
printf("Error: Allocation failure\n");
exit(-1);
Expand Down
8 changes: 8 additions & 0 deletions include/valkey/adapters/ae.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,4 +150,12 @@ static int valkeyClusterAeAttach(valkeyClusterAsyncContext *acc,
acc->attach_data = loop;
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_AE(opts, loop) \
do { \
(opts)->attach_fn = valkeyAeAttachAdapter; \
(opts)->attach_data = loop; \
} while (0)

#endif /* VALKEY_ADAPTERS_AE_H */
7 changes: 7 additions & 0 deletions include/valkey/adapters/glib.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,4 +164,11 @@ static int valkeyClusterGlibAttach(valkeyClusterAsyncContext *acc,
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_GLIB(opts, context) \
do { \
(opts)->attach_fn = valkeyGlibAttachAdapter; \
(opts)->attach_data = context; \
} while (0)

#endif /* VALKEY_ADAPTERS_GLIB_H */
10 changes: 8 additions & 2 deletions include/valkey/adapters/ivykis.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ 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);
}

Expand All @@ -95,8 +95,14 @@ static int valkeyClusterIvykisAttach(valkeyClusterAsyncContext *acc) {
return VALKEY_ERR;
}

acc->attach_fn = valkeyClusterIvykisAttachAdapter;
acc->attach_fn = valkeyIvykisAttachAdapter;
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_IVYKIS(opts) \
do { \
(opts)->attach_fn = valkeyAeAttachAdapter; \
} while (0)

#endif /* VALKEY_ADAPTERS_IVYKIS_H */
7 changes: 7 additions & 0 deletions include/valkey/adapters/libev.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,4 +204,11 @@ static int valkeyClusterLibevAttach(valkeyClusterAsyncContext *acc,
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBEV(opts, loop) \
do { \
(opts)->attach_fn = valkeyLibevAttachAdapter; \
(opts)->attach_data = loop; \
} while (0)

#endif /* VALKEY_ADAPTERS_LIBEV_H */
8 changes: 8 additions & 0 deletions include/valkey/adapters/libevent.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,4 +192,12 @@ static int valkeyClusterLibeventAttach(valkeyClusterAsyncContext *acc,
acc->attach_data = base;
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBEVENT(opts, event_base) \
do { \
(opts)->attach_fn = valkeyLibeventAttachAdapter; \
(opts)->attach_data = event_base; \
} while (0)

Comment on lines +196 to +202
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are these not static functions, like valkeyClusterLibeventAttach and others?

#endif /* VALKEY_ADAPTERS_LIBEVENT_H */
7 changes: 7 additions & 0 deletions include/valkey/adapters/libhv.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,4 +140,11 @@ static int valkeyClusterLibhvAttach(valkeyClusterAsyncContext *acc,
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBHV(opts, loop) \
do { \
(opts)->attach_fn = valkeyLibhvAttachAdapter; \
(opts)->attach_data = loop; \
} while (0)

#endif /* VALKEY_ADAPTERS_LIBHV_H */
7 changes: 7 additions & 0 deletions include/valkey/adapters/libsdevent.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,4 +195,11 @@ static int valkeyClusterLibsdeventAttach(valkeyClusterAsyncContext *acc,
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBSDEVENT(opts, event) \
do { \
(opts)->attach_fn = valkeyLibsdeventAttachAdapter; \
(opts)->attach_data = event; \
} while (0)

#endif /* VALKEY_ADAPTERS_LIBSDEVENT_H */
8 changes: 8 additions & 0 deletions include/valkey/adapters/libuv.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,4 +213,12 @@ static int valkeyClusterLibuvAttach(valkeyClusterAsyncContext *acc,
acc->attach_data = loop;
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_LIBUV(opts, loop) \
do { \
(opts)->attach_fn = valkeyLibuvAttachAdapter; \
(opts)->attach_data = loop; \
} while (0)

#endif /* VALKEY_ADAPTERS_LIBUV_H */
7 changes: 7 additions & 0 deletions include/valkey/adapters/macosx.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,4 +160,11 @@ static int valkeyClusterMacOSAttach(valkeyClusterAsyncContext *acc,
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_MACOSX(opts, loop) \
do { \
(opts)->attach_fn = valkeyMacOSAttachAdapter; \
(opts)->attach_data = loop; \
} while (0)

#endif /* VALKEY_ADAPTERS_MACOSX_H */
6 changes: 6 additions & 0 deletions include/valkey/adapters/poll.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,4 +211,10 @@ static int valkeyClusterPollAttach(valkeyClusterAsyncContext *acc) {
return VALKEY_OK;
}

/* Helper macro to initialize options. */
#define VALKEY_CLUSTER_OPTIONS_SET_ADAPTER_POLL(opts) \
do { \
(opts)->attach_fn = valkeyPollAttachAdapter; \
} while (0)

#endif /* VALKEY_ADAPTERS_POLL_H */
Loading
Loading