Unverified Commit 3f073b1d authored by Yossi Gottlieb's avatar Yossi Gottlieb Committed by GitHub
Browse files

Merge pull request #7609 from michael-grunder/redis-cli-resp3-push

Add redis-cli RESP3 Push support
parents 3f494cc4 81879bc1
...@@ -33,7 +33,9 @@ void disconnectCallback(const redisAsyncContext *c, int status) { ...@@ -33,7 +33,9 @@ void disconnectCallback(const redisAsyncContext *c, int status) {
} }
int main (int argc, char **argv) { int main (int argc, char **argv) {
#ifndef _WIN32
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
#endif
iv_init(); iv_init();
......
...@@ -33,7 +33,9 @@ void disconnectCallback(const redisAsyncContext *c, int status) { ...@@ -33,7 +33,9 @@ void disconnectCallback(const redisAsyncContext *c, int status) {
} }
int main (int argc, char **argv) { int main (int argc, char **argv) {
#ifndef _WIN32
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
#endif
redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379); redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
if (c->err) { if (c->err) {
......
...@@ -34,7 +34,10 @@ void disconnectCallback(const redisAsyncContext *c, int status) { ...@@ -34,7 +34,10 @@ void disconnectCallback(const redisAsyncContext *c, int status) {
} }
int main (int argc, char **argv) { int main (int argc, char **argv) {
#ifndef _WIN32
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
#endif
struct event_base *base = event_base_new(); struct event_base *base = event_base_new();
if (argc < 5) { if (argc < 5) {
fprintf(stderr, fprintf(stderr,
...@@ -52,13 +55,25 @@ int main (int argc, char **argv) { ...@@ -52,13 +55,25 @@ int main (int argc, char **argv) {
const char *certKey = argv[5]; const char *certKey = argv[5];
const char *caCert = argc > 5 ? argv[6] : NULL; const char *caCert = argc > 5 ? argv[6] : NULL;
redisSSLContext *ssl;
redisSSLContextError ssl_error;
redisInitOpenSSL();
ssl = redisCreateSSLContext(caCert, NULL,
cert, certKey, NULL, &ssl_error);
if (!ssl) {
printf("Error: %s\n", redisSSLContextGetError(ssl_error));
return 1;
}
redisAsyncContext *c = redisAsyncConnect(hostname, port); redisAsyncContext *c = redisAsyncConnect(hostname, port);
if (c->err) { if (c->err) {
/* Let *c leak for now... */ /* Let *c leak for now... */
printf("Error: %s\n", c->errstr); printf("Error: %s\n", c->errstr);
return 1; return 1;
} }
if (redisSecureConnection(&c->c, caCert, cert, certKey, "sni") != REDIS_OK) { if (redisInitiateSSLWithContext(&c->c, ssl) != REDIS_OK) {
printf("SSL Error!\n"); printf("SSL Error!\n");
exit(1); exit(1);
} }
...@@ -69,5 +84,7 @@ int main (int argc, char **argv) { ...@@ -69,5 +84,7 @@ int main (int argc, char **argv) {
redisAsyncCommand(c, NULL, NULL, "SET key %b", value, nvalue); redisAsyncCommand(c, NULL, NULL, "SET key %b", value, nvalue);
redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key"); redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
event_base_dispatch(base); event_base_dispatch(base);
redisFreeSSLContext(ssl);
return 0; return 0;
} }
...@@ -38,13 +38,16 @@ void disconnectCallback(const redisAsyncContext *c, int status) { ...@@ -38,13 +38,16 @@ void disconnectCallback(const redisAsyncContext *c, int status) {
} }
int main (int argc, char **argv) { int main (int argc, char **argv) {
#ifndef _WIN32
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
#endif
struct event_base *base = event_base_new(); struct event_base *base = event_base_new();
redisOptions options = {0}; redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, "127.0.0.1", 6379); REDIS_OPTIONS_SET_TCP(&options, "127.0.0.1", 6379);
struct timeval tv = {0}; struct timeval tv = {0};
tv.tv_sec = 1; tv.tv_sec = 1;
options.timeout = &tv; options.connect_timeout = &tv;
redisAsyncContext *c = redisAsyncConnectWithOptions(&options); redisAsyncContext *c = redisAsyncConnectWithOptions(&options);
......
...@@ -33,7 +33,10 @@ void disconnectCallback(const redisAsyncContext *c, int status) { ...@@ -33,7 +33,10 @@ void disconnectCallback(const redisAsyncContext *c, int status) {
} }
int main (int argc, char **argv) { int main (int argc, char **argv) {
#ifndef _WIN32
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
#endif
uv_loop_t* loop = uv_default_loop(); uv_loop_t* loop = uv_default_loop();
redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379); redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
......
/*
* Copyright (c) 2020, Michael Grunder <michael dot grunder at gmail dot com>
*
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <hiredis.h>
#include <win32.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(redisContext *context, redisReply *reply, int type) {
if (reply == NULL)
panicAbort("NULL reply from server (error: %s)", context->errstr);
if (reply->type != type) {
if (reply->type == REDIS_REPLY_ERROR)
fprintf(stderr, "Redis 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(redisContext *c) {
redisReply *reply = redisCommand(c, "HELLO 3");
if (reply == NULL || c->err) {
panicAbort("NULL reply or server error (error: %s)", c->errstr);
}
if (reply->type != REDIS_REPLY_MAP) {
fprintf(stderr, "Error: Can't send HELLO 3 command. Are you sure you're ");
fprintf(stderr, "connected to redis-server >= 6.0.0?\nRedis error: %s\n",
reply->type == REDIS_REPLY_ERROR ? reply->str : "(unknown)");
exit(-1);
}
freeReplyObject(reply);
/* Enable client tracking */
reply = redisCommand(c, "CLIENT TRACKING ON");
assertReplyAndFree(c, reply, REDIS_REPLY_STATUS);
}
void pushReplyHandler(void *privdata, void *r) {
redisReply *reply = r;
int *invalidations = privdata;
/* Sanity check on the invalidation reply */
if (reply->type != REDIS_REPLY_PUSH || reply->elements != 2 ||
reply->element[1]->type != REDIS_REPLY_ARRAY ||
reply->element[1]->element[0]->type != REDIS_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 hiredis 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;
redisContext *c;
redisReply *reply;
const char *hostname = (argc > 1) ? argv[1] : "127.0.0.1";
int port = (argc > 2) ? atoi(argv[2]) : 6379;
redisOptions o = {0};
REDIS_OPTIONS_SET_TCP(&o, hostname, port);
/* Set our context privdata to the address of our invalidation counter. Each
* time our PUSH handler is called, hiredis 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;
*/
REDIS_OPTIONS_SET_PRIVDATA(&o, &invalidations, privdata_dtor);
/* Set our custom PUSH message handler */
o.push_cb = pushReplyHandler;
c = redisConnectWithOptions(&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, Redis will deliver
* invalidation push messages whenever the key is modified */
for (j = 0; j < KEY_COUNT; j++) {
reply = redisCommand(c, "SET key:%d initial:%d", j, j);
assertReplyAndFree(c, reply, REDIS_REPLY_STATUS);
reply = redisCommand(c, "GET key:%d", j);
assertReplyAndFree(c, reply, REDIS_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 = redisCommand(c, "SET key:%d update:%d", j, j);
assertReplyAndFree(c, reply, REDIS_REPLY_STATUS);
printf(" main(): SET REPLY OK\n");
}
printf("\nTotal detected invalidations: %d, expected: %d\n", invalidations, KEY_COUNT);
/* PING server */
redisFree(c);
}
...@@ -4,9 +4,12 @@ ...@@ -4,9 +4,12 @@
#include <hiredis.h> #include <hiredis.h>
#include <hiredis_ssl.h> #include <hiredis_ssl.h>
#include <win32.h>
int main(int argc, char **argv) { int main(int argc, char **argv) {
unsigned int j; unsigned int j;
redisSSLContext *ssl;
redisSSLContextError ssl_error;
redisContext *c; redisContext *c;
redisReply *reply; redisReply *reply;
if (argc < 4) { if (argc < 4) {
...@@ -19,10 +22,18 @@ int main(int argc, char **argv) { ...@@ -19,10 +22,18 @@ int main(int argc, char **argv) {
const char *key = argv[4]; const char *key = argv[4];
const char *ca = argc > 4 ? argv[5] : NULL; const char *ca = argc > 4 ? argv[5] : NULL;
redisInitOpenSSL();
ssl = redisCreateSSLContext(ca, NULL, cert, key, NULL, &ssl_error);
if (!ssl) {
printf("SSL Context error: %s\n",
redisSSLContextGetError(ssl_error));
exit(1);
}
struct timeval tv = { 1, 500000 }; // 1.5 seconds struct timeval tv = { 1, 500000 }; // 1.5 seconds
redisOptions options = {0}; redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, hostname, port); REDIS_OPTIONS_SET_TCP(&options, hostname, port);
options.timeout = &tv; options.connect_timeout = &tv;
c = redisConnectWithOptions(&options); c = redisConnectWithOptions(&options);
if (c == NULL || c->err) { if (c == NULL || c->err) {
...@@ -35,7 +46,7 @@ int main(int argc, char **argv) { ...@@ -35,7 +46,7 @@ int main(int argc, char **argv) {
exit(1); exit(1);
} }
if (redisSecureConnection(c, ca, cert, key, "sni") != REDIS_OK) { if (redisInitiateSSLWithContext(c, ssl) != REDIS_OK) {
printf("Couldn't initialize SSL!\n"); printf("Couldn't initialize SSL!\n");
printf("Error: %s\n", c->errstr); printf("Error: %s\n", c->errstr);
redisFree(c); redisFree(c);
...@@ -93,5 +104,7 @@ int main(int argc, char **argv) { ...@@ -93,5 +104,7 @@ int main(int argc, char **argv) {
/* Disconnects and frees the context */ /* Disconnects and frees the context */
redisFree(c); redisFree(c);
redisFreeSSLContext(ssl);
return 0; return 0;
} }
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <hiredis.h> #include <hiredis.h>
#include <win32.h>
int main(int argc, char **argv) { int main(int argc, char **argv) {
unsigned int j, isunix = 0; unsigned int j, isunix = 0;
......
@PACKAGE_INIT@
set_and_check(hiredis_INCLUDEDIR "@PACKAGE_INCLUDE_INSTALL_DIR@")
IF (NOT TARGET hiredis::hiredis)
INCLUDE(${CMAKE_CURRENT_LIST_DIR}/hiredis-targets.cmake)
ENDIF()
SET(hiredis_LIBRARIES hiredis::hiredis)
SET(hiredis_INCLUDE_DIRS ${hiredis_INCLUDEDIR})
check_required_components(hiredis)
...@@ -44,8 +44,11 @@ ...@@ -44,8 +44,11 @@
#include "async.h" #include "async.h"
#include "win32.h" #include "win32.h"
extern int redisContextUpdateConnectTimeout(redisContext *c, const struct timeval *timeout);
extern int redisContextUpdateCommandTimeout(redisContext *c, const struct timeval *timeout);
static redisContextFuncs redisContextDefaultFuncs = { static redisContextFuncs redisContextDefaultFuncs = {
.free_privdata = NULL, .free_privctx = NULL,
.async_read = redisAsyncRead, .async_read = redisAsyncRead,
.async_write = redisAsyncWrite, .async_write = redisAsyncWrite,
.read = redisNetRead, .read = redisNetRead,
...@@ -74,7 +77,7 @@ static redisReplyObjectFunctions defaultFunctions = { ...@@ -74,7 +77,7 @@ static redisReplyObjectFunctions defaultFunctions = {
/* Create a reply object */ /* Create a reply object */
static redisReply *createReplyObject(int type) { static redisReply *createReplyObject(int type) {
redisReply *r = calloc(1,sizeof(*r)); redisReply *r = hi_calloc(1,sizeof(*r));
if (r == NULL) if (r == NULL)
return NULL; return NULL;
...@@ -97,20 +100,22 @@ void freeReplyObject(void *reply) { ...@@ -97,20 +100,22 @@ void freeReplyObject(void *reply) {
case REDIS_REPLY_ARRAY: case REDIS_REPLY_ARRAY:
case REDIS_REPLY_MAP: case REDIS_REPLY_MAP:
case REDIS_REPLY_SET: case REDIS_REPLY_SET:
case REDIS_REPLY_PUSH:
if (r->element != NULL) { if (r->element != NULL) {
for (j = 0; j < r->elements; j++) for (j = 0; j < r->elements; j++)
freeReplyObject(r->element[j]); freeReplyObject(r->element[j]);
free(r->element); hi_free(r->element);
} }
break; break;
case REDIS_REPLY_ERROR: case REDIS_REPLY_ERROR:
case REDIS_REPLY_STATUS: case REDIS_REPLY_STATUS:
case REDIS_REPLY_STRING: case REDIS_REPLY_STRING:
case REDIS_REPLY_DOUBLE: case REDIS_REPLY_DOUBLE:
free(r->str); case REDIS_REPLY_VERB:
hi_free(r->str);
break; break;
} }
free(r); hi_free(r);
} }
static void *createStringObject(const redisReadTask *task, char *str, size_t len) { static void *createStringObject(const redisReadTask *task, char *str, size_t len) {
...@@ -128,22 +133,18 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len ...@@ -128,22 +133,18 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len
/* Copy string value */ /* Copy string value */
if (task->type == REDIS_REPLY_VERB) { if (task->type == REDIS_REPLY_VERB) {
buf = malloc(len-4+1); /* Skip 4 bytes of verbatim type header. */ buf = hi_malloc(len-4+1); /* Skip 4 bytes of verbatim type header. */
if (buf == NULL) { if (buf == NULL) goto oom;
freeReplyObject(r);
return NULL;
}
memcpy(r->vtype,str,3); memcpy(r->vtype,str,3);
r->vtype[3] = '\0'; r->vtype[3] = '\0';
memcpy(buf,str+4,len-4); memcpy(buf,str+4,len-4);
buf[len-4] = '\0'; buf[len-4] = '\0';
r->len = len-4; r->len = len - 4;
} else { } else {
buf = malloc(len+1); buf = hi_malloc(len+1);
if (buf == NULL) { if (buf == NULL) goto oom;
freeReplyObject(r);
return NULL;
}
memcpy(buf,str,len); memcpy(buf,str,len);
buf[len] = '\0'; buf[len] = '\0';
r->len = len; r->len = len;
...@@ -154,10 +155,15 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len ...@@ -154,10 +155,15 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len
parent = task->parent->obj; parent = task->parent->obj;
assert(parent->type == REDIS_REPLY_ARRAY || assert(parent->type == REDIS_REPLY_ARRAY ||
parent->type == REDIS_REPLY_MAP || parent->type == REDIS_REPLY_MAP ||
parent->type == REDIS_REPLY_SET); parent->type == REDIS_REPLY_SET ||
parent->type == REDIS_REPLY_PUSH);
parent->element[task->idx] = r; parent->element[task->idx] = r;
} }
return r; return r;
oom:
freeReplyObject(r);
return NULL;
} }
static void *createArrayObject(const redisReadTask *task, size_t elements) { static void *createArrayObject(const redisReadTask *task, size_t elements) {
...@@ -168,7 +174,7 @@ static void *createArrayObject(const redisReadTask *task, size_t elements) { ...@@ -168,7 +174,7 @@ static void *createArrayObject(const redisReadTask *task, size_t elements) {
return NULL; return NULL;
if (elements > 0) { if (elements > 0) {
r->element = calloc(elements,sizeof(redisReply*)); r->element = hi_calloc(elements,sizeof(redisReply*));
if (r->element == NULL) { if (r->element == NULL) {
freeReplyObject(r); freeReplyObject(r);
return NULL; return NULL;
...@@ -181,7 +187,8 @@ static void *createArrayObject(const redisReadTask *task, size_t elements) { ...@@ -181,7 +187,8 @@ static void *createArrayObject(const redisReadTask *task, size_t elements) {
parent = task->parent->obj; parent = task->parent->obj;
assert(parent->type == REDIS_REPLY_ARRAY || assert(parent->type == REDIS_REPLY_ARRAY ||
parent->type == REDIS_REPLY_MAP || parent->type == REDIS_REPLY_MAP ||
parent->type == REDIS_REPLY_SET); parent->type == REDIS_REPLY_SET ||
parent->type == REDIS_REPLY_PUSH);
parent->element[task->idx] = r; parent->element[task->idx] = r;
} }
return r; return r;
...@@ -200,7 +207,8 @@ static void *createIntegerObject(const redisReadTask *task, long long value) { ...@@ -200,7 +207,8 @@ static void *createIntegerObject(const redisReadTask *task, long long value) {
parent = task->parent->obj; parent = task->parent->obj;
assert(parent->type == REDIS_REPLY_ARRAY || assert(parent->type == REDIS_REPLY_ARRAY ||
parent->type == REDIS_REPLY_MAP || parent->type == REDIS_REPLY_MAP ||
parent->type == REDIS_REPLY_SET); parent->type == REDIS_REPLY_SET ||
parent->type == REDIS_REPLY_PUSH);
parent->element[task->idx] = r; parent->element[task->idx] = r;
} }
return r; return r;
...@@ -214,7 +222,7 @@ static void *createDoubleObject(const redisReadTask *task, double value, char *s ...@@ -214,7 +222,7 @@ static void *createDoubleObject(const redisReadTask *task, double value, char *s
return NULL; return NULL;
r->dval = value; r->dval = value;
r->str = malloc(len+1); r->str = hi_malloc(len+1);
if (r->str == NULL) { if (r->str == NULL) {
freeReplyObject(r); freeReplyObject(r);
return NULL; return NULL;
...@@ -318,7 +326,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -318,7 +326,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
if (*c != '%' || c[1] == '\0') { if (*c != '%' || c[1] == '\0') {
if (*c == ' ') { if (*c == ' ') {
if (touched) { if (touched) {
newargv = realloc(curargv,sizeof(char*)*(argc+1)); newargv = hi_realloc(curargv,sizeof(char*)*(argc+1));
if (newargv == NULL) goto memory_err; if (newargv == NULL) goto memory_err;
curargv = newargv; curargv = newargv;
curargv[argc++] = curarg; curargv[argc++] = curarg;
...@@ -467,7 +475,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -467,7 +475,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
/* Add the last argument if needed */ /* Add the last argument if needed */
if (touched) { if (touched) {
newargv = realloc(curargv,sizeof(char*)*(argc+1)); newargv = hi_realloc(curargv,sizeof(char*)*(argc+1));
if (newargv == NULL) goto memory_err; if (newargv == NULL) goto memory_err;
curargv = newargv; curargv = newargv;
curargv[argc++] = curarg; curargv[argc++] = curarg;
...@@ -483,7 +491,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -483,7 +491,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
totlen += 1+countDigits(argc)+2; totlen += 1+countDigits(argc)+2;
/* Build the command at protocol level */ /* Build the command at protocol level */
cmd = malloc(totlen+1); cmd = hi_malloc(totlen+1);
if (cmd == NULL) goto memory_err; if (cmd == NULL) goto memory_err;
pos = sprintf(cmd,"*%d\r\n",argc); pos = sprintf(cmd,"*%d\r\n",argc);
...@@ -498,7 +506,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -498,7 +506,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
assert(pos == totlen); assert(pos == totlen);
cmd[pos] = '\0'; cmd[pos] = '\0';
free(curargv); hi_free(curargv);
*target = cmd; *target = cmd;
return totlen; return totlen;
...@@ -514,11 +522,11 @@ cleanup: ...@@ -514,11 +522,11 @@ cleanup:
if (curargv) { if (curargv) {
while(argc--) while(argc--)
sdsfree(curargv[argc]); sdsfree(curargv[argc]);
free(curargv); hi_free(curargv);
} }
sdsfree(curarg); sdsfree(curarg);
free(cmd); hi_free(cmd);
return error_type; return error_type;
} }
...@@ -559,7 +567,7 @@ int redisFormatCommand(char **target, const char *format, ...) { ...@@ -559,7 +567,7 @@ int redisFormatCommand(char **target, const char *format, ...) {
int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv, int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv,
const size_t *argvlen) const size_t *argvlen)
{ {
sds cmd; sds cmd, aux;
unsigned long long totlen; unsigned long long totlen;
int j; int j;
size_t len; size_t len;
...@@ -581,9 +589,13 @@ int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv, ...@@ -581,9 +589,13 @@ int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv,
return -1; return -1;
/* We already know how much storage we need */ /* We already know how much storage we need */
cmd = sdsMakeRoomFor(cmd, totlen); aux = sdsMakeRoomFor(cmd, totlen);
if (cmd == NULL) if (aux == NULL) {
sdsfree(cmd);
return -1; return -1;
}
cmd = aux;
/* Construct command */ /* Construct command */
cmd = sdscatfmt(cmd, "*%i\r\n", argc); cmd = sdscatfmt(cmd, "*%i\r\n", argc);
...@@ -627,7 +639,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz ...@@ -627,7 +639,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz
} }
/* Build the command at protocol level */ /* Build the command at protocol level */
cmd = malloc(totlen+1); cmd = hi_malloc(totlen+1);
if (cmd == NULL) if (cmd == NULL)
return -1; return -1;
...@@ -648,7 +660,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz ...@@ -648,7 +660,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz
} }
void redisFreeCommand(char *cmd) { void redisFreeCommand(char *cmd) {
free(cmd); hi_free(cmd);
} }
void __redisSetError(redisContext *c, int type, const char *str) { void __redisSetError(redisContext *c, int type, const char *str) {
...@@ -671,14 +683,20 @@ redisReader *redisReaderCreate(void) { ...@@ -671,14 +683,20 @@ redisReader *redisReaderCreate(void) {
return redisReaderCreateWithFunctions(&defaultFunctions); return redisReaderCreateWithFunctions(&defaultFunctions);
} }
static redisContext *redisContextInit(const redisOptions *options) { static void redisPushAutoFree(void *privdata, void *reply) {
(void)privdata;
freeReplyObject(reply);
}
static redisContext *redisContextInit(void) {
redisContext *c; redisContext *c;
c = calloc(1, sizeof(*c)); c = hi_calloc(1, sizeof(*c));
if (c == NULL) if (c == NULL)
return NULL; return NULL;
c->funcs = &redisContextDefaultFuncs; c->funcs = &redisContextDefaultFuncs;
c->obuf = sdsempty(); c->obuf = sdsempty();
c->reader = redisReaderCreate(); c->reader = redisReaderCreate();
c->fd = REDIS_INVALID_FD; c->fd = REDIS_INVALID_FD;
...@@ -687,7 +705,7 @@ static redisContext *redisContextInit(const redisOptions *options) { ...@@ -687,7 +705,7 @@ static redisContext *redisContextInit(const redisOptions *options) {
redisFree(c); redisFree(c);
return NULL; return NULL;
} }
(void)options; /* options are used in other functions */
return c; return c;
} }
...@@ -698,16 +716,21 @@ void redisFree(redisContext *c) { ...@@ -698,16 +716,21 @@ void redisFree(redisContext *c) {
sdsfree(c->obuf); sdsfree(c->obuf);
redisReaderFree(c->reader); redisReaderFree(c->reader);
free(c->tcp.host); hi_free(c->tcp.host);
free(c->tcp.source_addr); hi_free(c->tcp.source_addr);
free(c->unix_sock.path); hi_free(c->unix_sock.path);
free(c->timeout); hi_free(c->connect_timeout);
free(c->saddr); hi_free(c->command_timeout);
if (c->funcs->free_privdata) { hi_free(c->saddr);
c->funcs->free_privdata(c->privdata);
} if (c->privdata && c->free_privdata)
c->free_privdata(c->privdata);
if (c->funcs->free_privctx)
c->funcs->free_privctx(c->privctx);
memset(c, 0xff, sizeof(*c)); memset(c, 0xff, sizeof(*c));
free(c); hi_free(c);
} }
redisFD redisFreeKeepFd(redisContext *c) { redisFD redisFreeKeepFd(redisContext *c) {
...@@ -721,9 +744,9 @@ int redisReconnect(redisContext *c) { ...@@ -721,9 +744,9 @@ int redisReconnect(redisContext *c) {
c->err = 0; c->err = 0;
memset(c->errstr, '\0', strlen(c->errstr)); memset(c->errstr, '\0', strlen(c->errstr));
if (c->privdata && c->funcs->free_privdata) { if (c->privctx && c->funcs->free_privctx) {
c->funcs->free_privdata(c->privdata); c->funcs->free_privctx(c->privctx);
c->privdata = NULL; c->privctx = NULL;
} }
redisNetClose(c); redisNetClose(c);
...@@ -734,22 +757,33 @@ int redisReconnect(redisContext *c) { ...@@ -734,22 +757,33 @@ int redisReconnect(redisContext *c) {
c->obuf = sdsempty(); c->obuf = sdsempty();
c->reader = redisReaderCreate(); c->reader = redisReaderCreate();
if (c->obuf == NULL || c->reader == NULL) {
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
}
int ret = REDIS_ERR;
if (c->connection_type == REDIS_CONN_TCP) { if (c->connection_type == REDIS_CONN_TCP) {
return redisContextConnectBindTcp(c, c->tcp.host, c->tcp.port, ret = redisContextConnectBindTcp(c, c->tcp.host, c->tcp.port,
c->timeout, c->tcp.source_addr); c->connect_timeout, c->tcp.source_addr);
} else if (c->connection_type == REDIS_CONN_UNIX) { } else if (c->connection_type == REDIS_CONN_UNIX) {
return redisContextConnectUnix(c, c->unix_sock.path, c->timeout); ret = redisContextConnectUnix(c, c->unix_sock.path, c->connect_timeout);
} else { } else {
/* Something bad happened here and shouldn't have. There isn't /* Something bad happened here and shouldn't have. There isn't
enough information in the context to reconnect. */ enough information in the context to reconnect. */
__redisSetError(c,REDIS_ERR_OTHER,"Not enough information to reconnect"); __redisSetError(c,REDIS_ERR_OTHER,"Not enough information to reconnect");
ret = REDIS_ERR;
} }
return REDIS_ERR; if (c->command_timeout != NULL && (c->flags & REDIS_BLOCK) && c->fd != REDIS_INVALID_FD) {
redisContextSetTimeout(c, *c->command_timeout);
}
return ret;
} }
redisContext *redisConnectWithOptions(const redisOptions *options) { redisContext *redisConnectWithOptions(const redisOptions *options) {
redisContext *c = redisContextInit(options); redisContext *c = redisContextInit();
if (c == NULL) { if (c == NULL) {
return NULL; return NULL;
} }
...@@ -763,13 +797,29 @@ redisContext *redisConnectWithOptions(const redisOptions *options) { ...@@ -763,13 +797,29 @@ redisContext *redisConnectWithOptions(const redisOptions *options) {
c->flags |= REDIS_NO_AUTO_FREE; c->flags |= REDIS_NO_AUTO_FREE;
} }
/* Set any user supplied RESP3 PUSH handler or use freeReplyObject
* as a default unless specifically flagged that we don't want one. */
if (options->push_cb != NULL)
redisSetPushCallback(c, options->push_cb);
else if (!(options->options & REDIS_OPT_NO_PUSH_AUTOFREE))
redisSetPushCallback(c, redisPushAutoFree);
c->privdata = options->privdata;
c->free_privdata = options->free_privdata;
if (redisContextUpdateConnectTimeout(c, options->connect_timeout) != REDIS_OK ||
redisContextUpdateCommandTimeout(c, options->command_timeout) != REDIS_OK) {
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return c;
}
if (options->type == REDIS_CONN_TCP) { if (options->type == REDIS_CONN_TCP) {
redisContextConnectBindTcp(c, options->endpoint.tcp.ip, redisContextConnectBindTcp(c, options->endpoint.tcp.ip,
options->endpoint.tcp.port, options->timeout, options->endpoint.tcp.port, options->connect_timeout,
options->endpoint.tcp.source_addr); options->endpoint.tcp.source_addr);
} else if (options->type == REDIS_CONN_UNIX) { } else if (options->type == REDIS_CONN_UNIX) {
redisContextConnectUnix(c, options->endpoint.unix_socket, redisContextConnectUnix(c, options->endpoint.unix_socket,
options->timeout); options->connect_timeout);
} else if (options->type == REDIS_CONN_USERFD) { } else if (options->type == REDIS_CONN_USERFD) {
c->fd = options->endpoint.fd; c->fd = options->endpoint.fd;
c->flags |= REDIS_CONNECTED; c->flags |= REDIS_CONNECTED;
...@@ -777,9 +827,11 @@ redisContext *redisConnectWithOptions(const redisOptions *options) { ...@@ -777,9 +827,11 @@ redisContext *redisConnectWithOptions(const redisOptions *options) {
// Unknown type - FIXME - FREE // Unknown type - FIXME - FREE
return NULL; return NULL;
} }
if (options->timeout != NULL && (c->flags & REDIS_BLOCK) && c->fd != REDIS_INVALID_FD) {
redisContextSetTimeout(c, *options->timeout); if (options->command_timeout != NULL && (c->flags & REDIS_BLOCK) && c->fd != REDIS_INVALID_FD) {
redisContextSetTimeout(c, *options->command_timeout);
} }
return c; return c;
} }
...@@ -795,7 +847,7 @@ redisContext *redisConnect(const char *ip, int port) { ...@@ -795,7 +847,7 @@ redisContext *redisConnect(const char *ip, int port) {
redisContext *redisConnectWithTimeout(const char *ip, int port, const struct timeval tv) { redisContext *redisConnectWithTimeout(const char *ip, int port, const struct timeval tv) {
redisOptions options = {0}; redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, ip, port); REDIS_OPTIONS_SET_TCP(&options, ip, port);
options.timeout = &tv; options.connect_timeout = &tv;
return redisConnectWithOptions(&options); return redisConnectWithOptions(&options);
} }
...@@ -833,7 +885,7 @@ redisContext *redisConnectUnix(const char *path) { ...@@ -833,7 +885,7 @@ redisContext *redisConnectUnix(const char *path) {
redisContext *redisConnectUnixWithTimeout(const char *path, const struct timeval tv) { redisContext *redisConnectUnixWithTimeout(const char *path, const struct timeval tv) {
redisOptions options = {0}; redisOptions options = {0};
REDIS_OPTIONS_SET_UNIX(&options, path); REDIS_OPTIONS_SET_UNIX(&options, path);
options.timeout = &tv; options.connect_timeout = &tv;
return redisConnectWithOptions(&options); return redisConnectWithOptions(&options);
} }
...@@ -865,6 +917,13 @@ int redisEnableKeepAlive(redisContext *c) { ...@@ -865,6 +917,13 @@ int redisEnableKeepAlive(redisContext *c) {
return REDIS_OK; return REDIS_OK;
} }
/* Set a user provided RESP3 PUSH handler and return any old one set. */
redisPushFn *redisSetPushCallback(redisContext *c, redisPushFn *fn) {
redisPushFn *old = c->push_cb;
c->push_cb = fn;
return old;
}
/* Use this function to handle a read event on the descriptor. It will try /* Use this function to handle a read event on the descriptor. It will try
* and read some bytes from the socket and feed them to the reply parser. * and read some bytes from the socket and feed them to the reply parser.
* *
...@@ -907,20 +966,26 @@ int redisBufferWrite(redisContext *c, int *done) { ...@@ -907,20 +966,26 @@ int redisBufferWrite(redisContext *c, int *done) {
return REDIS_ERR; return REDIS_ERR;
if (sdslen(c->obuf) > 0) { if (sdslen(c->obuf) > 0) {
int nwritten = c->funcs->write(c); ssize_t nwritten = c->funcs->write(c);
if (nwritten < 0) { if (nwritten < 0) {
return REDIS_ERR; return REDIS_ERR;
} else if (nwritten > 0) { } else if (nwritten > 0) {
if (nwritten == (signed)sdslen(c->obuf)) { if (nwritten == (ssize_t)sdslen(c->obuf)) {
sdsfree(c->obuf); sdsfree(c->obuf);
c->obuf = sdsempty(); c->obuf = sdsempty();
if (c->obuf == NULL)
goto oom;
} else { } else {
sdsrange(c->obuf,nwritten,-1); if (sdsrange(c->obuf,nwritten,-1) < 0) goto oom;
} }
} }
} }
if (done != NULL) *done = (sdslen(c->obuf) == 0); if (done != NULL) *done = (sdslen(c->obuf) == 0);
return REDIS_OK; return REDIS_OK;
oom:
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
} }
/* Internal helper function to try and get a reply from the reader, /* Internal helper function to try and get a reply from the reader,
...@@ -930,9 +995,21 @@ int redisGetReplyFromReader(redisContext *c, void **reply) { ...@@ -930,9 +995,21 @@ int redisGetReplyFromReader(redisContext *c, void **reply) {
__redisSetError(c,c->reader->err,c->reader->errstr); __redisSetError(c,c->reader->err,c->reader->errstr);
return REDIS_ERR; return REDIS_ERR;
} }
return REDIS_OK; return REDIS_OK;
} }
/* Internal helper that returns 1 if the reply was a RESP3 PUSH
* message and we handled it with a user-provided callback. */
static int redisHandledPushReply(redisContext *c, void *reply) {
if (reply && c->push_cb && redisIsPushReply(reply)) {
c->push_cb(c->privdata, reply);
return 1;
}
return 0;
}
int redisGetReply(redisContext *c, void **reply) { int redisGetReply(redisContext *c, void **reply) {
int wdone = 0; int wdone = 0;
void *aux = NULL; void *aux = NULL;
...@@ -953,13 +1030,23 @@ int redisGetReply(redisContext *c, void **reply) { ...@@ -953,13 +1030,23 @@ int redisGetReply(redisContext *c, void **reply) {
do { do {
if (redisBufferRead(c) == REDIS_ERR) if (redisBufferRead(c) == REDIS_ERR)
return REDIS_ERR; return REDIS_ERR;
/* We loop here in case the user has specified a RESP3
* PUSH handler (e.g. for client tracking). */
do {
if (redisGetReplyFromReader(c,&aux) == REDIS_ERR) if (redisGetReplyFromReader(c,&aux) == REDIS_ERR)
return REDIS_ERR; return REDIS_ERR;
} while (redisHandledPushReply(c, aux));
} while (aux == NULL); } while (aux == NULL);
} }
/* Set reply object */ /* Set reply or free it if we were passed NULL */
if (reply != NULL) *reply = aux; if (reply != NULL) {
*reply = aux;
} else {
freeReplyObject(aux);
}
return REDIS_OK; return REDIS_OK;
} }
...@@ -1006,11 +1093,11 @@ int redisvAppendCommand(redisContext *c, const char *format, va_list ap) { ...@@ -1006,11 +1093,11 @@ int redisvAppendCommand(redisContext *c, const char *format, va_list ap) {
} }
if (__redisAppendCommand(c,cmd,len) != REDIS_OK) { if (__redisAppendCommand(c,cmd,len) != REDIS_OK) {
free(cmd); hi_free(cmd);
return REDIS_ERR; return REDIS_ERR;
} }
free(cmd); hi_free(cmd);
return REDIS_OK; return REDIS_OK;
} }
......
...@@ -39,14 +39,16 @@ ...@@ -39,14 +39,16 @@
#include <sys/time.h> /* for struct timeval */ #include <sys/time.h> /* for struct timeval */
#else #else
struct timeval; /* forward declaration */ struct timeval; /* forward declaration */
typedef long long ssize_t;
#endif #endif
#include <stdint.h> /* uintXX_t, etc */ #include <stdint.h> /* uintXX_t, etc */
#include "sds.h" /* for sds */ #include "sds.h" /* for sds */
#include "alloc.h" /* for allocation wrappers */
#define HIREDIS_MAJOR 0 #define HIREDIS_MAJOR 1
#define HIREDIS_MINOR 14 #define HIREDIS_MINOR 0
#define HIREDIS_PATCH 0 #define HIREDIS_PATCH 0
#define HIREDIS_SONAME 0.14 #define HIREDIS_SONAME 1.0.0
/* Connection type can be blocking or non-blocking and is set in the /* Connection type can be blocking or non-blocking and is set in the
* least significant bit of the flags field in redisContext. */ * least significant bit of the flags field in redisContext. */
...@@ -90,6 +92,15 @@ struct timeval; /* forward declaration */ ...@@ -90,6 +92,15 @@ struct timeval; /* forward declaration */
* SO_REUSEADDR is being used. */ * SO_REUSEADDR is being used. */
#define REDIS_CONNECT_RETRIES 10 #define REDIS_CONNECT_RETRIES 10
/* Forward declarations for structs defined elsewhere */
struct redisAsyncContext;
struct redisContext;
/* RESP3 push helpers and callback prototypes */
#define redisIsPushReply(r) (((redisReply*)(r))->type == REDIS_REPLY_PUSH)
typedef void (redisPushFn)(void *, void *);
typedef void (redisAsyncPushFn)(struct redisAsyncContext *, void *);
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
...@@ -101,7 +112,7 @@ typedef struct redisReply { ...@@ -101,7 +112,7 @@ typedef struct redisReply {
double dval; /* The double when type is REDIS_REPLY_DOUBLE */ double dval; /* The double when type is REDIS_REPLY_DOUBLE */
size_t len; /* Length of string */ size_t len; /* Length of string */
char *str; /* Used for REDIS_REPLY_ERROR, REDIS_REPLY_STRING char *str; /* Used for REDIS_REPLY_ERROR, REDIS_REPLY_STRING
and REDIS_REPLY_DOUBLE (in additionl to dval). */ REDIS_REPLY_VERB, and REDIS_REPLY_DOUBLE (in additional to dval). */
char vtype[4]; /* Used for REDIS_REPLY_VERB, contains the null char vtype[4]; /* Used for REDIS_REPLY_VERB, contains the null
terminated 3 character content type, such as "txt". */ terminated 3 character content type, such as "txt". */
size_t elements; /* number of elements, for REDIS_REPLY_ARRAY */ size_t elements; /* number of elements, for REDIS_REPLY_ARRAY */
...@@ -138,6 +149,9 @@ struct redisSsl; ...@@ -138,6 +149,9 @@ struct redisSsl;
*/ */
#define REDIS_OPT_NOAUTOFREE 0x04 #define REDIS_OPT_NOAUTOFREE 0x04
/* Don't automatically intercept and free RESP3 PUSH replies. */
#define REDIS_OPT_NO_PUSH_AUTOFREE 0x08
/* In Unix systems a file descriptor is a regular signed int, with -1 /* In Unix systems a file descriptor is a regular signed int, with -1
* representing an invalid descriptor. In Windows it is a SOCKET * representing an invalid descriptor. In Windows it is a SOCKET
* (32- or 64-bit unsigned integer depending on the architecture), where * (32- or 64-bit unsigned integer depending on the architecture), where
...@@ -162,8 +176,11 @@ typedef struct { ...@@ -162,8 +176,11 @@ typedef struct {
int type; int type;
/* bit field of REDIS_OPT_xxx */ /* bit field of REDIS_OPT_xxx */
int options; int options;
/* timeout value. if NULL, no timeout is used */ /* timeout value for connect operation. If NULL, no timeout is used */
const struct timeval *timeout; const struct timeval *connect_timeout;
/* timeout value for commands. If NULL, no timeout is used. This can be
* updated at runtime with redisSetTimeout/redisAsyncSetTimeout. */
const struct timeval *command_timeout;
union { union {
/** use this field for tcp/ip connections */ /** use this field for tcp/ip connections */
struct { struct {
...@@ -178,6 +195,14 @@ typedef struct { ...@@ -178,6 +195,14 @@ typedef struct {
* file descriptor */ * file descriptor */
redisFD fd; redisFD fd;
} endpoint; } endpoint;
/* Optional user defined data/destructor */
void *privdata;
void (*free_privdata)(void *);
/* A user defined PUSH message callback */
redisPushFn *push_cb;
redisAsyncPushFn *async_push_cb;
} redisOptions; } redisOptions;
/** /**
...@@ -192,15 +217,16 @@ typedef struct { ...@@ -192,15 +217,16 @@ typedef struct {
(opts)->type = REDIS_CONN_UNIX; \ (opts)->type = REDIS_CONN_UNIX; \
(opts)->endpoint.unix_socket = path; (opts)->endpoint.unix_socket = path;
struct redisAsyncContext; #define REDIS_OPTIONS_SET_PRIVDATA(opts, data, dtor) \
struct redisContext; (opts)->privdata = data; \
(opts)->free_privdata = dtor; \
typedef struct redisContextFuncs { typedef struct redisContextFuncs {
void (*free_privdata)(void *); void (*free_privctx)(void *);
void (*async_read)(struct redisAsyncContext *); void (*async_read)(struct redisAsyncContext *);
void (*async_write)(struct redisAsyncContext *); void (*async_write)(struct redisAsyncContext *);
int (*read)(struct redisContext *, char *, size_t); ssize_t (*read)(struct redisContext *, char *, size_t);
int (*write)(struct redisContext *); ssize_t (*write)(struct redisContext *);
} redisContextFuncs; } redisContextFuncs;
/* Context for a connection to Redis */ /* Context for a connection to Redis */
...@@ -215,7 +241,8 @@ typedef struct redisContext { ...@@ -215,7 +241,8 @@ typedef struct redisContext {
redisReader *reader; /* Protocol reader */ redisReader *reader; /* Protocol reader */
enum redisConnectionType connection_type; enum redisConnectionType connection_type;
struct timeval *timeout; struct timeval *connect_timeout;
struct timeval *command_timeout;
struct { struct {
char *host; char *host;
...@@ -231,8 +258,17 @@ typedef struct redisContext { ...@@ -231,8 +258,17 @@ typedef struct redisContext {
struct sockadr *saddr; struct sockadr *saddr;
size_t addrlen; size_t addrlen;
/* Additional private data for hiredis addons such as SSL */ /* Optional data and corresponding destructor users can use to provide
* context to a given redisContext. Not used by hiredis. */
void *privdata; void *privdata;
void (*free_privdata)(void *);
/* Internal context pointer presently used by hiredis to manage
* SSL connections. */
void *privctx;
/* An optional RESP3 PUSH handler */
redisPushFn *push_cb;
} redisContext; } redisContext;
redisContext *redisConnectWithOptions(const redisOptions *options); redisContext *redisConnectWithOptions(const redisOptions *options);
...@@ -259,6 +295,7 @@ redisContext *redisConnectFd(redisFD fd); ...@@ -259,6 +295,7 @@ redisContext *redisConnectFd(redisFD fd);
*/ */
int redisReconnect(redisContext *c); int redisReconnect(redisContext *c);
redisPushFn *redisSetPushCallback(redisContext *c, redisPushFn *fn);
int redisSetTimeout(redisContext *c, const struct timeval tv); int redisSetTimeout(redisContext *c, const struct timeval tv);
int redisEnableKeepAlive(redisContext *c); int redisEnableKeepAlive(redisContext *c);
void redisFree(redisContext *c); void redisFree(redisContext *c);
......
prefix=@CMAKE_INSTALL_PREFIX@ prefix=@CMAKE_INSTALL_PREFIX@
install_libdir=@CMAKE_INSTALL_LIBDIR@
exec_prefix=${prefix} exec_prefix=${prefix}
libdir=${exec_prefix}/lib libdir=${exec_prefix}/${install_libdir}
includedir=${prefix}/include includedir=${prefix}/include
pkgincludedir=${includedir}/hiredis pkgincludedir=${includedir}/hiredis
......
@PACKAGE_INIT@
set_and_check(hiredis_ssl_INCLUDEDIR "@PACKAGE_INCLUDE_INSTALL_DIR@")
IF (NOT TARGET hiredis::hiredis_ssl)
INCLUDE(${CMAKE_CURRENT_LIST_DIR}/hiredis_ssl-targets.cmake)
ENDIF()
SET(hiredis_ssl_LIBRARIES hiredis::hiredis_ssl)
SET(hiredis_ssl_INCLUDE_DIRS ${hiredis_ssl_INCLUDEDIR})
check_required_components(hiredis_ssl)
...@@ -32,22 +32,96 @@ ...@@ -32,22 +32,96 @@
#ifndef __HIREDIS_SSL_H #ifndef __HIREDIS_SSL_H
#define __HIREDIS_SSL_H #define __HIREDIS_SSL_H
#ifdef __cplusplus
extern "C" {
#endif
/* This is the underlying struct for SSL in ssl.h, which is not included to /* This is the underlying struct for SSL in ssl.h, which is not included to
* keep build dependencies short here. * keep build dependencies short here.
*/ */
struct ssl_st; struct ssl_st;
/* A wrapper around OpenSSL SSL_CTX to allow easy SSL use without directly
* calling OpenSSL.
*/
typedef struct redisSSLContext redisSSLContext;
/**
* Initialization errors that redisCreateSSLContext() may return.
*/
typedef enum {
REDIS_SSL_CTX_NONE = 0, /* No Error */
REDIS_SSL_CTX_CREATE_FAILED, /* Failed to create OpenSSL SSL_CTX */
REDIS_SSL_CTX_CERT_KEY_REQUIRED, /* Client cert and key must both be specified or skipped */
REDIS_SSL_CTX_CA_CERT_LOAD_FAILED, /* Failed to load CA Certificate or CA Path */
REDIS_SSL_CTX_CLIENT_CERT_LOAD_FAILED, /* Failed to load client certificate */
REDIS_SSL_CTX_PRIVATE_KEY_LOAD_FAILED /* Failed to load private key */
} redisSSLContextError;
/**
* Return the error message corresponding with the specified error code.
*/
const char *redisSSLContextGetError(redisSSLContextError error);
/**
* Helper function to initialize the OpenSSL library.
*
* OpenSSL requires one-time initialization before it can be used. Callers should
* call this function only once, and only if OpenSSL is not directly initialized
* elsewhere.
*/
int redisInitOpenSSL(void);
/** /**
* Secure the connection using SSL. This should be done before any command is * Helper function to initialize an OpenSSL context that can be used
* executed on the connection. * to initiate SSL connections.
*
* cacert_filename is an optional name of a CA certificate/bundle file to load
* and use for validation.
*
* capath is an optional directory path where trusted CA certificate files are
* stored in an OpenSSL-compatible structure.
*
* cert_filename and private_key_filename are optional names of a client side
* certificate and private key files to use for authentication. They need to
* be both specified or omitted.
*
* server_name is an optional and will be used as a server name indication
* (SNI) TLS extension.
*
* If error is non-null, it will be populated in case the context creation fails
* (returning a NULL).
*/ */
int redisSecureConnection(redisContext *c, const char *capath, const char *certpath,
const char *keypath, const char *servername); redisSSLContext *redisCreateSSLContext(const char *cacert_filename, const char *capath,
const char *cert_filename, const char *private_key_filename,
const char *server_name, redisSSLContextError *error);
/** /**
* Initiate SSL/TLS negotiation on a provided context. * Free a previously created OpenSSL context.
*/
void redisFreeSSLContext(redisSSLContext *redis_ssl_ctx);
/**
* Initiate SSL on an existing redisContext.
*
* This is similar to redisInitiateSSL() but does not require the caller
* to directly interact with OpenSSL, and instead uses a redisSSLContext
* previously created using redisCreateSSLContext().
*/
int redisInitiateSSLWithContext(redisContext *c, redisSSLContext *redis_ssl_ctx);
/**
* Initiate SSL/TLS negotiation on a provided OpenSSL SSL object.
*/ */
int redisInitiateSSL(redisContext *c, struct ssl_st *ssl); int redisInitiateSSL(redisContext *c, struct ssl_st *ssl);
#ifdef __cplusplus
}
#endif
#endif /* __HIREDIS_SSL_H */ #endif /* __HIREDIS_SSL_H */
...@@ -57,8 +57,8 @@ void redisNetClose(redisContext *c) { ...@@ -57,8 +57,8 @@ void redisNetClose(redisContext *c) {
} }
} }
int redisNetRead(redisContext *c, char *buf, size_t bufcap) { ssize_t redisNetRead(redisContext *c, char *buf, size_t bufcap) {
int nread = recv(c->fd, buf, bufcap, 0); ssize_t nread = recv(c->fd, buf, bufcap, 0);
if (nread == -1) { if (nread == -1) {
if ((errno == EWOULDBLOCK && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) { if ((errno == EWOULDBLOCK && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) {
/* Try again later */ /* Try again later */
...@@ -79,8 +79,8 @@ int redisNetRead(redisContext *c, char *buf, size_t bufcap) { ...@@ -79,8 +79,8 @@ int redisNetRead(redisContext *c, char *buf, size_t bufcap) {
} }
} }
int redisNetWrite(redisContext *c) { ssize_t redisNetWrite(redisContext *c) {
int nwritten = send(c->fd, c->obuf, sdslen(c->obuf), 0); ssize_t nwritten = send(c->fd, c->obuf, sdslen(c->obuf), 0);
if (nwritten < 0) { if (nwritten < 0) {
if ((errno == EWOULDBLOCK && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) { if ((errno == EWOULDBLOCK && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) {
/* Try again later */ /* Try again later */
...@@ -203,7 +203,7 @@ int redisKeepAlive(redisContext *c, int interval) { ...@@ -203,7 +203,7 @@ int redisKeepAlive(redisContext *c, int interval) {
return REDIS_OK; return REDIS_OK;
} }
static int redisSetTcpNoDelay(redisContext *c) { int redisSetTcpNoDelay(redisContext *c) {
int yes = 1; int yes = 1;
if (setsockopt(c->fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) == -1) { if (setsockopt(c->fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) == -1) {
__redisSetErrorFromErrno(c,REDIS_ERR_IO,"setsockopt(TCP_NODELAY)"); __redisSetErrorFromErrno(c,REDIS_ERR_IO,"setsockopt(TCP_NODELAY)");
...@@ -217,7 +217,7 @@ static int redisSetTcpNoDelay(redisContext *c) { ...@@ -217,7 +217,7 @@ static int redisSetTcpNoDelay(redisContext *c) {
static int redisContextTimeoutMsec(redisContext *c, long *result) static int redisContextTimeoutMsec(redisContext *c, long *result)
{ {
const struct timeval *timeout = c->timeout; const struct timeval *timeout = c->connect_timeout;
long msec = -1; long msec = -1;
/* Only use timeout when not NULL. */ /* Only use timeout when not NULL. */
...@@ -316,11 +316,7 @@ int redisCheckSocketError(redisContext *c) { ...@@ -316,11 +316,7 @@ int redisCheckSocketError(redisContext *c) {
int redisContextSetTimeout(redisContext *c, const struct timeval tv) { int redisContextSetTimeout(redisContext *c, const struct timeval tv) {
const void *to_ptr = &tv; const void *to_ptr = &tv;
size_t to_sz = sizeof(tv); size_t to_sz = sizeof(tv);
#ifdef _WIN32
DWORD timeout_msec = tv.tv_sec * 1000 + tv.tv_usec / 1000;
to_ptr = &timeout_msec;
to_sz = sizeof(timeout_msec);
#endif
if (setsockopt(c->fd,SOL_SOCKET,SO_RCVTIMEO,to_ptr,to_sz) == -1) { if (setsockopt(c->fd,SOL_SOCKET,SO_RCVTIMEO,to_ptr,to_sz) == -1) {
__redisSetErrorFromErrno(c,REDIS_ERR_IO,"setsockopt(SO_RCVTIMEO)"); __redisSetErrorFromErrno(c,REDIS_ERR_IO,"setsockopt(SO_RCVTIMEO)");
return REDIS_ERR; return REDIS_ERR;
...@@ -332,6 +328,38 @@ int redisContextSetTimeout(redisContext *c, const struct timeval tv) { ...@@ -332,6 +328,38 @@ int redisContextSetTimeout(redisContext *c, const struct timeval tv) {
return REDIS_OK; return REDIS_OK;
} }
int redisContextUpdateConnectTimeout(redisContext *c, const struct timeval *timeout) {
/* Same timeval struct, short circuit */
if (c->connect_timeout == timeout)
return REDIS_OK;
/* Allocate context timeval if we need to */
if (c->connect_timeout == NULL) {
c->connect_timeout = hi_malloc(sizeof(*c->connect_timeout));
if (c->connect_timeout == NULL)
return REDIS_ERR;
}
memcpy(c->connect_timeout, timeout, sizeof(*c->connect_timeout));
return REDIS_OK;
}
int redisContextUpdateCommandTimeout(redisContext *c, const struct timeval *timeout) {
/* Same timeval struct, short circuit */
if (c->command_timeout == timeout)
return REDIS_OK;
/* Allocate context timeval if we need to */
if (c->command_timeout == NULL) {
c->command_timeout = hi_malloc(sizeof(*c->command_timeout));
if (c->command_timeout == NULL)
return REDIS_ERR;
}
memcpy(c->command_timeout, timeout, sizeof(*c->command_timeout));
return REDIS_OK;
}
static int _redisContextConnectTcp(redisContext *c, const char *addr, int port, static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
const struct timeval *timeout, const struct timeval *timeout,
const char *source_addr) { const char *source_addr) {
...@@ -356,21 +384,19 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port, ...@@ -356,21 +384,19 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
* This is a bit ugly, but atleast it works and doesn't leak memory. * This is a bit ugly, but atleast it works and doesn't leak memory.
**/ **/
if (c->tcp.host != addr) { if (c->tcp.host != addr) {
free(c->tcp.host); hi_free(c->tcp.host);
c->tcp.host = strdup(addr); c->tcp.host = hi_strdup(addr);
if (c->tcp.host == NULL)
goto oom;
} }
if (timeout) { if (timeout) {
if (c->timeout != timeout) { if (redisContextUpdateConnectTimeout(c, timeout) == REDIS_ERR)
if (c->timeout == NULL) goto oom;
c->timeout = malloc(sizeof(struct timeval));
memcpy(c->timeout, timeout, sizeof(struct timeval));
}
} else { } else {
free(c->timeout); hi_free(c->connect_timeout);
c->timeout = NULL; c->connect_timeout = NULL;
} }
if (redisContextTimeoutMsec(c, &timeout_msec) != REDIS_OK) { if (redisContextTimeoutMsec(c, &timeout_msec) != REDIS_OK) {
...@@ -379,11 +405,11 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port, ...@@ -379,11 +405,11 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
} }
if (source_addr == NULL) { if (source_addr == NULL) {
free(c->tcp.source_addr); hi_free(c->tcp.source_addr);
c->tcp.source_addr = NULL; c->tcp.source_addr = NULL;
} else if (c->tcp.source_addr != source_addr) { } else if (c->tcp.source_addr != source_addr) {
free(c->tcp.source_addr); hi_free(c->tcp.source_addr);
c->tcp.source_addr = strdup(source_addr); c->tcp.source_addr = hi_strdup(source_addr);
} }
snprintf(_port, 6, "%d", port); snprintf(_port, 6, "%d", port);
...@@ -446,8 +472,11 @@ addrretry: ...@@ -446,8 +472,11 @@ addrretry:
} }
/* For repeat connection */ /* For repeat connection */
free(c->saddr); hi_free(c->saddr);
c->saddr = malloc(p->ai_addrlen); c->saddr = hi_malloc(p->ai_addrlen);
if (c->saddr == NULL)
goto oom;
memcpy(c->saddr, p->ai_addr, p->ai_addrlen); memcpy(c->saddr, p->ai_addr, p->ai_addrlen);
c->addrlen = p->ai_addrlen; c->addrlen = p->ai_addrlen;
...@@ -474,12 +503,12 @@ addrretry: ...@@ -474,12 +503,12 @@ addrretry:
wait_for_ready: wait_for_ready:
if (redisContextWaitReady(c,timeout_msec) != REDIS_OK) if (redisContextWaitReady(c,timeout_msec) != REDIS_OK)
goto error; goto error;
if (redisSetTcpNoDelay(c) != REDIS_OK)
goto error;
} }
} }
if (blocking && redisSetBlocking(c,1) != REDIS_OK) if (blocking && redisSetBlocking(c,1) != REDIS_OK)
goto error; goto error;
if (redisSetTcpNoDelay(c) != REDIS_OK)
goto error;
c->flags |= REDIS_CONNECTED; c->flags |= REDIS_CONNECTED;
rv = REDIS_OK; rv = REDIS_OK;
...@@ -492,6 +521,8 @@ addrretry: ...@@ -492,6 +521,8 @@ addrretry:
goto error; goto error;
} }
oom:
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
error: error:
rv = REDIS_ERR; rv = REDIS_ERR;
end: end:
...@@ -525,25 +556,32 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time ...@@ -525,25 +556,32 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time
return REDIS_ERR; return REDIS_ERR;
c->connection_type = REDIS_CONN_UNIX; c->connection_type = REDIS_CONN_UNIX;
if (c->unix_sock.path != path) if (c->unix_sock.path != path) {
c->unix_sock.path = strdup(path); hi_free(c->unix_sock.path);
if (timeout) {
if (c->timeout != timeout) {
if (c->timeout == NULL)
c->timeout = malloc(sizeof(struct timeval));
memcpy(c->timeout, timeout, sizeof(struct timeval)); c->unix_sock.path = hi_strdup(path);
if (c->unix_sock.path == NULL)
goto oom;
} }
if (timeout) {
if (redisContextUpdateConnectTimeout(c, timeout) == REDIS_ERR)
goto oom;
} else { } else {
free(c->timeout); hi_free(c->connect_timeout);
c->timeout = NULL; c->connect_timeout = NULL;
} }
if (redisContextTimeoutMsec(c,&timeout_msec) != REDIS_OK) if (redisContextTimeoutMsec(c,&timeout_msec) != REDIS_OK)
return REDIS_ERR; return REDIS_ERR;
sa = (struct sockaddr_un*)(c->saddr = malloc(sizeof(struct sockaddr_un))); /* Don't leak sockaddr if we're reconnecting */
if (c->saddr) hi_free(c->saddr);
sa = (struct sockaddr_un*)(c->saddr = hi_malloc(sizeof(struct sockaddr_un)));
if (sa == NULL)
goto oom;
c->addrlen = sizeof(struct sockaddr_un); c->addrlen = sizeof(struct sockaddr_un);
sa->sun_family = AF_UNIX; sa->sun_family = AF_UNIX;
strncpy(sa->sun_path, path, sizeof(sa->sun_path) - 1); strncpy(sa->sun_path, path, sizeof(sa->sun_path) - 1);
...@@ -568,4 +606,7 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time ...@@ -568,4 +606,7 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time
errno = EPROTONOSUPPORT; errno = EPROTONOSUPPORT;
return REDIS_ERR; return REDIS_ERR;
#endif /* _WIN32 */ #endif /* _WIN32 */
oom:
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
} }
...@@ -38,8 +38,8 @@ ...@@ -38,8 +38,8 @@
#include "hiredis.h" #include "hiredis.h"
void redisNetClose(redisContext *c); void redisNetClose(redisContext *c);
int redisNetRead(redisContext *c, char *buf, size_t bufcap); ssize_t redisNetRead(redisContext *c, char *buf, size_t bufcap);
int redisNetWrite(redisContext *c); ssize_t redisNetWrite(redisContext *c);
int redisCheckSocketError(redisContext *c); int redisCheckSocketError(redisContext *c);
int redisContextSetTimeout(redisContext *c, const struct timeval tv); int redisContextSetTimeout(redisContext *c, const struct timeval tv);
...@@ -51,4 +51,6 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time ...@@ -51,4 +51,6 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time
int redisKeepAlive(redisContext *c, int interval); int redisKeepAlive(redisContext *c, int interval);
int redisCheckConnectDone(redisContext *c, int *completed); int redisCheckConnectDone(redisContext *c, int *completed);
int redisSetTcpNoDelay(redisContext *c);
#endif #endif
...@@ -42,10 +42,14 @@ ...@@ -42,10 +42,14 @@
#include <limits.h> #include <limits.h>
#include <math.h> #include <math.h>
#include "alloc.h"
#include "read.h" #include "read.h"
#include "sds.h" #include "sds.h"
#include "win32.h" #include "win32.h"
/* Initial size of our nested reply stack and how much we grow it when needd */
#define REDIS_READER_STACK_SIZE 9
static void __redisReaderSetError(redisReader *r, int type, const char *str) { static void __redisReaderSetError(redisReader *r, int type, const char *str) {
size_t len; size_t len;
...@@ -243,11 +247,12 @@ static void moveToNextTask(redisReader *r) { ...@@ -243,11 +247,12 @@ static void moveToNextTask(redisReader *r) {
return; return;
} }
cur = &(r->rstack[r->ridx]); cur = r->task[r->ridx];
prv = &(r->rstack[r->ridx-1]); prv = r->task[r->ridx-1];
assert(prv->type == REDIS_REPLY_ARRAY || assert(prv->type == REDIS_REPLY_ARRAY ||
prv->type == REDIS_REPLY_MAP || prv->type == REDIS_REPLY_MAP ||
prv->type == REDIS_REPLY_SET); prv->type == REDIS_REPLY_SET ||
prv->type == REDIS_REPLY_PUSH);
if (cur->idx == prv->elements-1) { if (cur->idx == prv->elements-1) {
r->ridx--; r->ridx--;
} else { } else {
...@@ -262,7 +267,7 @@ static void moveToNextTask(redisReader *r) { ...@@ -262,7 +267,7 @@ static void moveToNextTask(redisReader *r) {
} }
static int processLineItem(redisReader *r) { static int processLineItem(redisReader *r) {
redisReadTask *cur = &(r->rstack[r->ridx]); redisReadTask *cur = r->task[r->ridx];
void *obj; void *obj;
char *p; char *p;
int len; int len;
...@@ -297,7 +302,7 @@ static int processLineItem(redisReader *r) { ...@@ -297,7 +302,7 @@ static int processLineItem(redisReader *r) {
if (strcasecmp(buf,",inf") == 0) { if (strcasecmp(buf,",inf") == 0) {
d = INFINITY; /* Positive infinite. */ d = INFINITY; /* Positive infinite. */
} else if (strcasecmp(buf,",-inf") == 0) { } else if (strcasecmp(buf,",-inf") == 0) {
d = -INFINITY; /* Nevative infinite. */ d = -INFINITY; /* Negative infinite. */
} else { } else {
d = strtod((char*)buf,&eptr); d = strtod((char*)buf,&eptr);
if (buf[0] == '\0' || eptr[0] != '\0' || isnan(d)) { if (buf[0] == '\0' || eptr[0] != '\0' || isnan(d)) {
...@@ -344,7 +349,7 @@ static int processLineItem(redisReader *r) { ...@@ -344,7 +349,7 @@ static int processLineItem(redisReader *r) {
} }
static int processBulkItem(redisReader *r) { static int processBulkItem(redisReader *r) {
redisReadTask *cur = &(r->rstack[r->ridx]); redisReadTask *cur = r->task[r->ridx];
void *obj = NULL; void *obj = NULL;
char *p, *s; char *p, *s;
long long len; long long len;
...@@ -415,18 +420,42 @@ static int processBulkItem(redisReader *r) { ...@@ -415,18 +420,42 @@ static int processBulkItem(redisReader *r) {
return REDIS_ERR; return REDIS_ERR;
} }
static int redisReaderGrow(redisReader *r) {
redisReadTask **aux;
int newlen;
/* Grow our stack size */
newlen = r->tasks + REDIS_READER_STACK_SIZE;
aux = hi_realloc(r->task, sizeof(*r->task) * newlen);
if (aux == NULL)
goto oom;
r->task = aux;
/* Allocate new tasks */
for (; r->tasks < newlen; r->tasks++) {
r->task[r->tasks] = hi_calloc(1, sizeof(**r->task));
if (r->task[r->tasks] == NULL)
goto oom;
}
return REDIS_OK;
oom:
__redisReaderSetErrorOOM(r);
return REDIS_ERR;
}
/* Process the array, map and set types. */ /* Process the array, map and set types. */
static int processAggregateItem(redisReader *r) { static int processAggregateItem(redisReader *r) {
redisReadTask *cur = &(r->rstack[r->ridx]); redisReadTask *cur = r->task[r->ridx];
void *obj; void *obj;
char *p; char *p;
long long elements; long long elements;
int root = 0, len; int root = 0, len;
/* Set error for nested multi bulks with depth > 7 */ /* Set error for nested multi bulks with depth > 7 */
if (r->ridx == 8) { if (r->ridx == r->tasks - 1) {
__redisReaderSetError(r,REDIS_ERR_PROTOCOL, if (redisReaderGrow(r) == REDIS_ERR)
"No support for nested multi bulk replies with depth > 7");
return REDIS_ERR; return REDIS_ERR;
} }
...@@ -439,7 +468,9 @@ static int processAggregateItem(redisReader *r) { ...@@ -439,7 +468,9 @@ static int processAggregateItem(redisReader *r) {
root = (r->ridx == 0); root = (r->ridx == 0);
if (elements < -1 || (LLONG_MAX > SIZE_MAX && elements > SIZE_MAX)) { if (elements < -1 || (LLONG_MAX > SIZE_MAX && elements > SIZE_MAX) ||
(r->maxelements > 0 && elements > r->maxelements))
{
__redisReaderSetError(r,REDIS_ERR_PROTOCOL, __redisReaderSetError(r,REDIS_ERR_PROTOCOL,
"Multi-bulk length out of range"); "Multi-bulk length out of range");
return REDIS_ERR; return REDIS_ERR;
...@@ -475,12 +506,12 @@ static int processAggregateItem(redisReader *r) { ...@@ -475,12 +506,12 @@ static int processAggregateItem(redisReader *r) {
cur->elements = elements; cur->elements = elements;
cur->obj = obj; cur->obj = obj;
r->ridx++; r->ridx++;
r->rstack[r->ridx].type = -1; r->task[r->ridx]->type = -1;
r->rstack[r->ridx].elements = -1; r->task[r->ridx]->elements = -1;
r->rstack[r->ridx].idx = 0; r->task[r->ridx]->idx = 0;
r->rstack[r->ridx].obj = NULL; r->task[r->ridx]->obj = NULL;
r->rstack[r->ridx].parent = cur; r->task[r->ridx]->parent = cur;
r->rstack[r->ridx].privdata = r->privdata; r->task[r->ridx]->privdata = r->privdata;
} else { } else {
moveToNextTask(r); moveToNextTask(r);
} }
...@@ -495,7 +526,7 @@ static int processAggregateItem(redisReader *r) { ...@@ -495,7 +526,7 @@ static int processAggregateItem(redisReader *r) {
} }
static int processItem(redisReader *r) { static int processItem(redisReader *r) {
redisReadTask *cur = &(r->rstack[r->ridx]); redisReadTask *cur = r->task[r->ridx];
char *p; char *p;
/* check if we need to read type */ /* check if we need to read type */
...@@ -535,6 +566,9 @@ static int processItem(redisReader *r) { ...@@ -535,6 +566,9 @@ static int processItem(redisReader *r) {
case '=': case '=':
cur->type = REDIS_REPLY_VERB; cur->type = REDIS_REPLY_VERB;
break; break;
case '>':
cur->type = REDIS_REPLY_PUSH;
break;
default: default:
__redisReaderSetErrorProtocolByte(r,*p); __redisReaderSetErrorProtocolByte(r,*p);
return REDIS_ERR; return REDIS_ERR;
...@@ -560,6 +594,7 @@ static int processItem(redisReader *r) { ...@@ -560,6 +594,7 @@ static int processItem(redisReader *r) {
case REDIS_REPLY_ARRAY: case REDIS_REPLY_ARRAY:
case REDIS_REPLY_MAP: case REDIS_REPLY_MAP:
case REDIS_REPLY_SET: case REDIS_REPLY_SET:
case REDIS_REPLY_PUSH:
return processAggregateItem(r); return processAggregateItem(r);
default: default:
assert(NULL); assert(NULL);
...@@ -570,29 +605,53 @@ static int processItem(redisReader *r) { ...@@ -570,29 +605,53 @@ static int processItem(redisReader *r) {
redisReader *redisReaderCreateWithFunctions(redisReplyObjectFunctions *fn) { redisReader *redisReaderCreateWithFunctions(redisReplyObjectFunctions *fn) {
redisReader *r; redisReader *r;
r = calloc(1,sizeof(redisReader)); r = hi_calloc(1,sizeof(redisReader));
if (r == NULL) if (r == NULL)
return NULL; return NULL;
r->fn = fn;
r->buf = sdsempty(); r->buf = sdsempty();
r->maxbuf = REDIS_READER_MAX_BUF; if (r->buf == NULL)
if (r->buf == NULL) { goto oom;
free(r);
return NULL; r->task = hi_calloc(REDIS_READER_STACK_SIZE, sizeof(*r->task));
if (r->task == NULL)
goto oom;
for (; r->tasks < REDIS_READER_STACK_SIZE; r->tasks++) {
r->task[r->tasks] = hi_calloc(1, sizeof(**r->task));
if (r->task[r->tasks] == NULL)
goto oom;
} }
r->fn = fn;
r->maxbuf = REDIS_READER_MAX_BUF;
r->maxelements = REDIS_READER_MAX_ARRAY_ELEMENTS;
r->ridx = -1; r->ridx = -1;
return r; return r;
oom:
redisReaderFree(r);
return NULL;
} }
void redisReaderFree(redisReader *r) { void redisReaderFree(redisReader *r) {
if (r == NULL) if (r == NULL)
return; return;
if (r->reply != NULL && r->fn && r->fn->freeObject) if (r->reply != NULL && r->fn && r->fn->freeObject)
r->fn->freeObject(r->reply); r->fn->freeObject(r->reply);
if (r->task) {
/* We know r->task[i] is allocated if i < r->tasks */
for (int i = 0; i < r->tasks; i++) {
hi_free(r->task[i]);
}
hi_free(r->task);
}
sdsfree(r->buf); sdsfree(r->buf);
free(r); hi_free(r);
} }
int redisReaderFeed(redisReader *r, const char *buf, size_t len) { int redisReaderFeed(redisReader *r, const char *buf, size_t len) {
...@@ -608,23 +667,22 @@ int redisReaderFeed(redisReader *r, const char *buf, size_t len) { ...@@ -608,23 +667,22 @@ int redisReaderFeed(redisReader *r, const char *buf, size_t len) {
if (r->len == 0 && r->maxbuf != 0 && sdsavail(r->buf) > r->maxbuf) { if (r->len == 0 && r->maxbuf != 0 && sdsavail(r->buf) > r->maxbuf) {
sdsfree(r->buf); sdsfree(r->buf);
r->buf = sdsempty(); r->buf = sdsempty();
r->pos = 0; if (r->buf == 0) goto oom;
/* r->buf should not be NULL since we just free'd a larger one. */ r->pos = 0;
assert(r->buf != NULL);
} }
newbuf = sdscatlen(r->buf,buf,len); newbuf = sdscatlen(r->buf,buf,len);
if (newbuf == NULL) { if (newbuf == NULL) goto oom;
__redisReaderSetErrorOOM(r);
return REDIS_ERR;
}
r->buf = newbuf; r->buf = newbuf;
r->len = sdslen(r->buf); r->len = sdslen(r->buf);
} }
return REDIS_OK; return REDIS_OK;
oom:
__redisReaderSetErrorOOM(r);
return REDIS_ERR;
} }
int redisReaderGetReply(redisReader *r, void **reply) { int redisReaderGetReply(redisReader *r, void **reply) {
...@@ -642,12 +700,12 @@ int redisReaderGetReply(redisReader *r, void **reply) { ...@@ -642,12 +700,12 @@ int redisReaderGetReply(redisReader *r, void **reply) {
/* Set first item to process when the stack is empty. */ /* Set first item to process when the stack is empty. */
if (r->ridx == -1) { if (r->ridx == -1) {
r->rstack[0].type = -1; r->task[0]->type = -1;
r->rstack[0].elements = -1; r->task[0]->elements = -1;
r->rstack[0].idx = -1; r->task[0]->idx = -1;
r->rstack[0].obj = NULL; r->task[0]->obj = NULL;
r->rstack[0].parent = NULL; r->task[0]->parent = NULL;
r->rstack[0].privdata = r->privdata; r->task[0]->privdata = r->privdata;
r->ridx = 0; r->ridx = 0;
} }
...@@ -663,7 +721,7 @@ int redisReaderGetReply(redisReader *r, void **reply) { ...@@ -663,7 +721,7 @@ int redisReaderGetReply(redisReader *r, void **reply) {
/* Discard part of the buffer when we've consumed at least 1k, to avoid /* Discard part of the buffer when we've consumed at least 1k, to avoid
* doing unnecessary calls to memmove() in sds.c. */ * doing unnecessary calls to memmove() in sds.c. */
if (r->pos >= 1024) { if (r->pos >= 1024) {
sdsrange(r->buf,r->pos,-1); if (sdsrange(r->buf,r->pos,-1) < 0) return REDIS_ERR;
r->pos = 0; r->pos = 0;
r->len = sdslen(r->buf); r->len = sdslen(r->buf);
} }
......
...@@ -63,7 +63,11 @@ ...@@ -63,7 +63,11 @@
#define REDIS_REPLY_BIGNUM 13 #define REDIS_REPLY_BIGNUM 13
#define REDIS_REPLY_VERB 14 #define REDIS_REPLY_VERB 14
#define REDIS_READER_MAX_BUF (1024*16) /* Default max unused reader buffer. */ /* Default max unused reader buffer. */
#define REDIS_READER_MAX_BUF (1024*16)
/* Default multi-bulk element limit */
#define REDIS_READER_MAX_ARRAY_ELEMENTS ((1LL<<32) - 1)
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
...@@ -71,7 +75,7 @@ extern "C" { ...@@ -71,7 +75,7 @@ extern "C" {
typedef struct redisReadTask { typedef struct redisReadTask {
int type; int type;
int elements; /* number of elements in multibulk container */ long long elements; /* number of elements in multibulk container */
int idx; /* index in parent (array) object */ int idx; /* index in parent (array) object */
void *obj; /* holds user-generated value for a read task */ void *obj; /* holds user-generated value for a read task */
struct redisReadTask *parent; /* parent task */ struct redisReadTask *parent; /* parent task */
...@@ -96,8 +100,11 @@ typedef struct redisReader { ...@@ -96,8 +100,11 @@ typedef struct redisReader {
size_t pos; /* Buffer cursor */ size_t pos; /* Buffer cursor */
size_t len; /* Buffer length */ size_t len; /* Buffer length */
size_t maxbuf; /* Max length of unused buffer */ size_t maxbuf; /* Max length of unused buffer */
long long maxelements; /* Max multi-bulk elements */
redisReadTask **task;
int tasks;
redisReadTask rstack[9];
int ridx; /* Index of current read task */ int ridx; /* Index of current read task */
void *reply; /* Temporary reply pointer */ void *reply; /* Temporary reply pointer */
......
...@@ -30,11 +30,13 @@ ...@@ -30,11 +30,13 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#include "fmacros.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <ctype.h> #include <ctype.h>
#include <assert.h> #include <assert.h>
#include <limits.h>
#include "sds.h" #include "sds.h"
#include "sdsalloc.h" #include "sdsalloc.h"
...@@ -219,10 +221,7 @@ sds sdsMakeRoomFor(sds s, size_t addlen) { ...@@ -219,10 +221,7 @@ sds sdsMakeRoomFor(sds s, size_t addlen) {
hdrlen = sdsHdrSize(type); hdrlen = sdsHdrSize(type);
if (oldtype==type) { if (oldtype==type) {
newsh = s_realloc(sh, hdrlen+newlen+1); newsh = s_realloc(sh, hdrlen+newlen+1);
if (newsh == NULL) { if (newsh == NULL) return NULL;
s_free(sh);
return NULL;
}
s = (char*)newsh+hdrlen; s = (char*)newsh+hdrlen;
} else { } else {
/* Since the header size changes, need to move the string forward, /* Since the header size changes, need to move the string forward,
...@@ -715,15 +714,20 @@ sds sdstrim(sds s, const char *cset) { ...@@ -715,15 +714,20 @@ sds sdstrim(sds s, const char *cset) {
* *
* The string is modified in-place. * The string is modified in-place.
* *
* Return value:
* -1 (error) if sdslen(s) is larger than maximum positive ssize_t value.
* 0 on success.
*
* Example: * Example:
* *
* s = sdsnew("Hello World"); * s = sdsnew("Hello World");
* sdsrange(s,1,-1); => "ello World" * sdsrange(s,1,-1); => "ello World"
*/ */
void sdsrange(sds s, int start, int end) { int sdsrange(sds s, ssize_t start, ssize_t end) {
size_t newlen, len = sdslen(s); size_t newlen, len = sdslen(s);
if (len > SSIZE_MAX) return -1;
if (len == 0) return; if (len == 0) return 0;
if (start < 0) { if (start < 0) {
start = len+start; start = len+start;
if (start < 0) start = 0; if (start < 0) start = 0;
...@@ -734,9 +738,9 @@ void sdsrange(sds s, int start, int end) { ...@@ -734,9 +738,9 @@ void sdsrange(sds s, int start, int end) {
} }
newlen = (start > end) ? 0 : (end-start)+1; newlen = (start > end) ? 0 : (end-start)+1;
if (newlen != 0) { if (newlen != 0) {
if (start >= (signed)len) { if (start >= (ssize_t)len) {
newlen = 0; newlen = 0;
} else if (end >= (signed)len) { } else if (end >= (ssize_t)len) {
end = len-1; end = len-1;
newlen = (start > end) ? 0 : (end-start)+1; newlen = (start > end) ? 0 : (end-start)+1;
} }
...@@ -746,6 +750,7 @@ void sdsrange(sds s, int start, int end) { ...@@ -746,6 +750,7 @@ void sdsrange(sds s, int start, int end) {
if (start && newlen) memmove(s, s+start, newlen); if (start && newlen) memmove(s, s+start, newlen);
s[newlen] = 0; s[newlen] = 0;
sdssetlen(s,newlen); sdssetlen(s,newlen);
return 0;
} }
/* Apply tolower() to every character of the sds string 's'. */ /* Apply tolower() to every character of the sds string 's'. */
...@@ -889,13 +894,6 @@ sds sdscatrepr(sds s, const char *p, size_t len) { ...@@ -889,13 +894,6 @@ sds sdscatrepr(sds s, const char *p, size_t len) {
return sdscatlen(s,"\"",1); return sdscatlen(s,"\"",1);
} }
/* Helper function for sdssplitargs() that returns non zero if 'c'
* is a valid hex digit. */
int is_hex_digit(char c) {
return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') ||
(c >= 'A' && c <= 'F');
}
/* Helper function for sdssplitargs() that converts a hex digit into an /* Helper function for sdssplitargs() that converts a hex digit into an
* integer from 0 to 15 */ * integer from 0 to 15 */
int hex_digit_to_int(char c) { int hex_digit_to_int(char c) {
...@@ -958,8 +956,8 @@ sds *sdssplitargs(const char *line, int *argc) { ...@@ -958,8 +956,8 @@ sds *sdssplitargs(const char *line, int *argc) {
while(!done) { while(!done) {
if (inq) { if (inq) {
if (*p == '\\' && *(p+1) == 'x' && if (*p == '\\' && *(p+1) == 'x' &&
is_hex_digit(*(p+2)) && isxdigit(*(p+2)) &&
is_hex_digit(*(p+3))) isxdigit(*(p+3)))
{ {
unsigned char byte; unsigned char byte;
......
...@@ -36,6 +36,8 @@ ...@@ -36,6 +36,8 @@
#define SDS_MAX_PREALLOC (1024*1024) #define SDS_MAX_PREALLOC (1024*1024)
#ifdef _MSC_VER #ifdef _MSC_VER
#define __attribute__(x) #define __attribute__(x)
typedef long long ssize_t;
#define SSIZE_MAX (LLONG_MAX >> 1)
#endif #endif
#include <sys/types.h> #include <sys/types.h>
...@@ -239,7 +241,7 @@ sds sdscatprintf(sds s, const char *fmt, ...); ...@@ -239,7 +241,7 @@ sds sdscatprintf(sds s, const char *fmt, ...);
sds sdscatfmt(sds s, char const *fmt, ...); sds sdscatfmt(sds s, char const *fmt, ...);
sds sdstrim(sds s, const char *cset); sds sdstrim(sds s, const char *cset);
void sdsrange(sds s, int start, int end); int sdsrange(sds s, ssize_t start, ssize_t end);
void sdsupdatelen(sds s); void sdsupdatelen(sds s);
void sdsclear(sds s); void sdsclear(sds s);
int sdscmp(const sds s1, const sds s2); int sdscmp(const sds s1, const sds s2);
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment