server/modules/selva/module/subscriptions.c
int SelvaSubscriptions_AddMissingCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    int err;

    if (argc < 4) {
        return RedisModule_WrongArity(ctx);
    }

    const int ARGV_REDIS_KEY = 1;
    const int ARGV_SUB_ID    = 2;
    const int ARGV_IDS       = 3;

    Selva_SubscriptionId sub_id;
    err = SelvaArgParser_SubscriptionId(sub_id, argv[ARGV_SUB_ID]);
    if (err) {
        return replyWithSelvaErrorf(ctx, err, "Invalid Subscription ID");
    }

    /*
     * Open the Redis key.
     */
    SelvaModify_Hierarchy *hierarchy = SelvaModify_OpenHierarchy(ctx, argv[ARGV_REDIS_KEY], REDISMODULE_READ | REDISMODULE_WRITE);
    if (!hierarchy) {
        /* Do not send redis messages here. */
        return REDISMODULE_OK;
    }

    /*
     * Open the subscription.
     */
    struct Selva_Subscription *sub;
    sub = find_sub(hierarchy, sub_id);
    if (!sub) {
        sub = create_subscription(hierarchy, sub_id);
        if (!sub) {
            return replyWithSelvaError(ctx, SELVA_SUBSCRIPTIONS_ENOMEM);
        }
    }

    long long n = 0;
    for (int i = ARGV_IDS; i < argc; i++) {
        char sub_id_str[SELVA_SUBSCRIPTION_ID_STR_LEN + 1];
        RedisModuleString *missing;

        missing = RedisModule_CreateStringPrintf(
            ctx, "%s.%.*s",
            RedisModule_StringPtrLen(argv[i], NULL),
            (int)SELVA_SUBSCRIPTION_ID_STR_LEN, Selva_SubscriptionId2str(sub_id_str, sub_id));
        if (!missing) {
            err = SELVA_SUBSCRIPTIONS_ENOMEM;
            return replyWithSelvaErrorf(ctx, err, "Creating missing markers");
        }

        err = SelvaObject_SetPointer(hierarchy->subs.missing, missing, sub, NULL);
        if (err) {
            return replyWithSelvaError(ctx, err);
        }

        n++;
    }

    return RedisModule_ReplyWithLongLong(ctx, n);
}
/**
 * Add missing node/alias markers.
 * SELVA.SUBSCRIPTIONS.ADDMISSING KEY SUB_ID NODEID|ALIAS...
 */
Add missing node/alias markers.
Add missing node / alias markers .
int SelvaSubscriptions_AddMissingCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    int err;
    if (argc < 4) {
        return RedisModule_WrongArity(ctx);
    }
    const int ARGV_REDIS_KEY = 1;
    const int ARGV_SUB_ID    = 2;
    const int ARGV_IDS       = 3;
    Selva_SubscriptionId sub_id;
    err = SelvaArgParser_SubscriptionId(sub_id, argv[ARGV_SUB_ID]);
    if (err) {
        return replyWithSelvaErrorf(ctx, err, "Invalid Subscription ID");
    }
    SelvaModify_Hierarchy *hierarchy = SelvaModify_OpenHierarchy(ctx, argv[ARGV_REDIS_KEY], REDISMODULE_READ | REDISMODULE_WRITE);
    if (!hierarchy) {
        return REDISMODULE_OK;
    }
    struct Selva_Subscription *sub;
    sub = find_sub(hierarchy, sub_id);
    if (!sub) {
        sub = create_subscription(hierarchy, sub_id);
        if (!sub) {
            return replyWithSelvaError(ctx, SELVA_SUBSCRIPTIONS_ENOMEM);
        }
    }
    long long n = 0;
    for (int i = ARGV_IDS; i < argc; i++) {
        char sub_id_str[SELVA_SUBSCRIPTION_ID_STR_LEN + 1];
        RedisModuleString *missing;
        missing = RedisModule_CreateStringPrintf(
            ctx, "%s.%.*s",
            RedisModule_StringPtrLen(argv[i], NULL),
            (int)SELVA_SUBSCRIPTION_ID_STR_LEN, Selva_SubscriptionId2str(sub_id_str, sub_id));
        if (!missing) {
            err = SELVA_SUBSCRIPTIONS_ENOMEM;
            return replyWithSelvaErrorf(ctx, err, "Creating missing markers");
        }
        err = SelvaObject_SetPointer(hierarchy->subs.missing, missing, sub, NULL);
        if (err) {
            return replyWithSelvaError(ctx, err);
        }
        n++;
    }
    return RedisModule_ReplyWithLongLong(ctx, n);
}
int SelvaSubscriptions_AddMissingCommand ( RedisModuleCtx * ctx , RedisModuleString * * argv , int argc ) { int err ; if ( argc < 4 ) { return RedisModule_WrongArity ( ctx ) ; } const int ARGV_REDIS_KEY = 1 ; const int ARGV_SUB_ID = 2 ; const int ARGV_IDS = 3 ; Selva_SubscriptionId sub_id ; err = SelvaArgParser_SubscriptionId ( sub_id , argv [ ARGV_SUB_ID ] ) ; if ( err ) { return replyWithSelvaErrorf ( ctx , err , " " ) ; } SelvaModify_Hierarchy * hierarchy = SelvaModify_OpenHierarchy ( ctx , argv [ ARGV_REDIS_KEY ] , REDISMODULE_READ | REDISMODULE_WRITE ) ; if ( ! hierarchy ) { return REDISMODULE_OK ; } struct Selva_Subscription * sub ; sub = find_sub ( hierarchy , sub_id ) ; if ( ! sub ) { sub = create_subscription ( hierarchy , sub_id ) ; if ( ! sub ) { return replyWithSelvaError ( ctx , SELVA_SUBSCRIPTIONS_ENOMEM ) ; } } long long n = 0 ; for ( int i = ARGV_IDS ; i < argc ; i ++ ) { char sub_id_str [ SELVA_SUBSCRIPTION_ID_STR_LEN + 1 ] ; RedisModuleString * missing ; missing = RedisModule_CreateStringPrintf ( ctx , " " , RedisModule_StringPtrLen ( argv [ i ] , NULL ) , ( int ) SELVA_SUBSCRIPTION_ID_STR_LEN , Selva_SubscriptionId2str ( sub_id_str , sub_id ) ) ; if ( ! missing ) { err = SELVA_SUBSCRIPTIONS_ENOMEM ; return replyWithSelvaErrorf ( ctx , err , " " ) ; } err = SelvaObject_SetPointer ( hierarchy -> subs . missing , missing , sub , NULL ) ; if ( err ) { return replyWithSelvaError ( ctx , err ) ; } n ++ ; } return RedisModule_ReplyWithLongLong ( ctx , n ) ; }
Add missing node/alias markers.
Add missing node / alias markers .
/*
     * Open the Redis key.
     */ /* Do not send redis messages here. */ /*
     * Open the subscription.
     */