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