-
Notifications
You must be signed in to change notification settings - Fork 19
/
disquecmd.c
131 lines (115 loc) · 4.5 KB
/
disquecmd.c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
/* Disque Command implementation.
*
* Copyright (c) 2019, Salvatore Sanfilippo <antirez at gmail dot com>
* All rights reserved. This code is under the AGPL license, check the
* LICENSE file for more info. */
#include "disque.h"
#define DISQUE_INFO_TYPE_INT 0
#define DISQUE_INFO_TYPE_UINT 1
#define DISQUE_INFO_TYPE_LONG 2
#define DISQUE_INFO_TYPE_ULONG 3
#define DISQUE_INFO_TYPE_LONGLONG 4
#define DISQUE_INFO_TYPE_ULONGLONG 5
/* INFO getter function for Rax size. */
void disqueInfoGetRaxSize(void *aux, void *valueptr, size_t *lenptr) {
UNUSED(lenptr);
unsigned long long val = raxSize(*((rax**)aux));
memcpy(valueptr,&val,sizeof(val));
}
struct disqueInfoProperty {
char *name;
int type;
void *valueptr;
void (*getvalue)(void *aux, void *valueptr, size_t *lenptr);
void *aux;
} DisqueInfoPropertiesTable[] = {
/* cluster.* */
{"cluster.nodes.reachable",
DISQUE_INFO_TYPE_INT,
&ClusterReachableNodesCount,NULL,NULL},
/* jobs.* */
{"jobs.registered",
DISQUE_INFO_TYPE_ULONGLONG,
NULL,disqueInfoGetRaxSize,&Jobs},
/* queues.* */
{"queues.registered",
DISQUE_INFO_TYPE_ULONGLONG,
NULL,disqueInfoGetRaxSize,&Queues},
/* Final terminator. */
{NULL,0,NULL,NULL,NULL}
};
/* DISQUE INFO [option name] */
int disqueInfo(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 2 & argc != 3) return RedisModule_WrongArity(ctx);
int allfields = 0, numfields = 0;
const char *wanted = NULL;
/* If no info property name is given, reply with a human readable string.
* Otherwise just provide the final value of the specified info
* property. */
if (argc == 3) {
wanted = RedisModule_StringPtrLen(argv[2],NULL);
} else {
allfields = 1;
RedisModule_ReplyWithArray(ctx,REDISMODULE_POSTPONED_ARRAY_LEN);
}
/* Scan all the options: either search a match in case of single option
* specified, or accumulate all the data in a string. */
for (int j = 0; DisqueInfoPropertiesTable[j].name; j++) {
/* Fetch the info. */
struct disqueInfoProperty *dp = DisqueInfoPropertiesTable+j;
int intval;
unsigned long long ulonglongval;
if (dp->type == DISQUE_INFO_TYPE_INT) {
if (dp->getvalue) {
dp->getvalue(dp->aux,&intval,NULL);
} else {
intval = *(int*)(dp->valueptr);
}
} else if (dp->type == DISQUE_INFO_TYPE_ULONGLONG) {
if (dp->getvalue) {
dp->getvalue(dp->aux,&ulonglongval,NULL);
} else {
ulonglongval = *(unsigned long long*)(dp->valueptr);
}
}
/* Either emit every entry as an array of strings or, in case the
* user specified a single property, emit it if there is a match. */
if (allfields) {
sds output = sdsempty();
if (dp->type == DISQUE_INFO_TYPE_INT)
output = sdscatprintf(output,"%s:%d",dp->name,intval);
else if (dp->type == DISQUE_INFO_TYPE_ULONGLONG)
output = sdscatprintf(output,"%s:%llu",dp->name,ulonglongval);
RedisModule_ReplyWithStringBuffer(ctx,output,sdslen(output));
sdsfree(output);
numfields++;
} else if (!strcasecmp(wanted,dp->name)) {
if (dp->type == DISQUE_INFO_TYPE_INT)
return RedisModule_ReplyWithLongLong(ctx,intval);
else if (dp->type == DISQUE_INFO_TYPE_ULONGLONG)
return RedisModule_ReplyWithLongLong(ctx,ulonglongval);
}
}
/* Emit the arrays of strings or return NULL if we got no match in case the
* caller was looking for a specific property value. */
if (allfields) {
RedisModule_ReplySetArrayLength(ctx,numfields);
} else {
RedisModule_ReplyWithNull(ctx);
}
return REDISMODULE_OK;
}
/* The DISQUE command implements all the disque specific features. */
int disqueCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 2) return RedisModule_WrongArity(ctx);
const char *opt = RedisModule_StringPtrLen(argv[1],NULL);
if (!strcasecmp(opt,"info")) {
return disqueInfo(ctx,argv,argc);
} else if (!strcasecmp(opt,"flushall")) {
flushAllJobsAndQueues(ctx);
RedisModule_ReplyWithSimpleString(ctx,"OK");
} else {
RedisModule_ReplyWithError(ctx,"ERR Unknown DISQUE subcommand");
}
return REDISMODULE_OK;
}