redict/tests/modules/blockonkeys.c

// SPDX-FileCopyrightText: 2024 Redict Contributors
// SPDX-FileCopyrightText: 2024 Salvatore Sanfilippo <antirez at gmail dot com>
//
// SPDX-License-Identifier: BSD-3-Clause
// SPDX-License-Identifier: LGPL-3.0-only

#include "redictmodule.h"

#include <string.h>
#include <strings.h>
#include <assert.h>
#include <unistd.h>

#define UNUSED(V) ((void) V)

#define LIST_SIZE 1024

/* The FSL (Fixed-Size List) data type is a low-budget imitation of the
 * native Redict list, in order to test list-like commands implemented
 * by a module.
 * Examples: FSL.PUSH, FSL.BPOP, etc. */

typedef struct {
    long long list[LIST_SIZE];
    long long length;
} fsl_t; /* Fixed-size list */

static RedictModuleType *fsltype = NULL;

fsl_t *fsl_type_create(void) {
    fsl_t *o;
    o = RedictModule_Alloc(sizeof(*o));
    o->length = 0;
    return o;
}

void fsl_type_free(fsl_t *o) {
    RedictModule_Free(o);
}

/* ========================== "fsltype" type methods ======================= */

void *fsl_rdb_load(RedictModuleIO *rdb, int encver) {
    if (encver != 0) {
        return NULL;
    }
    fsl_t *fsl = fsl_type_create();
    fsl->length = RedictModule_LoadUnsigned(rdb);
    for (long long i = 0; i < fsl->length; i++)
        fsl->list[i] = RedictModule_LoadSigned(rdb);
    return fsl;
}

void fsl_rdb_save(RedictModuleIO *rdb, void *value) {
    fsl_t *fsl = value;
    RedictModule_SaveUnsigned(rdb,fsl->length);
    for (long long i = 0; i < fsl->length; i++)
        RedictModule_SaveSigned(rdb, fsl->list[i]);
}

void fsl_aofrw(RedictModuleIO *aof, RedictModuleString *key, void *value) {
    fsl_t *fsl = value;
    for (long long i = 0; i < fsl->length; i++)
        RedictModule_EmitAOF(aof, "FSL.PUSH","sl", key, fsl->list[i]);
}

void fsl_free(void *value) {
    fsl_type_free(value);
}

/* ========================== helper methods ======================= */

/* Wrapper to the boilerplate code of opening a key, checking its type, etc.
 * Returns 0 if `keyname` exists in the dataset, but it's of the wrong type (i.e. not FSL) */
int get_fsl(RedictModuleCtx *ctx, RedictModuleString *keyname, int mode, int create, fsl_t **fsl, int reply_on_failure) {
    *fsl = NULL;
    RedictModuleKey *key = RedictModule_OpenKey(ctx, keyname, mode);

    if (RedictModule_KeyType(key) != REDICTMODULE_KEYTYPE_EMPTY) {
        /* Key exists */
        if (RedictModule_ModuleTypeGetType(key) != fsltype) {
            /* Key is not FSL */
            RedictModule_CloseKey(key);
            if (reply_on_failure)
                RedictModule_ReplyWithError(ctx, REDICTMODULE_ERRORMSG_WRONGTYPE);
            RedictModuleCallReply *reply = RedictModule_Call(ctx, "INCR", "c", "fsl_wrong_type");
            RedictModule_FreeCallReply(reply);
            return 0;
        }

        *fsl = RedictModule_ModuleTypeGetValue(key);
        if (*fsl && !(*fsl)->length && mode & REDICTMODULE_WRITE) {
            /* Key exists, but it's logically empty */
            if (create) {
                create = 0; /* No need to create, key exists in its basic state */
            } else {
                RedictModule_DeleteKey(key);
                *fsl = NULL;
            }
        } else {
            /* Key exists, and has elements in it - no need to create anything */
            create = 0;
        }
    }

    if (create) {
        *fsl = fsl_type_create();
        RedictModule_ModuleTypeSetValue(key, fsltype, *fsl);
    }

    RedictModule_CloseKey(key);
    return 1;
}

/* ========================== commands ======================= */

/* FSL.PUSH <key> <int> - Push an integer to the fixed-size list (to the right).
 * It must be greater than the element in the head of the list. */
int fsl_push(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc != 3)
        return RedictModule_WrongArity(ctx);

    long long ele;
    if (RedictModule_StringToLongLong(argv[2],&ele) != REDICTMODULE_OK)
        return RedictModule_ReplyWithError(ctx,"ERR invalid integer");

    fsl_t *fsl;
    if (!get_fsl(ctx, argv[1], REDICTMODULE_WRITE, 1, &fsl, 1))
        return REDICTMODULE_OK;

    if (fsl->length == LIST_SIZE)
        return RedictModule_ReplyWithError(ctx,"ERR list is full");

    if (fsl->length != 0 && fsl->list[fsl->length-1] >= ele)
        return RedictModule_ReplyWithError(ctx,"ERR new element has to be greater than the head element");

    fsl->list[fsl->length++] = ele;
    RedictModule_SignalKeyAsReady(ctx, argv[1]);

    RedictModule_ReplicateVerbatim(ctx);

    return RedictModule_ReplyWithSimpleString(ctx, "OK");
}

typedef struct {
    RedictModuleString *keyname;
    long long ele;
} timer_data_t;

static void timer_callback(RedictModuleCtx *ctx, void *data)
{
    timer_data_t *td = data;

    fsl_t *fsl;
    if (!get_fsl(ctx, td->keyname, REDICTMODULE_WRITE, 1, &fsl, 1))
        return;

    if (fsl->length == LIST_SIZE)
        return; /* list is full */

    if (fsl->length != 0 && fsl->list[fsl->length-1] >= td->ele)
        return; /* new element has to be greater than the head element */

    fsl->list[fsl->length++] = td->ele;
    RedictModule_SignalKeyAsReady(ctx, td->keyname);

    RedictModule_Replicate(ctx, "FSL.PUSH", "sl", td->keyname, td->ele);

    RedictModule_FreeString(ctx, td->keyname);
    RedictModule_Free(td);
}

/* FSL.PUSHTIMER <key> <int> <period-in-ms> - Push the number 9000 to the fixed-size list (to the right).
 * It must be greater than the element in the head of the list. */
int fsl_pushtimer(RedictModuleCtx *ctx, RedictModuleString **argv, int argc)
{
    if (argc != 4)
        return RedictModule_WrongArity(ctx);

    long long ele;
    if (RedictModule_StringToLongLong(argv[2],&ele) != REDICTMODULE_OK)
        return RedictModule_ReplyWithError(ctx,"ERR invalid integer");

    long long period;
    if (RedictModule_StringToLongLong(argv[3],&period) != REDICTMODULE_OK)
        return RedictModule_ReplyWithError(ctx,"ERR invalid period");

    fsl_t *fsl;
    if (!get_fsl(ctx, argv[1], REDICTMODULE_WRITE, 1, &fsl, 1))
        return REDICTMODULE_OK;

    if (fsl->length == LIST_SIZE)
        return RedictModule_ReplyWithError(ctx,"ERR list is full");

    timer_data_t *td = RedictModule_Alloc(sizeof(*td));
    td->keyname = argv[1];
    RedictModule_RetainString(ctx, td->keyname);
    td->ele = ele;

    RedictModuleTimerID id = RedictModule_CreateTimer(ctx, period, timer_callback, td);
    RedictModule_ReplyWithLongLong(ctx, id);

    return REDICTMODULE_OK;
}

int bpop_reply_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    RedictModuleString *keyname = RedictModule_GetBlockedClientReadyKey(ctx);

    fsl_t *fsl;
    if (!get_fsl(ctx, keyname, REDICTMODULE_WRITE, 0, &fsl, 0) || !fsl)
        return REDICTMODULE_ERR;

    RedictModule_Assert(fsl->length);
    RedictModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);

    /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
    RedictModule_ReplicateVerbatim(ctx);
    return REDICTMODULE_OK;
}

int bpop_timeout_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    return RedictModule_ReplyWithSimpleString(ctx, "Request timedout");
}

/* FSL.BPOP <key> <timeout> [NO_TO_CB]- Block clients until list has two or more elements.
 * When that happens, unblock client and pop the last two elements (from the right). */
int fsl_bpop(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc < 3)
        return RedictModule_WrongArity(ctx);

    long long timeout;
    if (RedictModule_StringToLongLong(argv[2],&timeout) != REDICTMODULE_OK || timeout < 0)
        return RedictModule_ReplyWithError(ctx,"ERR invalid timeout");

    int to_cb = 1;
    if (argc == 4) {
        if (strcasecmp("NO_TO_CB", RedictModule_StringPtrLen(argv[3], NULL)))
            return RedictModule_ReplyWithError(ctx,"ERR invalid argument");
        to_cb = 0;
    }

    fsl_t *fsl;
    if (!get_fsl(ctx, argv[1], REDICTMODULE_WRITE, 0, &fsl, 1))
        return REDICTMODULE_OK;

    if (!fsl) {
        RedictModule_BlockClientOnKeys(ctx, bpop_reply_callback, to_cb ? bpop_timeout_callback : NULL,
                                      NULL, timeout, &argv[1], 1, NULL);
    } else {
        RedictModule_Assert(fsl->length);
        RedictModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
        /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
        RedictModule_ReplicateVerbatim(ctx);
    }

    return REDICTMODULE_OK;
}

int bpopgt_reply_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    RedictModuleString *keyname = RedictModule_GetBlockedClientReadyKey(ctx);
    long long *pgt = RedictModule_GetBlockedClientPrivateData(ctx);

    fsl_t *fsl;
    if (!get_fsl(ctx, keyname, REDICTMODULE_WRITE, 0, &fsl, 0) || !fsl)
        return RedictModule_ReplyWithError(ctx,"UNBLOCKED key no longer exists");

    if (fsl->list[fsl->length-1] <= *pgt)
        return REDICTMODULE_ERR;

    RedictModule_Assert(fsl->length);
    RedictModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
    /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
    RedictModule_ReplicateVerbatim(ctx);
    return REDICTMODULE_OK;
}

int bpopgt_timeout_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    return RedictModule_ReplyWithSimpleString(ctx, "Request timedout");
}

void bpopgt_free_privdata(RedictModuleCtx *ctx, void *privdata) {
    REDICTMODULE_NOT_USED(ctx);
    RedictModule_Free(privdata);
}

/* FSL.BPOPGT <key> <gt> <timeout> - Block clients until list has an element greater than <gt>.
 * When that happens, unblock client and pop the last element (from the right). */
int fsl_bpopgt(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc != 4)
        return RedictModule_WrongArity(ctx);

    long long gt;
    if (RedictModule_StringToLongLong(argv[2],&gt) != REDICTMODULE_OK)
        return RedictModule_ReplyWithError(ctx,"ERR invalid integer");

    long long timeout;
    if (RedictModule_StringToLongLong(argv[3],&timeout) != REDICTMODULE_OK || timeout < 0)
        return RedictModule_ReplyWithError(ctx,"ERR invalid timeout");

    fsl_t *fsl;
    if (!get_fsl(ctx, argv[1], REDICTMODULE_WRITE, 0, &fsl, 1))
        return REDICTMODULE_OK;

    if (!fsl)
        return RedictModule_ReplyWithError(ctx,"ERR key must exist");

    if (fsl->list[fsl->length-1] <= gt) {
        /* We use malloc so the tests in blockedonkeys.tcl can check for memory leaks */
        long long *pgt = RedictModule_Alloc(sizeof(long long));
        *pgt = gt;
        RedictModule_BlockClientOnKeysWithFlags(
            ctx, bpopgt_reply_callback, bpopgt_timeout_callback,
            bpopgt_free_privdata, timeout, &argv[1], 1, pgt,
            REDICTMODULE_BLOCK_UNBLOCK_DELETED);
    } else {
        RedictModule_Assert(fsl->length);
        RedictModule_ReplyWithLongLong(ctx, fsl->list[--fsl->length]);
        /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
        RedictModule_ReplicateVerbatim(ctx);
    }

    return REDICTMODULE_OK;
}

int bpoppush_reply_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    RedictModuleString *src_keyname = RedictModule_GetBlockedClientReadyKey(ctx);
    RedictModuleString *dst_keyname = RedictModule_GetBlockedClientPrivateData(ctx);

    fsl_t *src;
    if (!get_fsl(ctx, src_keyname, REDICTMODULE_WRITE, 0, &src, 0) || !src)
        return REDICTMODULE_ERR;

    fsl_t *dst;
    if (!get_fsl(ctx, dst_keyname, REDICTMODULE_WRITE, 1, &dst, 0) || !dst)
        return REDICTMODULE_ERR;

    RedictModule_Assert(src->length);
    long long ele = src->list[--src->length];
    dst->list[dst->length++] = ele;
    RedictModule_SignalKeyAsReady(ctx, dst_keyname);
    /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
    RedictModule_ReplicateVerbatim(ctx);
    return RedictModule_ReplyWithLongLong(ctx, ele);
}

int bpoppush_timeout_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    return RedictModule_ReplyWithSimpleString(ctx, "Request timedout");
}

void bpoppush_free_privdata(RedictModuleCtx *ctx, void *privdata) {
    RedictModule_FreeString(ctx, privdata);
}

/* FSL.BPOPPUSH <src> <dst> <timeout> - Block clients until <src> has an element.
 * When that happens, unblock client, pop the last element from <src> and push it to <dst>
 * (from the right). */
int fsl_bpoppush(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc != 4)
        return RedictModule_WrongArity(ctx);

    long long timeout;
    if (RedictModule_StringToLongLong(argv[3],&timeout) != REDICTMODULE_OK || timeout < 0)
        return RedictModule_ReplyWithError(ctx,"ERR invalid timeout");

    fsl_t *src;
    if (!get_fsl(ctx, argv[1], REDICTMODULE_WRITE, 0, &src, 1))
        return REDICTMODULE_OK;

    if (!src) {
        /* Retain string for reply callback */
        RedictModule_RetainString(ctx, argv[2]);
        /* Key is empty, we must block */
        RedictModule_BlockClientOnKeys(ctx, bpoppush_reply_callback, bpoppush_timeout_callback,
                                      bpoppush_free_privdata, timeout, &argv[1], 1, argv[2]);
    } else {
        fsl_t *dst;
        if (!get_fsl(ctx, argv[2], REDICTMODULE_WRITE, 1, &dst, 1))
            return REDICTMODULE_OK;

        RedictModule_Assert(src->length);
        long long ele = src->list[--src->length];
        dst->list[dst->length++] = ele;
        RedictModule_SignalKeyAsReady(ctx, argv[2]);
        RedictModule_ReplyWithLongLong(ctx, ele);
        /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
        RedictModule_ReplicateVerbatim(ctx);
    }

    return REDICTMODULE_OK;
}

/* FSL.GETALL <key> - Reply with an array containing all elements. */
int fsl_getall(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc != 2)
        return RedictModule_WrongArity(ctx);

    fsl_t *fsl;
    if (!get_fsl(ctx, argv[1], REDICTMODULE_READ, 0, &fsl, 1))
        return REDICTMODULE_OK;

    if (!fsl)
        return RedictModule_ReplyWithArray(ctx, 0);

    RedictModule_ReplyWithArray(ctx, fsl->length);
    for (int i = 0; i < fsl->length; i++)
        RedictModule_ReplyWithLongLong(ctx, fsl->list[i]);
    return REDICTMODULE_OK;
}

/* Callback for blockonkeys_popall */
int blockonkeys_popall_reply_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argc);
    RedictModuleKey *key = RedictModule_OpenKey(ctx, argv[1], REDICTMODULE_WRITE);
    if (RedictModule_KeyType(key) == REDICTMODULE_KEYTYPE_LIST) {
        RedictModuleString *elem;
        long len = 0;
        RedictModule_ReplyWithArray(ctx, REDICTMODULE_POSTPONED_ARRAY_LEN);
        while ((elem = RedictModule_ListPop(key, REDICTMODULE_LIST_HEAD)) != NULL) {
            len++;
            RedictModule_ReplyWithString(ctx, elem);
            RedictModule_FreeString(ctx, elem);
        }
        /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
        RedictModule_ReplicateVerbatim(ctx);
        RedictModule_ReplySetArrayLength(ctx, len);
    } else {
        RedictModule_ReplyWithError(ctx, "ERR Not a list");
    }
    RedictModule_CloseKey(key);
    return REDICTMODULE_OK;
}

int blockonkeys_popall_timeout_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    return RedictModule_ReplyWithError(ctx, "ERR Timeout");
}

/* BLOCKONKEYS.POPALL key
 *
 * Blocks on an empty key for up to 3 seconds. When unblocked by a list
 * operation like LPUSH, all the elements are popped and returned. Fails with an
 * error on timeout. */
int blockonkeys_popall(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc != 2)
        return RedictModule_WrongArity(ctx);

    RedictModuleKey *key = RedictModule_OpenKey(ctx, argv[1], REDICTMODULE_READ);
    if (RedictModule_KeyType(key) == REDICTMODULE_KEYTYPE_EMPTY) {
        RedictModule_BlockClientOnKeys(ctx, blockonkeys_popall_reply_callback,
                                      blockonkeys_popall_timeout_callback,
                                      NULL, 3000, &argv[1], 1, NULL);
    } else {
        RedictModule_ReplyWithError(ctx, "ERR Key not empty");
    }
    RedictModule_CloseKey(key);
    return REDICTMODULE_OK;
}

/* BLOCKONKEYS.LPUSH key val [val ..]
 * BLOCKONKEYS.LPUSH_UNBLOCK key val [val ..]
 *
 * A module equivalent of LPUSH. If the name LPUSH_UNBLOCK is used,
 * RM_SignalKeyAsReady() is also called. */
int blockonkeys_lpush(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc < 3)
        return RedictModule_WrongArity(ctx);

    RedictModuleKey *key = RedictModule_OpenKey(ctx, argv[1], REDICTMODULE_WRITE);
    if (RedictModule_KeyType(key) != REDICTMODULE_KEYTYPE_EMPTY &&
        RedictModule_KeyType(key) != REDICTMODULE_KEYTYPE_LIST) {
        RedictModule_ReplyWithError(ctx, REDICTMODULE_ERRORMSG_WRONGTYPE);
    } else {
        for (int i = 2; i < argc; i++) {
            if (RedictModule_ListPush(key, REDICTMODULE_LIST_HEAD,
                                     argv[i]) != REDICTMODULE_OK) {
                RedictModule_CloseKey(key);
                return RedictModule_ReplyWithError(ctx, "ERR Push failed");
            }
        }
    }
    RedictModule_CloseKey(key);

    /* signal key as ready if the command is lpush_unblock */
    size_t len;
    const char *str = RedictModule_StringPtrLen(argv[0], &len);
    if (!strncasecmp(str, "blockonkeys.lpush_unblock", len)) {
        RedictModule_SignalKeyAsReady(ctx, argv[1]);
    }
    RedictModule_ReplicateVerbatim(ctx);
    return RedictModule_ReplyWithSimpleString(ctx, "OK");
}

/* Callback for the BLOCKONKEYS.BLPOPN command */
int blockonkeys_blpopn_reply_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argc);
    long long n;
    RedictModule_StringToLongLong(argv[2], &n);
    RedictModuleKey *key = RedictModule_OpenKey(ctx, argv[1], REDICTMODULE_WRITE);
    int result;
    if (RedictModule_KeyType(key) == REDICTMODULE_KEYTYPE_LIST &&
        RedictModule_ValueLength(key) >= (size_t)n) {
        RedictModule_ReplyWithArray(ctx, n);
        for (long i = 0; i < n; i++) {
            RedictModuleString *elem = RedictModule_ListPop(key, REDICTMODULE_LIST_HEAD);
            RedictModule_ReplyWithString(ctx, elem);
            RedictModule_FreeString(ctx, elem);
        }
        /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
        RedictModule_ReplicateVerbatim(ctx);
        result = REDICTMODULE_OK;
    } else if (RedictModule_KeyType(key) == REDICTMODULE_KEYTYPE_LIST ||
               RedictModule_KeyType(key) == REDICTMODULE_KEYTYPE_EMPTY) {
        const char *module_cmd = RedictModule_StringPtrLen(argv[0], NULL);
        if (!strcasecmp(module_cmd, "blockonkeys.blpopn_or_unblock"))
            RedictModule_UnblockClient(RedictModule_GetBlockedClientHandle(ctx), NULL);

        /* continue blocking */
        result = REDICTMODULE_ERR;
    } else {
        result = RedictModule_ReplyWithError(ctx, REDICTMODULE_ERRORMSG_WRONGTYPE);
    }
    RedictModule_CloseKey(key);
    return result;
}

int blockonkeys_blpopn_timeout_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    return RedictModule_ReplyWithError(ctx, "ERR Timeout");
}

int blockonkeys_blpopn_abort_callback(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);
    return RedictModule_ReplyWithSimpleString(ctx, "Action aborted");
}

/* BLOCKONKEYS.BLPOPN key N
 *
 * Blocks until key has N elements and then pops them or fails after 3 seconds.
 */
int blockonkeys_blpopn(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    if (argc < 3) return RedictModule_WrongArity(ctx);

    long long n, timeout = 3000LL;
    if (RedictModule_StringToLongLong(argv[2], &n) != REDICTMODULE_OK) {
        return RedictModule_ReplyWithError(ctx, "ERR Invalid N");
    }

    if (argc > 3 ) {
        if (RedictModule_StringToLongLong(argv[3], &timeout) != REDICTMODULE_OK) {
            return RedictModule_ReplyWithError(ctx, "ERR Invalid timeout value");
        }
    }
    RedictModuleKey *key = RedictModule_OpenKey(ctx, argv[1], REDICTMODULE_WRITE);
    int keytype = RedictModule_KeyType(key);
    if (keytype != REDICTMODULE_KEYTYPE_EMPTY &&
        keytype != REDICTMODULE_KEYTYPE_LIST) {
        RedictModule_ReplyWithError(ctx, REDICTMODULE_ERRORMSG_WRONGTYPE);
    } else if (keytype == REDICTMODULE_KEYTYPE_LIST &&
               RedictModule_ValueLength(key) >= (size_t)n) {
        RedictModule_ReplyWithArray(ctx, n);
        for (long i = 0; i < n; i++) {
            RedictModuleString *elem = RedictModule_ListPop(key, REDICTMODULE_LIST_HEAD);
            RedictModule_ReplyWithString(ctx, elem);
            RedictModule_FreeString(ctx, elem);
        }
        /* I'm lazy so i'll replicate a potentially blocking command, it shouldn't block in this flow. */
        RedictModule_ReplicateVerbatim(ctx);
    } else {
        RedictModule_BlockClientOnKeys(ctx, blockonkeys_blpopn_reply_callback,
                                      timeout ? blockonkeys_blpopn_timeout_callback : blockonkeys_blpopn_abort_callback,
                                      NULL, timeout, &argv[1], 1, NULL);
    }
    RedictModule_CloseKey(key);
    return REDICTMODULE_OK;
}

int RedictModule_OnLoad(RedictModuleCtx *ctx, RedictModuleString **argv, int argc) {
    REDICTMODULE_NOT_USED(argv);
    REDICTMODULE_NOT_USED(argc);

    if (RedictModule_Init(ctx, "blockonkeys", 1, REDICTMODULE_APIVER_1)== REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    RedictModuleTypeMethods tm = {
        .version = REDICTMODULE_TYPE_METHOD_VERSION,
        .rdb_load = fsl_rdb_load,
        .rdb_save = fsl_rdb_save,
        .aof_rewrite = fsl_aofrw,
        .mem_usage = NULL,
        .free = fsl_free,
        .digest = NULL,
    };

    fsltype = RedictModule_CreateDataType(ctx, "fsltype_t", 0, &tm);
    if (fsltype == NULL)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx,"fsl.push",fsl_push,"write",1,1,1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx,"fsl.pushtimer",fsl_pushtimer,"write",1,1,1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx,"fsl.bpop",fsl_bpop,"write",1,1,1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx,"fsl.bpopgt",fsl_bpopgt,"write",1,1,1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx,"fsl.bpoppush",fsl_bpoppush,"write",1,2,1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx,"fsl.getall",fsl_getall,"",1,1,1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx, "blockonkeys.popall", blockonkeys_popall,
                                  "write", 1, 1, 1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx, "blockonkeys.lpush", blockonkeys_lpush,
                                  "write", 1, 1, 1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx, "blockonkeys.lpush_unblock", blockonkeys_lpush,
                                  "write", 1, 1, 1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx, "blockonkeys.blpopn", blockonkeys_blpopn,
                                  "write", 1, 1, 1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;

    if (RedictModule_CreateCommand(ctx, "blockonkeys.blpopn_or_unblock", blockonkeys_blpopn,
                                      "write", 1, 1, 1) == REDICTMODULE_ERR)
        return REDICTMODULE_ERR;
    return REDICTMODULE_OK;
}