XCLAIM * [IDLE ] [TIME ] [RETRYCOUNT ] * [FORCE] [JUSTID] * * Gets ownership of one or multiple messages in the Pending Entries List * of a given stream consumer group. * * If the message ID (among the specified ones) exists, and its idle * time greater or equal to , then the me
| 2843 | * successfully claimed, so that the caller is able to understand |
| 2844 | * what messages it is now in charge of. */ |
| 2845 | void xclaimCommand(client *c) { |
| 2846 | streamCG *group = NULL; |
| 2847 | robj *o = lookupKeyRead(c->db,c->argv[1]); |
| 2848 | long long minidle; /* Minimum idle time argument. */ |
| 2849 | long long retrycount = -1; /* -1 means RETRYCOUNT option not given. */ |
| 2850 | mstime_t deliverytime = -1; /* -1 means IDLE/TIME options not given. */ |
| 2851 | int force = 0; |
| 2852 | int justid = 0; |
| 2853 | |
| 2854 | if (o) { |
| 2855 | if (checkType(c,o,OBJ_STREAM)) return; /* Type error. */ |
| 2856 | group = streamLookupCG(o->ptr,c->argv[2]->ptr); |
| 2857 | } |
| 2858 | |
| 2859 | /* No key or group? Send an error given that the group creation |
| 2860 | * is mandatory. */ |
| 2861 | if (o == NULL || group == NULL) { |
| 2862 | addReplyErrorFormat(c,"-NOGROUP No such key '%s' or " |
| 2863 | "consumer group '%s'", (char*)c->argv[1]->ptr, |
| 2864 | (char*)c->argv[2]->ptr); |
| 2865 | return; |
| 2866 | } |
| 2867 | |
| 2868 | if (getLongLongFromObjectOrReply(c,c->argv[4],&minidle, |
| 2869 | "Invalid min-idle-time argument for XCLAIM") |
| 2870 | != C_OK) return; |
| 2871 | if (minidle < 0) minidle = 0; |
| 2872 | |
| 2873 | /* Start parsing the IDs, so that we abort ASAP if there is a syntax |
| 2874 | * error: the return value of this command cannot be an error in case |
| 2875 | * the client successfully claimed some message, so it should be |
| 2876 | * executed in a "all or nothing" fashion. */ |
| 2877 | int j; |
| 2878 | streamID static_ids[STREAMID_STATIC_VECTOR_LEN]; |
| 2879 | streamID *ids = static_ids; |
| 2880 | int id_count = c->argc-5; |
| 2881 | if (id_count > STREAMID_STATIC_VECTOR_LEN) |
| 2882 | ids = zmalloc(sizeof(streamID)*id_count); |
| 2883 | for (j = 5; j < c->argc; j++) { |
| 2884 | if (streamParseStrictIDOrReply(NULL,c->argv[j],&ids[j-5],0) != C_OK) break; |
| 2885 | } |
| 2886 | int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */ |
| 2887 | |
| 2888 | /* If we stopped because some IDs cannot be parsed, perhaps they |
| 2889 | * are trailing options. */ |
| 2890 | mstime_t now = mstime(); |
| 2891 | streamID last_id = {0,0}; |
| 2892 | int propagate_last_id = 0; |
| 2893 | for (; j < c->argc; j++) { |
| 2894 | int moreargs = (c->argc-1) - j; /* Number of additional arguments. */ |
| 2895 | char *opt = c->argv[j]->ptr; |
| 2896 | if (!strcasecmp(opt,"FORCE")) { |
| 2897 | force = 1; |
| 2898 | } else if (!strcasecmp(opt,"JUSTID")) { |
| 2899 | justid = 1; |
| 2900 | } else if (!strcasecmp(opt,"IDLE") && moreargs) { |
| 2901 | j++; |
| 2902 | if (getLongLongFromObjectOrReply(c,c->argv[j],&deliverytime, |
nothing calls this directly
no test coverage detected