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

    const int ARGV_REDIS_KEY     = 1;
    const int ARGV_SUB_ID        = 2;
    const int ARGV_MARKER_ID     = 3;
    const int ARGV_MARKER_DIR    = 4;
    const int ARGV_REF_FIELD     = 5;
    int ARGV_NODE_ID             = 5;
    int ARGV_FIELDS              = 6;
    int ARGV_FIELD_NAMES         = 7;
    int ARGV_FILTER_EXPR         = 6;
    int ARGV_FILTER_ARGS         = 7;
#define SHIFT_ARGS(i) \
    ARGV_NODE_ID += i; \
    ARGV_FIELDS += i; \
    ARGV_FIELD_NAMES += i; \
    ARGV_FILTER_EXPR += i; \
    ARGV_FILTER_ARGS += i

    if (argc < ARGV_NODE_ID + 1) {
        return RedisModule_WrongArity(ctx);
    }

    /*
     * 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;
    }

    /*
     * Get the subscription id.
     */
    Selva_SubscriptionId sub_id;
    err = SelvaArgParser_SubscriptionId(sub_id, argv[ARGV_SUB_ID]);
    if (err) {
        return replyWithSelvaErrorf(ctx, err, "Subscription ID");
    }

    /*
     * Get the marker id.
     */
    Selva_SubscriptionMarkerId marker_id;
    err = SelvaArgParser_MarkerId(&marker_id, argv[ARGV_MARKER_ID]);
    if (err) {
        return replyWithSelvaErrorf(ctx, err, "Marker ID");
    }

    if (SelvaSubscriptions_GetMarker(hierarchy, sub_id, marker_id)) {
        /* Marker already created. */
        return RedisModule_ReplyWithLongLong(ctx, 1);
    }

    /*
     * Parse the traversal argument.
     */
    enum SelvaTraversal sub_dir;
    const char *ref_field = NULL;
    err = SelvaTraversal_ParseDir2(&sub_dir, argv[ARGV_MARKER_DIR]);
    if (err ||
        !(sub_dir &
          (SELVA_HIERARCHY_TRAVERSAL_NONE | SELVA_HIERARCHY_TRAVERSAL_NODE | SELVA_HIERARCHY_TRAVERSAL_CHILDREN |
           SELVA_HIERARCHY_TRAVERSAL_PARENTS | SELVA_HIERARCHY_TRAVERSAL_BFS_ANCESTORS | SELVA_HIERARCHY_TRAVERSAL_BFS_DESCENDANTS |
           SELVA_HIERARCHY_TRAVERSAL_REF | SELVA_HIERARCHY_TRAVERSAL_EDGE_FIELD | SELVA_HIERARCHY_TRAVERSAL_BFS_EDGE_FIELD |
           SELVA_HIERARCHY_TRAVERSAL_BFS_EXPRESSION | SELVA_HIERARCHY_TRAVERSAL_EXPRESSION)
        )) {
        return replyWithSelvaErrorf(ctx, err, "Traversal argument");
    }

    if (sub_dir & (SELVA_HIERARCHY_TRAVERSAL_REF | SELVA_HIERARCHY_TRAVERSAL_BFS_EDGE_FIELD | SELVA_HIERARCHY_TRAVERSAL_EDGE_FIELD)) {
        ref_field = RedisModule_StringPtrLen(argv[ARGV_REF_FIELD], NULL);
        SHIFT_ARGS(1);
    }

    struct rpn_expression *traversal_expression = NULL;
    struct rpn_ctx *filter_ctx = NULL;
    struct rpn_expression *filter_expression = NULL;
    if (sub_dir & (SELVA_HIERARCHY_TRAVERSAL_BFS_EXPRESSION | SELVA_HIERARCHY_TRAVERSAL_EXPRESSION)) {
        const RedisModuleString *input = argv[ARGV_REF_FIELD];
        TO_STR(input);

        traversal_expression = rpn_compile(input_str);
        if (!traversal_expression) {
            err = SELVA_RPN_ECOMP;
            replyWithSelvaErrorf(ctx, err, "Failed to compile the traversal expression");
            goto out;
        }
        SHIFT_ARGS(1);
    }

    /*
     * Get the nodeId.
     */
    Selva_NodeId node_id;
    SelvaArgParser_NodeId(node_id, argv[ARGV_NODE_ID]);

    /*
     * Get field names for change events.
     * Optional.
     */
    const char *fields = NULL;
    if (argc > ARGV_FIELD_NAMES) {
        err = SelvaArgParser_StrOpt(&fields, "fields", argv[ARGV_FIELDS], argv[ARGV_FIELD_NAMES]);
        if (err == 0) {
            SHIFT_ARGS(2);
        } else if (err != SELVA_ENOENT) {
            replyWithSelvaErrorf(ctx, err, "Fields");
            goto out;
        }
    }

    /*
     * Parse & compile the filter expression.
     * Optional.
     */
    if (argc >= ARGV_FILTER_EXPR + 1) {
        const int nr_reg = argc - ARGV_FILTER_ARGS + 2;
        const char *input;
        size_t input_len;

        filter_ctx = rpn_init(nr_reg);
        if (!filter_ctx) {
            err = SELVA_SUBSCRIPTIONS_ENOMEM;
            replyWithSelvaErrorf(ctx, err, "Filter");
            goto out;
        }

        /*
         * Compile the filter expression.
         */
        input = RedisModule_StringPtrLen(argv[ARGV_FILTER_EXPR], &input_len);
        filter_expression = rpn_compile(input);
        if (!filter_expression) {
            err = SELVA_RPN_ECOMP;
            replyWithSelvaErrorf(ctx, err, "Failed to compile the traversal expression");
            goto out;
        }

        /*
         * Get the filter expression arguments and set them to the registers.
         */
        for (int i = ARGV_FILTER_ARGS; i < argc; i++) {
            /* reg[0] is reserved for the current nodeId */
            const size_t reg_i = i - ARGV_FILTER_ARGS + 1;
            size_t str_len;
            const char *str;
            char *arg;

            /*
             * Args needs to be duplicated so the strings don't get freed
             * when the command returns.
             */
            str = RedisModule_StringPtrLen(argv[i], &str_len);
            str_len++;
            arg = RedisModule_Alloc(str_len);
            memcpy(arg, str, str_len);

            rpn_set_reg(filter_ctx, reg_i, arg, str_len, RPN_SET_REG_FLAG_RMFREE);
        }
    }

    unsigned short marker_flags = 0;

    if (!memcmp(node_id, ROOT_NODE_ID, SELVA_NODE_ID_SIZE)) {
        /*
         * A root node marker is a special case which is stored as detached to
         * save time and space.
         */
        marker_flags = SELVA_SUBSCRIPTION_FLAG_DETACH;
    } else if (sub_dir & (SELVA_HIERARCHY_TRAVERSAL_CHILDREN | SELVA_HIERARCHY_TRAVERSAL_PARENTS)) {
        /*
         * RFE We might want to have an arg for REF flag
         * but currently it seems to be enough to support
         * it only for these specific traversal types.
         */
        marker_flags = SELVA_SUBSCRIPTION_FLAG_REF;
    }

    /*
     * Marker is a hierarchy change marker only if it opts for hierarchy
     * field updates. Otherwise hierarchy events are only sent when the
     * subscription needs a refresh.
     */
    if (fields && contains_hierarchy_fields(fields)) {
        marker_flags |= SELVA_SUBSCRIPTION_FLAG_CH_HIERARCHY;
    }

    struct Selva_SubscriptionMarker *marker;
    err = new_marker(hierarchy, sub_id, marker_id, marker_flags, defer_update_event, &marker);
    if (err) {
        if (err == SELVA_SUBSCRIPTIONS_EEXIST) {
            /* This shouldn't happen as we check for this already before. */
            if (traversal_expression) {
                rpn_destroy_expression(traversal_expression);
            }
            if (filter_ctx) {
                rpn_destroy(filter_ctx);
                rpn_destroy_expression(filter_expression);
            }

            return RedisModule_ReplyWithLongLong(ctx, 1);
        }

        replyWithSelvaErrorf(ctx, err, "Failed to create a subscription");
        goto out;
    }

    marker_set_node_id(marker, node_id);
    marker_set_dir(marker, sub_dir);
    if (ref_field) {
        err = marker_set_ref_field(marker, ref_field);
        if (err) {
            replyWithSelvaErrorf(ctx, err, "ref_field");
            goto out;
        }
    }
    if (traversal_expression) {
        marker_set_traversal_expression(marker, traversal_expression);
    }

    if (fields) {
        err = marker_set_fields(marker, fields);
        if (err) {
            replyWithSelvaErrorf(ctx, err, "fields");
            goto out;
        }
    }
    marker_set_filter(marker, filter_ctx, filter_expression);

out:
    if (err) {
        if (traversal_expression) {
            rpn_destroy_expression(traversal_expression);
        }
        if (filter_ctx) {
            rpn_destroy(filter_ctx);
            rpn_destroy_expression(filter_expression);
        }
    } else {
        RedisModule_ReplyWithLongLong(ctx, 1);
    }

    return REDISMODULE_OK;
#undef SHIFT_ARGS
}
/*
 * Add a new marker to the subscription.
 * KEY SUB_ID MARKER_ID traversal_type [ref_field_name] NODE_ID [fields <fieldnames \n separated>] [filter expression] [filter args...]
 */
Add a new marker to the subscription.
Add a new marker to the subscription .
int SelvaSubscriptions_AddMarkerCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
    RedisModule_AutoMemory(ctx);
    int err;
    const int ARGV_REDIS_KEY     = 1;
    const int ARGV_SUB_ID        = 2;
    const int ARGV_MARKER_ID     = 3;
    const int ARGV_MARKER_DIR    = 4;
    const int ARGV_REF_FIELD     = 5;
    int ARGV_NODE_ID             = 5;
    int ARGV_FIELDS              = 6;
    int ARGV_FIELD_NAMES         = 7;
    int ARGV_FILTER_EXPR         = 6;
    int ARGV_FILTER_ARGS         = 7;
#define SHIFT_ARGS(i) \
    ARGV_NODE_ID += i; \
    ARGV_FIELDS += i; \
    ARGV_FIELD_NAMES += i; \
    ARGV_FILTER_EXPR += i; \
    ARGV_FILTER_ARGS += i
    if (argc < ARGV_NODE_ID + 1) {
        return RedisModule_WrongArity(ctx);
    }
    SelvaModify_Hierarchy *hierarchy = SelvaModify_OpenHierarchy(ctx, argv[ARGV_REDIS_KEY], REDISMODULE_READ | REDISMODULE_WRITE);
    if (!hierarchy) {
        return REDISMODULE_OK;
    }
    Selva_SubscriptionId sub_id;
    err = SelvaArgParser_SubscriptionId(sub_id, argv[ARGV_SUB_ID]);
    if (err) {
        return replyWithSelvaErrorf(ctx, err, "Subscription ID");
    }
    Selva_SubscriptionMarkerId marker_id;
    err = SelvaArgParser_MarkerId(&marker_id, argv[ARGV_MARKER_ID]);
    if (err) {
        return replyWithSelvaErrorf(ctx, err, "Marker ID");
    }
    if (SelvaSubscriptions_GetMarker(hierarchy, sub_id, marker_id)) {
        return RedisModule_ReplyWithLongLong(ctx, 1);
    }
    enum SelvaTraversal sub_dir;
    const char *ref_field = NULL;
    err = SelvaTraversal_ParseDir2(&sub_dir, argv[ARGV_MARKER_DIR]);
    if (err ||
        !(sub_dir &
          (SELVA_HIERARCHY_TRAVERSAL_NONE | SELVA_HIERARCHY_TRAVERSAL_NODE | SELVA_HIERARCHY_TRAVERSAL_CHILDREN |
           SELVA_HIERARCHY_TRAVERSAL_PARENTS | SELVA_HIERARCHY_TRAVERSAL_BFS_ANCESTORS | SELVA_HIERARCHY_TRAVERSAL_BFS_DESCENDANTS |
           SELVA_HIERARCHY_TRAVERSAL_REF | SELVA_HIERARCHY_TRAVERSAL_EDGE_FIELD | SELVA_HIERARCHY_TRAVERSAL_BFS_EDGE_FIELD |
           SELVA_HIERARCHY_TRAVERSAL_BFS_EXPRESSION | SELVA_HIERARCHY_TRAVERSAL_EXPRESSION)
        )) {
        return replyWithSelvaErrorf(ctx, err, "Traversal argument");
    }
    if (sub_dir & (SELVA_HIERARCHY_TRAVERSAL_REF | SELVA_HIERARCHY_TRAVERSAL_BFS_EDGE_FIELD | SELVA_HIERARCHY_TRAVERSAL_EDGE_FIELD)) {
        ref_field = RedisModule_StringPtrLen(argv[ARGV_REF_FIELD], NULL);
        SHIFT_ARGS(1);
    }
    struct rpn_expression *traversal_expression = NULL;
    struct rpn_ctx *filter_ctx = NULL;
    struct rpn_expression *filter_expression = NULL;
    if (sub_dir & (SELVA_HIERARCHY_TRAVERSAL_BFS_EXPRESSION | SELVA_HIERARCHY_TRAVERSAL_EXPRESSION)) {
        const RedisModuleString *input = argv[ARGV_REF_FIELD];
        TO_STR(input);
        traversal_expression = rpn_compile(input_str);
        if (!traversal_expression) {
            err = SELVA_RPN_ECOMP;
            replyWithSelvaErrorf(ctx, err, "Failed to compile the traversal expression");
            goto out;
        }
        SHIFT_ARGS(1);
    }
    Selva_NodeId node_id;
    SelvaArgParser_NodeId(node_id, argv[ARGV_NODE_ID]);
    const char *fields = NULL;
    if (argc > ARGV_FIELD_NAMES) {
        err = SelvaArgParser_StrOpt(&fields, "fields", argv[ARGV_FIELDS], argv[ARGV_FIELD_NAMES]);
        if (err == 0) {
            SHIFT_ARGS(2);
        } else if (err != SELVA_ENOENT) {
            replyWithSelvaErrorf(ctx, err, "Fields");
            goto out;
        }
    }
    if (argc >= ARGV_FILTER_EXPR + 1) {
        const int nr_reg = argc - ARGV_FILTER_ARGS + 2;
        const char *input;
        size_t input_len;
        filter_ctx = rpn_init(nr_reg);
        if (!filter_ctx) {
            err = SELVA_SUBSCRIPTIONS_ENOMEM;
            replyWithSelvaErrorf(ctx, err, "Filter");
            goto out;
        }
        input = RedisModule_StringPtrLen(argv[ARGV_FILTER_EXPR], &input_len);
        filter_expression = rpn_compile(input);
        if (!filter_expression) {
            err = SELVA_RPN_ECOMP;
            replyWithSelvaErrorf(ctx, err, "Failed to compile the traversal expression");
            goto out;
        }
        for (int i = ARGV_FILTER_ARGS; i < argc; i++) {
            const size_t reg_i = i - ARGV_FILTER_ARGS + 1;
            size_t str_len;
            const char *str;
            char *arg;
            str = RedisModule_StringPtrLen(argv[i], &str_len);
            str_len++;
            arg = RedisModule_Alloc(str_len);
            memcpy(arg, str, str_len);
            rpn_set_reg(filter_ctx, reg_i, arg, str_len, RPN_SET_REG_FLAG_RMFREE);
        }
    }
    unsigned short marker_flags = 0;
    if (!memcmp(node_id, ROOT_NODE_ID, SELVA_NODE_ID_SIZE)) {
        marker_flags = SELVA_SUBSCRIPTION_FLAG_DETACH;
    } else if (sub_dir & (SELVA_HIERARCHY_TRAVERSAL_CHILDREN | SELVA_HIERARCHY_TRAVERSAL_PARENTS)) {
        marker_flags = SELVA_SUBSCRIPTION_FLAG_REF;
    }
    if (fields && contains_hierarchy_fields(fields)) {
        marker_flags |= SELVA_SUBSCRIPTION_FLAG_CH_HIERARCHY;
    }
    struct Selva_SubscriptionMarker *marker;
    err = new_marker(hierarchy, sub_id, marker_id, marker_flags, defer_update_event, &marker);
    if (err) {
        if (err == SELVA_SUBSCRIPTIONS_EEXIST) {
            if (traversal_expression) {
                rpn_destroy_expression(traversal_expression);
            }
            if (filter_ctx) {
                rpn_destroy(filter_ctx);
                rpn_destroy_expression(filter_expression);
            }
            return RedisModule_ReplyWithLongLong(ctx, 1);
        }
        replyWithSelvaErrorf(ctx, err, "Failed to create a subscription");
        goto out;
    }
    marker_set_node_id(marker, node_id);
    marker_set_dir(marker, sub_dir);
    if (ref_field) {
        err = marker_set_ref_field(marker, ref_field);
        if (err) {
            replyWithSelvaErrorf(ctx, err, "ref_field");
            goto out;
        }
    }
    if (traversal_expression) {
        marker_set_traversal_expression(marker, traversal_expression);
    }
    if (fields) {
        err = marker_set_fields(marker, fields);
        if (err) {
            replyWithSelvaErrorf(ctx, err, "fields");
            goto out;
        }
    }
    marker_set_filter(marker, filter_ctx, filter_expression);
out:
    if (err) {
        if (traversal_expression) {
            rpn_destroy_expression(traversal_expression);
        }
        if (filter_ctx) {
            rpn_destroy(filter_ctx);
            rpn_destroy_expression(filter_expression);
        }
    } else {
        RedisModule_ReplyWithLongLong(ctx, 1);
    }
    return REDISMODULE_OK;
#undef SHIFT_ARGS
}
int SelvaSubscriptions_AddMarkerCommand ( RedisModuleCtx * ctx , RedisModuleString * * argv , int argc ) { RedisModule_AutoMemory ( ctx ) ; int err ; const int ARGV_REDIS_KEY = 1 ; const int ARGV_SUB_ID = 2 ; const int ARGV_MARKER_ID = 3 ; const int ARGV_MARKER_DIR = 4 ; const int ARGV_REF_FIELD = 5 ; int ARGV_NODE_ID = 5 ; int ARGV_FIELDS = 6 ; int ARGV_FIELD_NAMES = 7 ; int ARGV_FILTER_EXPR = 6 ; int ARGV_FILTER_ARGS = 7 ; #define SHIFT_ARGS ( i )  \
    ARGV_NODE_ID += i; \
    ARGV_FIELDS += i; \
    ARGV_FIELD_NAMES += i; \
    ARGV_FILTER_EXPR += i; \
    ARGV_FILTER_ARGS += i if ( argc < ARGV_NODE_ID + 1 ) { return RedisModule_WrongArity ( ctx ) ; } SelvaModify_Hierarchy * hierarchy = SelvaModify_OpenHierarchy ( ctx , argv [ ARGV_REDIS_KEY ] , REDISMODULE_READ | REDISMODULE_WRITE ) ; if ( ! hierarchy ) { return REDISMODULE_OK ; } Selva_SubscriptionId sub_id ; err = SelvaArgParser_SubscriptionId ( sub_id , argv [ ARGV_SUB_ID ] ) ; if ( err ) { return replyWithSelvaErrorf ( ctx , err , " " ) ; } Selva_SubscriptionMarkerId marker_id ; err = SelvaArgParser_MarkerId ( & marker_id , argv [ ARGV_MARKER_ID ] ) ; if ( err ) { return replyWithSelvaErrorf ( ctx , err , " " ) ; } if ( SelvaSubscriptions_GetMarker ( hierarchy , sub_id , marker_id ) ) { return RedisModule_ReplyWithLongLong ( ctx , 1 ) ; } enum SelvaTraversal sub_dir ; const char * ref_field = NULL ; err = SelvaTraversal_ParseDir2 ( & sub_dir , argv [ ARGV_MARKER_DIR ] ) ; if ( err || ! ( sub_dir & ( SELVA_HIERARCHY_TRAVERSAL_NONE | SELVA_HIERARCHY_TRAVERSAL_NODE | SELVA_HIERARCHY_TRAVERSAL_CHILDREN | SELVA_HIERARCHY_TRAVERSAL_PARENTS | SELVA_HIERARCHY_TRAVERSAL_BFS_ANCESTORS | SELVA_HIERARCHY_TRAVERSAL_BFS_DESCENDANTS | SELVA_HIERARCHY_TRAVERSAL_REF | SELVA_HIERARCHY_TRAVERSAL_EDGE_FIELD | SELVA_HIERARCHY_TRAVERSAL_BFS_EDGE_FIELD | SELVA_HIERARCHY_TRAVERSAL_BFS_EXPRESSION | SELVA_HIERARCHY_TRAVERSAL_EXPRESSION ) ) ) { return replyWithSelvaErrorf ( ctx , err , " " ) ; } if ( sub_dir & ( SELVA_HIERARCHY_TRAVERSAL_REF | SELVA_HIERARCHY_TRAVERSAL_BFS_EDGE_FIELD | SELVA_HIERARCHY_TRAVERSAL_EDGE_FIELD ) ) { ref_field = RedisModule_StringPtrLen ( argv [ ARGV_REF_FIELD ] , NULL ) ; SHIFT_ARGS ( 1 ) ; } struct rpn_expression * traversal_expression = NULL ; struct rpn_ctx * filter_ctx = NULL ; struct rpn_expression * filter_expression = NULL ; if ( sub_dir & ( SELVA_HIERARCHY_TRAVERSAL_BFS_EXPRESSION | SELVA_HIERARCHY_TRAVERSAL_EXPRESSION ) ) { const RedisModuleString * input = argv [ ARGV_REF_FIELD ] ; TO_STR ( input ) ; traversal_expression = rpn_compile ( input_str ) ; if ( ! traversal_expression ) { err = SELVA_RPN_ECOMP ; replyWithSelvaErrorf ( ctx , err , " " ) ; goto out ; } SHIFT_ARGS ( 1 ) ; } Selva_NodeId node_id ; SelvaArgParser_NodeId ( node_id , argv [ ARGV_NODE_ID ] ) ; const char * fields = NULL ; if ( argc > ARGV_FIELD_NAMES ) { err = SelvaArgParser_StrOpt ( & fields , " " , argv [ ARGV_FIELDS ] , argv [ ARGV_FIELD_NAMES ] ) ; if ( err == 0 ) { SHIFT_ARGS ( 2 ) ; } else if ( err != SELVA_ENOENT ) { replyWithSelvaErrorf ( ctx , err , " " ) ; goto out ; } } if ( argc >= ARGV_FILTER_EXPR + 1 ) { const int nr_reg = argc - ARGV_FILTER_ARGS + 2 ; const char * input ; size_t input_len ; filter_ctx = rpn_init ( nr_reg ) ; if ( ! filter_ctx ) { err = SELVA_SUBSCRIPTIONS_ENOMEM ; replyWithSelvaErrorf ( ctx , err , " " ) ; goto out ; } input = RedisModule_StringPtrLen ( argv [ ARGV_FILTER_EXPR ] , & input_len ) ; filter_expression = rpn_compile ( input ) ; if ( ! filter_expression ) { err = SELVA_RPN_ECOMP ; replyWithSelvaErrorf ( ctx , err , " " ) ; goto out ; } for ( int i = ARGV_FILTER_ARGS ; i < argc ; i ++ ) { const size_t reg_i = i - ARGV_FILTER_ARGS + 1 ; size_t str_len ; const char * str ; char * arg ; str = RedisModule_StringPtrLen ( argv [ i ] , & str_len ) ; str_len ++ ; arg = RedisModule_Alloc ( str_len ) ; memcpy ( arg , str , str_len ) ; rpn_set_reg ( filter_ctx , reg_i , arg , str_len , RPN_SET_REG_FLAG_RMFREE ) ; } } unsigned short marker_flags = 0 ; if ( ! memcmp ( node_id , ROOT_NODE_ID , SELVA_NODE_ID_SIZE ) ) { marker_flags = SELVA_SUBSCRIPTION_FLAG_DETACH ; } else if ( sub_dir & ( SELVA_HIERARCHY_TRAVERSAL_CHILDREN | SELVA_HIERARCHY_TRAVERSAL_PARENTS ) ) { marker_flags = SELVA_SUBSCRIPTION_FLAG_REF ; } if ( fields && contains_hierarchy_fields ( fields ) ) { marker_flags |= SELVA_SUBSCRIPTION_FLAG_CH_HIERARCHY ; } struct Selva_SubscriptionMarker * marker ; err = new_marker ( hierarchy , sub_id , marker_id , marker_flags , defer_update_event , & marker ) ; if ( err ) { if ( err == SELVA_SUBSCRIPTIONS_EEXIST ) { if ( traversal_expression ) { rpn_destroy_expression ( traversal_expression ) ; } if ( filter_ctx ) { rpn_destroy ( filter_ctx ) ; rpn_destroy_expression ( filter_expression ) ; } return RedisModule_ReplyWithLongLong ( ctx , 1 ) ; } replyWithSelvaErrorf ( ctx , err , " " ) ; goto out ; } marker_set_node_id ( marker , node_id ) ; marker_set_dir ( marker , sub_dir ) ; if ( ref_field ) { err = marker_set_ref_field ( marker , ref_field ) ; if ( err ) { replyWithSelvaErrorf ( ctx , err , " " ) ; goto out ; } } if ( traversal_expression ) { marker_set_traversal_expression ( marker , traversal_expression ) ; } if ( fields ) { err = marker_set_fields ( marker , fields ) ; if ( err ) { replyWithSelvaErrorf ( ctx , err , " " ) ; goto out ; } } marker_set_filter ( marker , filter_ctx , filter_expression ) ; out : if ( err ) { if ( traversal_expression ) { rpn_destroy_expression ( traversal_expression ) ; } if ( filter_ctx ) { rpn_destroy ( filter_ctx ) ; rpn_destroy_expression ( filter_expression ) ; } } else { RedisModule_ReplyWithLongLong ( ctx , 1 ) ; } return REDISMODULE_OK ; #undef  SHIFT_ARGS }
Add a new marker to the subscription.
Add a new marker to the subscription .
/*
     * Open the Redis key.
     */ /* Do not send redis messages here. */ /*
     * Get the subscription id.
     */ /*
     * Get the marker id.
     */ /* Marker already created. */ /*
     * Parse the traversal argument.
     */ /*
     * Get the nodeId.
     */ /*
     * Get field names for change events.
     * Optional.
     */ /*
     * Parse & compile the filter expression.
     * Optional.
     */ /*
         * Compile the filter expression.
         */ /*
         * Get the filter expression arguments and set them to the registers.
         */ /* reg[0] is reserved for the current nodeId */ /*
             * Args needs to be duplicated so the strings don't get freed
             * when the command returns.
             */ /*
         * A root node marker is a special case which is stored as detached to
         * save time and space.
         */ /*
         * RFE We might want to have an arg for REF flag
         * but currently it seems to be enough to support
         * it only for these specific traversal types.
         */ /*
     * Marker is a hierarchy change marker only if it opts for hierarchy
     * field updates. Otherwise hierarchy events are only sent when the
     * subscription needs a refresh.
     */ /* This shouldn't happen as we check for this already before. */