1
/* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
2
 * 0 if the client was already subscribed to that channel. */
3
int pubsubSubscribeChannel(client *c, robj *channel, pubsubtype type) {
4
    dictEntry *de, *existing;
5
    dict *clients = NULL;
6
    int retval = 0;
7
    unsigned int slot = 0;
8

9
    /* Add the channel to the client -> channels hash table */
10
    void *position = dictFindPositionForInsert(type.clientPubSubChannels(c),channel,NULL);
11
    if (position) { /* Not yet subscribed to this channel */
12
        retval = 1;
13
        /* Add the client to the channel -> list of clients hash table */
14
        if (server.cluster_enabled && type.shard) {
15
            slot = getKeySlot(channel->ptr);
16
        }
17

18
        de = kvstoreDictAddRaw(*type.serverPubSubChannels, slot, channel, &existing);
19

20
        if (existing) {
21
            clients = dictGetVal(existing);
22
            channel = dictGetKey(existing);
23
        } else {
24
            clients = dictCreate(&clientDictType);
25
            kvstoreDictSetVal(*type.serverPubSubChannels, slot, de, clients);
26
            incrRefCount(channel);
27
        }
28

29
        serverAssert(dictAdd(clients, c, NULL) != DICT_ERR);
30
        serverAssert(dictInsertAtPosition(type.clientPubSubChannels(c), channel, position));
31
        incrRefCount(channel);
32
    }
33
    /* Notify the client */
34
    addReplyPubsubSubscribed(c,channel,type);
35
    return retval;
36
}

1
/* Subscribe a client to a pattern. Returns 1 if the operation succeeded, or 0 if the client was already subscribed to that pattern. */
2
int pubsubSubscribePattern(client *c, robj *pattern) {
3
    dictEntry *de;
4
    dict *clients;
5
    int retval = 0;
6

7
    if (dictAdd(c->pubsub_patterns, pattern, NULL) == DICT_OK) {
8
        retval = 1;
9
        incrRefCount(pattern);
10
        /* Add the client to the pattern -> list of clients hash table */
11
        de = dictFind(server.pubsub_patterns,pattern);
12
        if (de == NULL) {
13
            clients = dictCreate(&clientDictType);
14
            dictAdd(server.pubsub_patterns,pattern,clients);
15
            incrRefCount(pattern);
16
        } else {
17
            clients = dictGetVal(de);
18
        }
19
        serverAssert(dictAdd(clients, c, NULL) != DICT_ERR);
20
    }
21
    /* Notify the client */
22
    addReplyPubsubPatSubscribed(c,pattern);
23
    return retval;
24
}

1
/*
2
 * Publish a message to all the subscribers.
3
 */
4
int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type) {
5
    int receivers = 0;
6
    dictEntry *de;
7
    dictIterator *di;
8
    unsigned int slot = 0;
9

10
    /* Send to clients listening for that channel */
11
    if (server.cluster_enabled && type.shard) {
12
        slot = keyHashSlot(channel->ptr, sdslen(channel->ptr));
13
    }
14
    de = kvstoreDictFind(*type.serverPubSubChannels, slot, channel);
15
    if (de) {
16
        dict *clients = dictGetVal(de);
17
        dictEntry *entry;
18
        dictIterator *iter = dictGetIterator(clients);
19
        while ((entry = dictNext(iter)) != NULL) {
20
            client *c = dictGetKey(entry);
21
            addReplyPubsubMessage(c,channel,message,*type.messageBulk);
22
            updateClientMemUsageAndBucket(c);
23
            receivers++;
24
        }
25
        dictReleaseIterator(iter);
26
    }
27

28
    if (type.shard) {
29
        /* Shard pubsub ignores patterns. */
30
        return receivers;
31
    }
32

33
    /* Send to clients listening to matching channels */
34
    di = dictGetIterator(server.pubsub_patterns);
35
    if (di) {
36
        channel = getDecodedObject(channel);
37
        while((de = dictNext(di)) != NULL) {
38
            robj *pattern = dictGetKey(de);
39
            dict *clients = dictGetVal(de);
40
            if (!stringmatchlen((char*)pattern->ptr,
41
                                sdslen(pattern->ptr),
42
                                (char*)channel->ptr,
43
                                sdslen(channel->ptr),0)) continue;
44

45
            dictEntry *entry;
46
            dictIterator *iter = dictGetIterator(clients);
47
            while ((entry = dictNext(iter)) != NULL) {
48
                client *c = dictGetKey(entry);
49
                addReplyPubsubPatMessage(c,pattern,channel,message);
50
                updateClientMemUsageAndBucket(c);
51
                receivers++;
52
            }
53
            dictReleaseIterator(iter);
54
        }
55
        decrRefCount(channel);
56
        dictReleaseIterator(di);
57
    }
58
    return receivers;
59
}

1
/* Glob-style pattern matching. */
2
static int stringmatchlen_impl(const char *pattern, int patternLen,
3
        const char *string, int stringLen, int nocase, int *skipLongerMatches, int nesting)
4
{
5
    /* Protection against abusive patterns. */
6
    if (nesting > 1000) return 0;
7

8
    while(patternLen && stringLen) {
9
        switch(pattern[0]) {
10
        case '*':
11
            while (patternLen && pattern[1] == '*') {
12
                pattern++;
13
                patternLen--;
14
            }
15
            if (patternLen == 1)
16
                return 1; /* match */
17
            while(stringLen) {
18
                if (stringmatchlen_impl(pattern+1, patternLen-1,
19
                            string, stringLen, nocase, skipLongerMatches, nesting+1))
20
                    return 1; /* match */
21
                if (*skipLongerMatches)
22

23
                string++;
24
                stringLen--;
25
            }
26
            /* There was no match for the rest of the pattern starting
27
             * from anywhere in the rest of the string. If there were
28
             * any '*' earlier in the pattern, we can terminate the
29
             * search early without trying to match them to longer
30
             * substrings. This is because a longer match for the
31
             * earlier part of the pattern would require the rest of the
32
             * pattern to match starting later in the string, and we
33
             * have just determined that there is no match for the rest
34
             * of the pattern starting from anywhere in the current
35
             * string. */
36
            *skipLongerMatches = 1;
37
            return 0; /* no match */
38
            break;
39
        case '?':
40
            string++;
41
            stringLen--;
42
            break;
43
        case '[':
44
        {
45
            int not, match;
46

47
            pattern++;
48
            patternLen--;
49
            not = patternLen && pattern[0] == '^';
50
            if (not) {
51
                pattern++;
52
                patternLen--;
53
            }
54
            match = 0;
55
            while(1) {
56
                if (patternLen >= 2 && pattern[0] == '\\') {
57
                    pattern++;
58
                    patternLen--;
59
                    if (pattern[0] == string[0])
60
                        match = 1;
61
                } else if (patternLen == 0) {
62
                    pattern--;
63
                    patternLen++;
64
                    break;
65
                } else if (pattern[0] == ']') {
66
                    break;
67
                } else if (patternLen >= 3 && pattern[1] == '-') {
68
                    int start = pattern[0];
69
                    int end = pattern[2];
70
                    int c = string[0];
71
                    if (start > end) {
72
                        int t = start;
73
                        start = end;
74
                        end = t;
75
                    }
76
                    if (nocase) {
77
                        start = tolower(start);
78
                        end = tolower(end);
79
                        c = tolower(c);
80
                    }
81
                    pattern += 2;
82
                    patternLen -= 2;
83
                    if (c >= start && c <= end)
84
                        match = 1;
85
                } else {
86
                    if (!nocase) {
87
                        if (pattern[0] == string[0])
88
                            match = 1;
89
                    } else {
90
                        if (tolower((int)pattern[0]) == tolower((int)string[0]))
91
                            match = 1;
92
                    }
93
                }
94
                pattern++;
95
                patternLen--;
96
            }
97
            if (not)
98
                match = !match;
99
            if (!match)
100
                return 0; /* no match */
101
            string++;
102
            stringLen--;
103
            break;
104
        }
105
        case '\\':
106
            if (patternLen >= 2) {
107
                pattern++;
108
                patternLen--;
109
            }
110
            /* fall through */
111
        default:
112
            if (!nocase) {
113
                if (pattern[0] != string[0])
114
                    return 0; /* no match */
115
            } else {
116
                if (tolower((int)pattern[0]) != tolower((int)string[0]))
117
                    return 0; /* no match */
118
            }
119
            string++;
120
            stringLen--;
121
            break;
122
        }
123
        pattern++;
124
        patternLen--;
125
        if (stringLen == 0) {
126
            while(patternLen && *pattern == '*') {
127
                pattern++;
128
                patternLen--;
129
            }
130
            break;
131
        }
132
    }
133
    if (patternLen == 0 && stringLen == 0)
134
        return 1;
135
    return 0;
136
}

1
#define NOTIFY_GENERIC (1<<2)     /* g */
2
#define NOTIFY_STRING (1<<3)      /* $ */
3
#define NOTIFY_LIST (1<<4)        /* l */
4
#define NOTIFY_SET (1<<5)         /* s */
5
#define NOTIFY_HASH (1<<6)        /* h */
6
#define NOTIFY_ZSET (1<<7)        /* z */
7
#define NOTIFY_EXPIRED (1<<8)     /* x */
8
#define NOTIFY_EVICTED (1<<9)     /* e */
9
#define NOTIFY_STREAM (1<<10)     /* t */
10
#define NOTIFY_KEY_MISS (1<<11)   /* m (Note: This one is excluded from NOTIFY_ALL on purpose) */
11
#define NOTIFY_LOADED (1<<12)     /* module only key space notification, indicate a key loaded from rdb */
12
#define NOTIFY_MODULE (1<<13)     /* d, module key space notification */
13
#define NOTIFY_NEW (1<<14)        /* n, new key notification */

1
/* The API provided to the rest of the Redis core is a simple function:
2
 *
3
 * notifyKeyspaceEvent(int type, char *event, robj *key, int dbid);
4
 *
5
 * 'type' is the notification class we define in `server.h`.
6
 * 'event' is a C string representing the event name.
7
 * 'key' is a Redis object representing the key name.
8
 * 'dbid' is the database ID where the key lives.  */
9
void notifyKeyspaceEvent(int type, const char *event, robj *key, int dbid) {
10
    sds chan;
11
    robj *chanobj, *eventobj;
12
    int len = -1;
13
    char buf[24];
14

15
    /* If any modules are interested in events, notify the module system now.
16
     * This bypasses the notifications configuration, but the module engine
17
     * will only call event subscribers if the event type matches the types
18
     * they are interested in. */
19
     moduleNotifyKeyspaceEvent(type, event, key, dbid);
20

21
    /* If notifications for this class of events are off, return ASAP. */
22
    if (!(server.notify_keyspace_events & type)) return;
23

24
    eventobj = createStringObject(event,strlen(event));
25

26
    /* __keyspace@<db>__:<key> <event> notifications. */
27
    if (server.notify_keyspace_events & NOTIFY_KEYSPACE) {
28
        chan = sdsnewlen("__keyspace@",11);
29
        len = ll2string(buf,sizeof(buf),dbid);
30
        chan = sdscatlen(chan, buf, len);
31
        chan = sdscatlen(chan, "__:", 3);
32
        chan = sdscatsds(chan, key->ptr);
33
        chanobj = createObject(OBJ_STRING, chan);
34
        pubsubPublishMessage(chanobj, eventobj, 0);
35
        decrRefCount(chanobj);
36
    }
37

38
    /* __keyevent@<db>__:<event> <key> notifications. */
39
    if (server.notify_keyspace_events & NOTIFY_KEYEVENT) {
40
        chan = sdsnewlen("__keyevent@",11);
41
        if (len == -1) len = ll2string(buf,sizeof(buf),dbid);
42
        chan = sdscatlen(chan, buf, len);
43
        chan = sdscatlen(chan, "__:", 3);
44
        chan = sdscatsds(chan, eventobj->ptr);
45
        chanobj = createObject(OBJ_STRING, chan);
46
        pubsubPublishMessage(chanobj, key, 0);
47
        decrRefCount(chanobj);
48
    }
49
    decrRefCount(eventobj);
50
}