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

valkey-cli auto-exit from subscribed mode #1432

Merged
merged 20 commits into from
Jan 8, 2025
Merged

valkey-cli auto-exit from subscribed mode #1432

Changes from 1 commit
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
c21bf9d
Fixed issue with exiting subscribe mode
Nikhil-Manglore Dec 12, 2024
cb71718
fixed issue with the counters when exit subscribe mode
Nikhil-Manglore Dec 18, 2024
926a713
refactored code in valkey-cli.cand fixed the test cases
Nikhil-Manglore Dec 19, 2024
2278fce
fixed typo in the comments
Nikhil-Manglore Dec 19, 2024
0c4a177
Refactored code again and added new tests to the test case
Nikhil-Manglore Dec 19, 2024
398c482
Merge branch 'valkey-io:unstable' into subscribe-mode-bug
Nikhil-Manglore Dec 20, 2024
a047778
Merge branch 'valkey-io:unstable' into subscribe-mode-bug
Nikhil-Manglore Dec 26, 2024
791a80c
Merge branch 'valkey-io:unstable' into subscribe-mode-bug
Nikhil-Manglore Jan 2, 2025
d6c863e
Merge branch 'valkey-io:unstable' into subscribe-mode-bug
Nikhil-Manglore Jan 3, 2025
79aff57
Fixed all issues with subscribe mode, refactored code, and finished i…
Nikhil-Manglore Jan 3, 2025
6743126
Merge branch 'subscribe-mode-bug' of github.com:Nikhil-Manglore/valke…
Nikhil-Manglore Jan 3, 2025
b09ef32
Refactored code, added punsubscribe functionality, and made test case…
Nikhil-Manglore Jan 6, 2025
930b4db
cleaned the code up to make it more readable
Nikhil-Manglore Jan 6, 2025
1849195
Fixed corner cases, added test case and split up existing test cases
Nikhil-Manglore Jan 6, 2025
9f99a51
Refactored code
Nikhil-Manglore Jan 7, 2025
022b2cc
Removed redundant code
Nikhil-Manglore Jan 7, 2025
fe091cf
Removed redundant code
Nikhil-Manglore Jan 8, 2025
9aff4d8
removed if statement and updated comments
Nikhil-Manglore Jan 8, 2025
6aebdb9
Fixed formatting errors
Nikhil-Manglore Jan 8, 2025
fcc2742
Fixed final formatting error
Nikhil-Manglore Jan 8, 2025
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
38 changes: 18 additions & 20 deletions src/valkey-cli.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2233,20 +2233,26 @@ static int cliReadReply(int output_raw_strings) {
}

/* Helper method to process unsubscribe responses */
zuiderkwast marked this conversation as resolved.
Show resolved Hide resolved
static void processUnsubscribeReply(redisReply *reply) {
static void handlePubSubMode(redisReply *reply) {
if (reply->elements >= 3) {
zuiderkwast marked this conversation as resolved.
Show resolved Hide resolved
char *cmd = reply->element[0]->str;
int count = reply->element[2]->integer;

if (strcmp(cmd, "unsubscribe") == 0 || strcmp(cmd, "punsubscribe") == 0) {
/* Update counts based on the command type */
if (strcmp(cmd, "subscribe") == 0 || strcmp(cmd, "psubscribe") == 0 ||
strcmp(cmd, "unsubscribe") == 0 || strcmp(cmd, "punsubscribe") == 0) {
config.pubsub_unsharded_count = count;
} else if (strcmp(cmd, "sunsubscribe") == 0) {
} else if (strcmp(cmd, "ssubscribe") == 0 || strcmp(cmd, "sunsubscribe") == 0) {
config.pubsub_sharded_count = count;
}

if (config.pubsub_unsharded_count == 0 && config.pubsub_sharded_count == 0) {
/* Update pubsub mode based on the current counts */
if (config.pubsub_unsharded_count == 0 && config.pubsub_sharded_count == 0 && config.pubsub_mode) {
zuiderkwast marked this conversation as resolved.
Show resolved Hide resolved
config.pubsub_mode = 0;
cliRefreshPrompt();
} else if (config.pubsub_unsharded_count + config.pubsub_sharded_count > 0 && !config.pubsub_mode) {
config.pubsub_mode = 1;
cliRefreshPrompt();
}
}
}
Expand All @@ -2270,7 +2276,7 @@ static void cliWaitForMessagesOrStdin(void) {
fflush(stdout);

if (isPubsubPush(reply)) {
processUnsubscribeReply(reply);
handlePubSubMode(reply);
}

sdsfree(out);
Expand Down Expand Up @@ -2428,22 +2434,14 @@ static int cliSendCommand(int argc, char **argv, long repeat) {
/* Handle pubsub mode */
zuiderkwast marked this conversation as resolved.
Show resolved Hide resolved
if (config.last_reply->elements >= 3) {
char *cmd = config.last_reply->element[0]->str;
int count = config.last_reply->element[2]->integer;


/* If it's an unsubscribe command, call the helper */
if (strcmp(cmd, "unsubscribe") == 0 || strcmp(cmd, "punsubscribe") == 0 || strcmp(cmd, "sunsubscribe") == 0) {
if (config.pubsub_mode && isPubsubPush(config.last_reply)) {
processUnsubscribeReply(config.last_reply);
}
}

/* Handle subscribe commands */
else if (strcmp(cmd, "subscribe") == 0 || strcmp(cmd, "psubscribe") == 0) {
config.pubsub_unsharded_count = count;
} else if (strcmp(cmd, "ssubscribe") == 0) {
config.pubsub_sharded_count = count;

/* Call the helper to handle both subscribe and unsubscribe commands */
if (strcmp(cmd, "subscribe") == 0 || strcmp(cmd, "psubscribe") == 0 ||
strcmp(cmd, "ssubscribe") == 0 || strcmp(cmd, "unsubscribe") == 0 ||
strcmp(cmd, "punsubscribe") == 0 || strcmp(cmd, "sunsubscribe") == 0) {
handlePubSubMode(config.last_reply);
}
zuiderkwast marked this conversation as resolved.
Show resolved Hide resolved

}

if (num_expected_pubsub_push > 0 && !strcasecmp(config.last_reply->element[0]->str, command)) {
Expand Down