| 1952 | * on slaves, XREAD-GROUP is not. */ |
| 1953 | #define XREAD_BLOCKED_DEFAULT_COUNT 1000 |
| 1954 | void xreadCommand(client *c) { |
| 1955 | long long timeout = -1; /* -1 means, no BLOCK argument given. */ |
| 1956 | long long count = 0; |
| 1957 | int streams_count = 0; |
| 1958 | int streams_arg = 0; |
| 1959 | int noack = 0; /* True if NOACK option was specified. */ |
| 1960 | streamID static_ids[STREAMID_STATIC_VECTOR_LEN]; |
| 1961 | streamID *ids = static_ids; |
| 1962 | streamCG **groups = NULL; |
| 1963 | int xreadgroup = sdslen(c->argv[0]->ptr) == 10; /* XREAD or XREADGROUP? */ |
| 1964 | robj *groupname = NULL; |
| 1965 | robj *consumername = NULL; |
| 1966 | |
| 1967 | /* Parse arguments. */ |
| 1968 | for (int i = 1; i < c->argc; i++) { |
| 1969 | int moreargs = c->argc-i-1; |
| 1970 | char *o = c->argv[i]->ptr; |
| 1971 | if (!strcasecmp(o,"BLOCK") && moreargs) { |
| 1972 | if (c->flags & CLIENT_LUA) { |
| 1973 | /* |
| 1974 | * Although the CLIENT_DENY_BLOCKING flag should protect from blocking the client |
| 1975 | * on Lua/MULTI/RM_Call we want special treatment for Lua to keep backword compatibility. |
| 1976 | * There is no sense to use BLOCK option within Lua. */ |
| 1977 | addReplyErrorFormat(c, "%s command is not allowed with BLOCK option from scripts", (char *)c->argv[0]->ptr); |
| 1978 | return; |
| 1979 | } |
| 1980 | i++; |
| 1981 | if (getTimeoutFromObjectOrReply(c,c->argv[i],&timeout, |
| 1982 | UNIT_MILLISECONDS) != C_OK) return; |
| 1983 | } else if (!strcasecmp(o,"COUNT") && moreargs) { |
| 1984 | i++; |
| 1985 | if (getLongLongFromObjectOrReply(c,c->argv[i],&count,NULL) != C_OK) |
| 1986 | return; |
| 1987 | if (count < 0) count = 0; |
| 1988 | } else if (!strcasecmp(o,"STREAMS") && moreargs) { |
| 1989 | streams_arg = i+1; |
| 1990 | streams_count = (c->argc-streams_arg); |
| 1991 | if ((streams_count % 2) != 0) { |
| 1992 | addReplyError(c,"Unbalanced XREAD list of streams: " |
| 1993 | "for each stream key an ID or '$' must be " |
| 1994 | "specified."); |
| 1995 | return; |
| 1996 | } |
| 1997 | streams_count /= 2; /* We have two arguments for each stream. */ |
| 1998 | break; |
| 1999 | } else if (!strcasecmp(o,"GROUP") && moreargs >= 2) { |
| 2000 | if (!xreadgroup) { |
| 2001 | addReplyError(c,"The GROUP option is only supported by " |
| 2002 | "XREADGROUP. You called XREAD instead."); |
| 2003 | return; |
| 2004 | } |
| 2005 | groupname = c->argv[i+1]; |
| 2006 | consumername = c->argv[i+2]; |
| 2007 | i += 2; |
| 2008 | } else if (!strcasecmp(o,"NOACK")) { |
| 2009 | if (!xreadgroup) { |
| 2010 | addReplyError(c,"The NOACK option is only supported by " |
| 2011 | "XREADGROUP. You called XREAD instead."); |
nothing calls this directly
no test coverage detected