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
132
133
134
135
136
137
138
|
// Copyright (c) 2020, Michael Grunder <michael dot grunder at gmail dot com>
//
// SPDX-FileCopyrightText: 2024 Hiredict Contributors
// SPDX-FileCopyrightText: 2024 Michael Grunder <michael dot grunder at gmail dot com>
//
// SPDX-License-Identifier: BSD-3-Clause
// SPDX-License-Identifier: LGPL-3.0-or-later
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <hiredict.h>
#define KEY_COUNT 5
#define panicAbort(fmt, ...) \
do { \
fprintf(stderr, "%s:%d:%s(): " fmt, __FILE__, __LINE__, __func__, __VA_ARGS__); \
exit(-1); \
} while (0)
static void assertReplyAndFree(redictContext *context, redictReply *reply, int type) {
if (reply == NULL)
panicAbort("NULL reply from server (error: %s)", context->errstr);
if (reply->type != type) {
if (reply->type == REDICT_REPLY_ERROR)
fprintf(stderr, "Redict Error: %s\n", reply->str);
panicAbort("Expected reply type %d but got type %d", type, reply->type);
}
freeReplyObject(reply);
}
/* Switch to the RESP3 protocol and enable client tracking */
static void enableClientTracking(redictContext *c) {
redictReply *reply = redictCommand(c, "HELLO 3");
if (reply == NULL || c->err) {
panicAbort("NULL reply or server error (error: %s)", c->errstr);
}
if (reply->type != REDICT_REPLY_MAP) {
fprintf(stderr, "Error: Can't send HELLO 3 command. Are you sure you're ");
fprintf(stderr, "connected to redict-server >= 6.0.0?\nRedict error: %s\n",
reply->type == REDICT_REPLY_ERROR ? reply->str : "(unknown)");
exit(-1);
}
freeReplyObject(reply);
/* Enable client tracking */
reply = redictCommand(c, "CLIENT TRACKING ON");
assertReplyAndFree(c, reply, REDICT_REPLY_STATUS);
}
void pushReplyHandler(void *privdata, void *r) {
redictReply *reply = r;
int *invalidations = privdata;
/* Sanity check on the invalidation reply */
if (reply->type != REDICT_REPLY_PUSH || reply->elements != 2 ||
reply->element[1]->type != REDICT_REPLY_ARRAY ||
reply->element[1]->element[0]->type != REDICT_REPLY_STRING)
{
panicAbort("%s", "Can't parse PUSH message!");
}
/* Increment our invalidation count */
*invalidations += 1;
printf("pushReplyHandler(): INVALIDATE '%s' (invalidation count: %d)\n",
reply->element[1]->element[0]->str, *invalidations);
freeReplyObject(reply);
}
/* We aren't actually freeing anything here, but it is included to show that we can
* have hiredict call our data destructor when freeing the context */
void privdata_dtor(void *privdata) {
unsigned int *icount = privdata;
printf("privdata_dtor(): In context privdata dtor (invalidations: %u)\n", *icount);
}
int main(int argc, char **argv) {
unsigned int j, invalidations = 0;
redictContext *c;
redictReply *reply;
const char *hostname = (argc > 1) ? argv[1] : "127.0.0.1";
int port = (argc > 2) ? atoi(argv[2]) : 6379;
redictOptions o = {0};
REDICT_OPTIONS_SET_TCP(&o, hostname, port);
/* Set our context privdata to the address of our invalidation counter. Each
* time our PUSH handler is called, hiredict will pass the privdata for context.
*
* This could also be done after we create the context like so:
*
* c->privdata = &invalidations;
* c->free_privdata = privdata_dtor;
*/
REDICT_OPTIONS_SET_PRIVDATA(&o, &invalidations, privdata_dtor);
/* Set our custom PUSH message handler */
o.push_cb = pushReplyHandler;
c = redictConnectWithOptions(&o);
if (c == NULL || c->err)
panicAbort("Connection error: %s", c ? c->errstr : "OOM");
/* Enable RESP3 and turn on client tracking */
enableClientTracking(c);
/* Set some keys and then read them back. Once we do that, Redict will deliver
* invalidation push messages whenever the key is modified */
for (j = 0; j < KEY_COUNT; j++) {
reply = redictCommand(c, "SET key:%d initial:%d", j, j);
assertReplyAndFree(c, reply, REDICT_REPLY_STATUS);
reply = redictCommand(c, "GET key:%d", j);
assertReplyAndFree(c, reply, REDICT_REPLY_STRING);
}
/* Trigger invalidation messages by updating keys we just read */
for (j = 0; j < KEY_COUNT; j++) {
printf(" main(): SET key:%d update:%d\n", j, j);
reply = redictCommand(c, "SET key:%d update:%d", j, j);
assertReplyAndFree(c, reply, REDICT_REPLY_STATUS);
printf(" main(): SET REPLY OK\n");
}
printf("\nTotal detected invalidations: %d, expected: %d\n", invalidations, KEY_COUNT);
/* PING server */
redictFree(c);
}
|
