/* * Copyright (c) 2009-2011, Salvatore Sanfilippo * Copyright (c) 2010-2014, Pieter Noordhuis * Copyright (c) 2015, Matt Stancliff , * Jan-Erik Rediger * * SPDX-FileCopyrightText: 2024 Hiredict Contributors * SPDX-FileCopyrightText: 2024 Salvatore Sanfilippo * SPDX-FileCopyrightText: 2024 Pieter Noordhuis * SPDX-FileCopyrightText: 2024 Matt Stancliff * SPDX-FileCopyrightText: 2024 Jan-Erik Rediger * * SPDX-License-Identifier: BSD-3-Clause * SPDX-License-Identifier: LGPL-3.0-or-later * */ #include "fmacros.h" #include #include #include #include #include #include "hiredict.h" #include "net.h" #include "sds.h" #include "async.h" #include "win32.h" extern int redictContextUpdateConnectTimeout(redictContext *c, const struct timeval *timeout); extern int redictContextUpdateCommandTimeout(redictContext *c, const struct timeval *timeout); static redictContextFuncs redictContextDefaultFuncs = { .close = redictNetClose, .free_privctx = NULL, .async_read = redictAsyncRead, .async_write = redictAsyncWrite, .read = redictNetRead, .write = redictNetWrite }; static redictReply *createReplyObject(int type); static void *createStringObject(const redictReadTask *task, char *str, size_t len); static void *createArrayObject(const redictReadTask *task, size_t elements); static void *createIntegerObject(const redictReadTask *task, long long value); static void *createDoubleObject(const redictReadTask *task, double value, char *str, size_t len); static void *createNilObject(const redictReadTask *task); static void *createBoolObject(const redictReadTask *task, int bval); /* Default set of functions to build the reply. Keep in mind that such a * function returning NULL is interpreted as OOM. */ static redictReplyObjectFunctions defaultFunctions = { createStringObject, createArrayObject, createIntegerObject, createDoubleObject, createNilObject, createBoolObject, freeReplyObject }; /* Create a reply object */ static redictReply *createReplyObject(int type) { redictReply *r = hi_calloc(1,sizeof(*r)); if (r == NULL) return NULL; r->type = type; return r; } /* Free a reply object */ void freeReplyObject(void *reply) { redictReply *r = reply; size_t j; if (r == NULL) return; switch(r->type) { case REDICT_REPLY_INTEGER: case REDICT_REPLY_NIL: case REDICT_REPLY_BOOL: break; /* Nothing to free */ case REDICT_REPLY_ARRAY: case REDICT_REPLY_MAP: case REDICT_REPLY_ATTR: case REDICT_REPLY_SET: case REDICT_REPLY_PUSH: if (r->element != NULL) { for (j = 0; j < r->elements; j++) freeReplyObject(r->element[j]); hi_free(r->element); } break; case REDICT_REPLY_ERROR: case REDICT_REPLY_STATUS: case REDICT_REPLY_STRING: case REDICT_REPLY_DOUBLE: case REDICT_REPLY_VERB: case REDICT_REPLY_BIGNUM: hi_free(r->str); break; } hi_free(r); } static void *createStringObject(const redictReadTask *task, char *str, size_t len) { redictReply *r, *parent; char *buf; r = createReplyObject(task->type); if (r == NULL) return NULL; assert(task->type == REDICT_REPLY_ERROR || task->type == REDICT_REPLY_STATUS || task->type == REDICT_REPLY_STRING || task->type == REDICT_REPLY_VERB || task->type == REDICT_REPLY_BIGNUM); /* Copy string value */ if (task->type == REDICT_REPLY_VERB) { buf = hi_malloc(len-4+1); /* Skip 4 bytes of verbatim type header. */ if (buf == NULL) goto oom; memcpy(r->vtype,str,3); r->vtype[3] = '\0'; memcpy(buf,str+4,len-4); buf[len-4] = '\0'; r->len = len - 4; } else { buf = hi_malloc(len+1); if (buf == NULL) goto oom; memcpy(buf,str,len); buf[len] = '\0'; r->len = len; } r->str = buf; if (task->parent) { parent = task->parent->obj; assert(parent->type == REDICT_REPLY_ARRAY || parent->type == REDICT_REPLY_MAP || parent->type == REDICT_REPLY_ATTR || parent->type == REDICT_REPLY_SET || parent->type == REDICT_REPLY_PUSH); parent->element[task->idx] = r; } return r; oom: freeReplyObject(r); return NULL; } static void *createArrayObject(const redictReadTask *task, size_t elements) { redictReply *r, *parent; r = createReplyObject(task->type); if (r == NULL) return NULL; if (elements > 0) { r->element = hi_calloc(elements,sizeof(redictReply*)); if (r->element == NULL) { freeReplyObject(r); return NULL; } } r->elements = elements; if (task->parent) { parent = task->parent->obj; assert(parent->type == REDICT_REPLY_ARRAY || parent->type == REDICT_REPLY_MAP || parent->type == REDICT_REPLY_ATTR || parent->type == REDICT_REPLY_SET || parent->type == REDICT_REPLY_PUSH); parent->element[task->idx] = r; } return r; } static void *createIntegerObject(const redictReadTask *task, long long value) { redictReply *r, *parent; r = createReplyObject(REDICT_REPLY_INTEGER); if (r == NULL) return NULL; r->integer = value; if (task->parent) { parent = task->parent->obj; assert(parent->type == REDICT_REPLY_ARRAY || parent->type == REDICT_REPLY_MAP || parent->type == REDICT_REPLY_ATTR || parent->type == REDICT_REPLY_SET || parent->type == REDICT_REPLY_PUSH); parent->element[task->idx] = r; } return r; } static void *createDoubleObject(const redictReadTask *task, double value, char *str, size_t len) { redictReply *r, *parent; if (len == SIZE_MAX) // Prevents hi_malloc(0) if len equals to SIZE_MAX return NULL; r = createReplyObject(REDICT_REPLY_DOUBLE); if (r == NULL) return NULL; r->dval = value; r->str = hi_malloc(len+1); if (r->str == NULL) { freeReplyObject(r); return NULL; } /* The double reply also has the original protocol string representing a * double as a null terminated string. This way the caller does not need * to format back for string conversion, especially since Redict does efforts * to make the string more human readable avoiding the calssical double * decimal string conversion artifacts. */ memcpy(r->str, str, len); r->str[len] = '\0'; r->len = len; if (task->parent) { parent = task->parent->obj; assert(parent->type == REDICT_REPLY_ARRAY || parent->type == REDICT_REPLY_MAP || parent->type == REDICT_REPLY_ATTR || parent->type == REDICT_REPLY_SET || parent->type == REDICT_REPLY_PUSH); parent->element[task->idx] = r; } return r; } static void *createNilObject(const redictReadTask *task) { redictReply *r, *parent; r = createReplyObject(REDICT_REPLY_NIL); if (r == NULL) return NULL; if (task->parent) { parent = task->parent->obj; assert(parent->type == REDICT_REPLY_ARRAY || parent->type == REDICT_REPLY_MAP || parent->type == REDICT_REPLY_ATTR || parent->type == REDICT_REPLY_SET || parent->type == REDICT_REPLY_PUSH); parent->element[task->idx] = r; } return r; } static void *createBoolObject(const redictReadTask *task, int bval) { redictReply *r, *parent; r = createReplyObject(REDICT_REPLY_BOOL); if (r == NULL) return NULL; r->integer = bval != 0; if (task->parent) { parent = task->parent->obj; assert(parent->type == REDICT_REPLY_ARRAY || parent->type == REDICT_REPLY_MAP || parent->type == REDICT_REPLY_ATTR || parent->type == REDICT_REPLY_SET || parent->type == REDICT_REPLY_PUSH); parent->element[task->idx] = r; } return r; } /* Return the number of digits of 'v' when converted to string in radix 10. * Implementation borrowed from link in redict/src/util.c:string2ll(). */ static uint32_t countDigits(uint64_t v) { uint32_t result = 1; for (;;) { if (v < 10) return result; if (v < 100) return result + 1; if (v < 1000) return result + 2; if (v < 10000) return result + 3; v /= 10000U; result += 4; } } /* Helper that calculates the bulk length given a certain string length. */ static size_t bulklen(size_t len) { return 1+countDigits(len)+2+len+2; } int redictvFormatCommand(char **target, const char *format, va_list ap) { const char *c = format; char *cmd = NULL; /* final command */ int pos; /* position in final command */ sds curarg, newarg; /* current argument */ int touched = 0; /* was the current argument touched? */ char **curargv = NULL, **newargv = NULL; int argc = 0; int totlen = 0; int error_type = 0; /* 0 = no error; -1 = memory error; -2 = format error */ int j; /* Abort if there is not target to set */ if (target == NULL) return -1; /* Build the command string accordingly to protocol */ curarg = sdsempty(); if (curarg == NULL) return -1; while(*c != '\0') { if (*c != '%' || c[1] == '\0') { if (*c == ' ') { if (touched) { newargv = hi_realloc(curargv,sizeof(char*)*(argc+1)); if (newargv == NULL) goto memory_err; curargv = newargv; curargv[argc++] = curarg; totlen += bulklen(sdslen(curarg)); /* curarg is put in argv so it can be overwritten. */ curarg = sdsempty(); if (curarg == NULL) goto memory_err; touched = 0; } } else { newarg = sdscatlen(curarg,c,1); if (newarg == NULL) goto memory_err; curarg = newarg; touched = 1; } } else { char *arg; size_t size; /* Set newarg so it can be checked even if it is not touched. */ newarg = curarg; switch(c[1]) { case 's': arg = va_arg(ap,char*); size = strlen(arg); if (size > 0) newarg = sdscatlen(curarg,arg,size); break; case 'b': arg = va_arg(ap,char*); size = va_arg(ap,size_t); if (size > 0) newarg = sdscatlen(curarg,arg,size); break; case '%': newarg = sdscat(curarg,"%"); break; default: /* Try to detect printf format */ { static const char intfmts[] = "diouxX"; static const char flags[] = "#0-+ "; char _format[16]; const char *_p = c+1; size_t _l = 0; va_list _cpy; /* Flags */ while (*_p != '\0' && strchr(flags,*_p) != NULL) _p++; /* Field width */ while (*_p != '\0' && isdigit((int) *_p)) _p++; /* Precision */ if (*_p == '.') { _p++; while (*_p != '\0' && isdigit((int) *_p)) _p++; } /* Copy va_list before consuming with va_arg */ va_copy(_cpy,ap); /* Make sure we have more characters otherwise strchr() accepts * '\0' as an integer specifier. This is checked after above * va_copy() to avoid UB in fmt_invalid's call to va_end(). */ if (*_p == '\0') goto fmt_invalid; /* Integer conversion (without modifiers) */ if (strchr(intfmts,*_p) != NULL) { va_arg(ap,int); goto fmt_valid; } /* Double conversion (without modifiers) */ if (strchr("eEfFgGaA",*_p) != NULL) { va_arg(ap,double); goto fmt_valid; } /* Size: char */ if (_p[0] == 'h' && _p[1] == 'h') { _p += 2; if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { va_arg(ap,int); /* char gets promoted to int */ goto fmt_valid; } goto fmt_invalid; } /* Size: short */ if (_p[0] == 'h') { _p += 1; if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { va_arg(ap,int); /* short gets promoted to int */ goto fmt_valid; } goto fmt_invalid; } /* Size: long long */ if (_p[0] == 'l' && _p[1] == 'l') { _p += 2; if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { va_arg(ap,long long); goto fmt_valid; } goto fmt_invalid; } /* Size: long */ if (_p[0] == 'l') { _p += 1; if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { va_arg(ap,long); goto fmt_valid; } goto fmt_invalid; } fmt_invalid: va_end(_cpy); goto format_err; fmt_valid: _l = (_p+1)-c; if (_l < sizeof(_format)-2) { memcpy(_format,c,_l); _format[_l] = '\0'; newarg = sdscatvprintf(curarg,_format,_cpy); /* Update current position (note: outer blocks * increment c twice so compensate here) */ c = _p-1; } va_end(_cpy); break; } } if (newarg == NULL) goto memory_err; curarg = newarg; touched = 1; c++; if (*c == '\0') break; } c++; } /* Add the last argument if needed */ if (touched) { newargv = hi_realloc(curargv,sizeof(char*)*(argc+1)); if (newargv == NULL) goto memory_err; curargv = newargv; curargv[argc++] = curarg; totlen += bulklen(sdslen(curarg)); } else { sdsfree(curarg); } /* Clear curarg because it was put in curargv or was free'd. */ curarg = NULL; /* Add bytes needed to hold multi bulk count */ totlen += 1+countDigits(argc)+2; /* Build the command at protocol level */ cmd = hi_malloc(totlen+1); if (cmd == NULL) goto memory_err; pos = sprintf(cmd,"*%d\r\n",argc); for (j = 0; j < argc; j++) { pos += sprintf(cmd+pos,"$%zu\r\n",sdslen(curargv[j])); memcpy(cmd+pos,curargv[j],sdslen(curargv[j])); pos += sdslen(curargv[j]); sdsfree(curargv[j]); cmd[pos++] = '\r'; cmd[pos++] = '\n'; } assert(pos == totlen); cmd[pos] = '\0'; hi_free(curargv); *target = cmd; return totlen; format_err: error_type = -2; goto cleanup; memory_err: error_type = -1; goto cleanup; cleanup: if (curargv) { while(argc--) sdsfree(curargv[argc]); hi_free(curargv); } sdsfree(curarg); hi_free(cmd); return error_type; } /* Format a command according to the Redict protocol. This function * takes a format similar to printf: * * %s represents a C null terminated string you want to interpolate * %b represents a binary safe string * * When using %b you need to provide both the pointer to the string * and the length in bytes as a size_t. Examples: * * len = redictFormatCommand(target, "GET %s", mykey); * len = redictFormatCommand(target, "SET %s %b", mykey, myval, myvallen); */ int redictFormatCommand(char **target, const char *format, ...) { va_list ap; int len; va_start(ap,format); len = redictvFormatCommand(target,format,ap); va_end(ap); /* The API says "-1" means bad result, but we now also return "-2" in some * cases. Force the return value to always be -1. */ if (len < 0) len = -1; return len; } /* Format a command according to the Redict protocol using an sds string and * sdscatfmt for the processing of arguments. This function takes the * number of arguments, an array with arguments and an array with their * lengths. If the latter is set to NULL, strlen will be used to compute the * argument lengths. */ long long redictFormatSdsCommandArgv(sds *target, int argc, const char **argv, const size_t *argvlen) { sds cmd, aux; unsigned long long totlen, len; int j; /* Abort on a NULL target */ if (target == NULL) return -1; /* Calculate our total size */ totlen = 1+countDigits(argc)+2; for (j = 0; j < argc; j++) { len = argvlen ? argvlen[j] : strlen(argv[j]); totlen += bulklen(len); } /* Use an SDS string for command construction */ cmd = sdsempty(); if (cmd == NULL) return -1; /* We already know how much storage we need */ aux = sdsMakeRoomFor(cmd, totlen); if (aux == NULL) { sdsfree(cmd); return -1; } cmd = aux; /* Construct command */ cmd = sdscatfmt(cmd, "*%i\r\n", argc); for (j=0; j < argc; j++) { len = argvlen ? argvlen[j] : strlen(argv[j]); cmd = sdscatfmt(cmd, "$%U\r\n", len); cmd = sdscatlen(cmd, argv[j], len); cmd = sdscatlen(cmd, "\r\n", sizeof("\r\n")-1); } assert(sdslen(cmd)==totlen); *target = cmd; return totlen; } void redictFreeSdsCommand(sds cmd) { sdsfree(cmd); } /* Format a command according to the Redict protocol. This function takes the * number of arguments, an array with arguments and an array with their * lengths. If the latter is set to NULL, strlen will be used to compute the * argument lengths. */ long long redictFormatCommandArgv(char **target, int argc, const char **argv, const size_t *argvlen) { char *cmd = NULL; /* final command */ size_t pos; /* position in final command */ size_t len, totlen; int j; /* Abort on a NULL target */ if (target == NULL) return -1; /* Calculate number of bytes needed for the command */ totlen = 1+countDigits(argc)+2; for (j = 0; j < argc; j++) { len = argvlen ? argvlen[j] : strlen(argv[j]); totlen += bulklen(len); } /* Build the command at protocol level */ cmd = hi_malloc(totlen+1); if (cmd == NULL) return -1; pos = sprintf(cmd,"*%d\r\n",argc); for (j = 0; j < argc; j++) { len = argvlen ? argvlen[j] : strlen(argv[j]); pos += sprintf(cmd+pos,"$%zu\r\n",len); memcpy(cmd+pos,argv[j],len); pos += len; cmd[pos++] = '\r'; cmd[pos++] = '\n'; } assert(pos == totlen); cmd[pos] = '\0'; *target = cmd; return totlen; } void redictFreeCommand(char *cmd) { hi_free(cmd); } void __redictSetError(redictContext *c, int type, const char *str) { size_t len; c->err = type; if (str != NULL) { len = strlen(str); len = len < (sizeof(c->errstr)-1) ? len : (sizeof(c->errstr)-1); memcpy(c->errstr,str,len); c->errstr[len] = '\0'; } else { /* Only REDICT_ERR_IO may lack a description! */ assert(type == REDICT_ERR_IO); strerror_r(errno, c->errstr, sizeof(c->errstr)); } } redictReader *redictReaderCreate(void) { return redictReaderCreateWithFunctions(&defaultFunctions); } static void redictPushAutoFree(void *privdata, void *reply) { (void)privdata; freeReplyObject(reply); } static redictContext *redictContextInit(void) { redictContext *c; c = hi_calloc(1, sizeof(*c)); if (c == NULL) return NULL; c->funcs = &redictContextDefaultFuncs; c->obuf = sdsempty(); c->reader = redictReaderCreate(); c->fd = REDICT_INVALID_FD; if (c->obuf == NULL || c->reader == NULL) { redictFree(c); return NULL; } return c; } void redictFree(redictContext *c) { if (c == NULL) return; if (c->funcs && c->funcs->close) { c->funcs->close(c); } sdsfree(c->obuf); redictReaderFree(c->reader); hi_free(c->tcp.host); hi_free(c->tcp.source_addr); hi_free(c->unix_sock.path); hi_free(c->connect_timeout); hi_free(c->command_timeout); hi_free(c->saddr); if (c->privdata && c->free_privdata) c->free_privdata(c->privdata); if (c->funcs && c->funcs->free_privctx) c->funcs->free_privctx(c->privctx); memset(c, 0xff, sizeof(*c)); hi_free(c); } redictFD redictFreeKeepFd(redictContext *c) { redictFD fd = c->fd; c->fd = REDICT_INVALID_FD; redictFree(c); return fd; } int redictReconnect(redictContext *c) { c->err = 0; memset(c->errstr, '\0', strlen(c->errstr)); if (c->privctx && c->funcs->free_privctx) { c->funcs->free_privctx(c->privctx); c->privctx = NULL; } if (c->funcs && c->funcs->close) { c->funcs->close(c); } sdsfree(c->obuf); redictReaderFree(c->reader); c->obuf = sdsempty(); c->reader = redictReaderCreate(); if (c->obuf == NULL || c->reader == NULL) { __redictSetError(c, REDICT_ERR_OOM, "Out of memory"); return REDICT_ERR; } int ret = REDICT_ERR; if (c->connection_type == REDICT_CONN_TCP) { ret = redictContextConnectBindTcp(c, c->tcp.host, c->tcp.port, c->connect_timeout, c->tcp.source_addr); } else if (c->connection_type == REDICT_CONN_UNIX) { ret = redictContextConnectUnix(c, c->unix_sock.path, c->connect_timeout); } else { /* Something bad happened here and shouldn't have. There isn't enough information in the context to reconnect. */ __redictSetError(c,REDICT_ERR_OTHER,"Not enough information to reconnect"); ret = REDICT_ERR; } if (c->command_timeout != NULL && (c->flags & REDICT_BLOCK) && c->fd != REDICT_INVALID_FD) { redictContextSetTimeout(c, *c->command_timeout); } return ret; } redictContext *redictConnectWithOptions(const redictOptions *options) { redictContext *c = redictContextInit(); if (c == NULL) { return NULL; } if (!(options->options & REDICT_OPT_NONBLOCK)) { c->flags |= REDICT_BLOCK; } if (options->options & REDICT_OPT_REUSEADDR) { c->flags |= REDICT_REUSEADDR; } if (options->options & REDICT_OPT_NOAUTOFREE) { c->flags |= REDICT_NO_AUTO_FREE; } if (options->options & REDICT_OPT_NOAUTOFREEREPLIES) { c->flags |= REDICT_NO_AUTO_FREE_REPLIES; } if (options->options & REDICT_OPT_PREFER_IPV4) { c->flags |= REDICT_PREFER_IPV4; } if (options->options & REDICT_OPT_PREFER_IPV6) { c->flags |= REDICT_PREFER_IPV6; } /* 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) redictSetPushCallback(c, options->push_cb); else if (!(options->options & REDICT_OPT_NO_PUSH_AUTOFREE)) redictSetPushCallback(c, redictPushAutoFree); c->privdata = options->privdata; c->free_privdata = options->free_privdata; if (redictContextUpdateConnectTimeout(c, options->connect_timeout) != REDICT_OK || redictContextUpdateCommandTimeout(c, options->command_timeout) != REDICT_OK) { __redictSetError(c, REDICT_ERR_OOM, "Out of memory"); return c; } if (options->type == REDICT_CONN_TCP) { redictContextConnectBindTcp(c, options->endpoint.tcp.ip, options->endpoint.tcp.port, options->connect_timeout, options->endpoint.tcp.source_addr); } else if (options->type == REDICT_CONN_UNIX) { redictContextConnectUnix(c, options->endpoint.unix_socket, options->connect_timeout); } else if (options->type == REDICT_CONN_USERFD) { c->fd = options->endpoint.fd; c->flags |= REDICT_CONNECTED; } else { redictFree(c); return NULL; } if (c->err == 0 && c->fd != REDICT_INVALID_FD && options->command_timeout != NULL && (c->flags & REDICT_BLOCK)) { redictContextSetTimeout(c, *options->command_timeout); } return c; } /* Connect to a Redict instance. On error the field error in the returned * context will be set to the return value of the error function. * When no set of reply functions is given, the default set will be used. */ redictContext *redictConnect(const char *ip, int port) { redictOptions options = {0}; REDICT_OPTIONS_SET_TCP(&options, ip, port); return redictConnectWithOptions(&options); } redictContext *redictConnectWithTimeout(const char *ip, int port, const struct timeval tv) { redictOptions options = {0}; REDICT_OPTIONS_SET_TCP(&options, ip, port); options.connect_timeout = &tv; return redictConnectWithOptions(&options); } redictContext *redictConnectNonBlock(const char *ip, int port) { redictOptions options = {0}; REDICT_OPTIONS_SET_TCP(&options, ip, port); options.options |= REDICT_OPT_NONBLOCK; return redictConnectWithOptions(&options); } redictContext *redictConnectBindNonBlock(const char *ip, int port, const char *source_addr) { redictOptions options = {0}; REDICT_OPTIONS_SET_TCP(&options, ip, port); options.endpoint.tcp.source_addr = source_addr; options.options |= REDICT_OPT_NONBLOCK; return redictConnectWithOptions(&options); } redictContext *redictConnectBindNonBlockWithReuse(const char *ip, int port, const char *source_addr) { redictOptions options = {0}; REDICT_OPTIONS_SET_TCP(&options, ip, port); options.endpoint.tcp.source_addr = source_addr; options.options |= REDICT_OPT_NONBLOCK|REDICT_OPT_REUSEADDR; return redictConnectWithOptions(&options); } redictContext *redictConnectUnix(const char *path) { redictOptions options = {0}; REDICT_OPTIONS_SET_UNIX(&options, path); return redictConnectWithOptions(&options); } redictContext *redictConnectUnixWithTimeout(const char *path, const struct timeval tv) { redictOptions options = {0}; REDICT_OPTIONS_SET_UNIX(&options, path); options.connect_timeout = &tv; return redictConnectWithOptions(&options); } redictContext *redictConnectUnixNonBlock(const char *path) { redictOptions options = {0}; REDICT_OPTIONS_SET_UNIX(&options, path); options.options |= REDICT_OPT_NONBLOCK; return redictConnectWithOptions(&options); } redictContext *redictConnectFd(redictFD fd) { redictOptions options = {0}; options.type = REDICT_CONN_USERFD; options.endpoint.fd = fd; return redictConnectWithOptions(&options); } /* Set read/write timeout on a blocking socket. */ int redictSetTimeout(redictContext *c, const struct timeval tv) { if (c->flags & REDICT_BLOCK) return redictContextSetTimeout(c,tv); return REDICT_ERR; } int redictEnableKeepAliveWithInterval(redictContext *c, int interval) { return redictKeepAlive(c, interval); } /* Enable connection KeepAlive. */ int redictEnableKeepAlive(redictContext *c) { return redictKeepAlive(c, REDICT_KEEPALIVE_INTERVAL); } /* Set the socket option TCP_USER_TIMEOUT. */ int redictSetTcpUserTimeout(redictContext *c, unsigned int timeout) { return redictContextSetTcpUserTimeout(c, timeout); } /* Set a user provided RESP3 PUSH handler and return any old one set. */ redictPushFn *redictSetPushCallback(redictContext *c, redictPushFn *fn) { redictPushFn *old = c->push_cb; c->push_cb = fn; return old; } /* 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. * * After this function is called, you may use redictGetReplyFromReader to * see if there is a reply available. */ int redictBufferRead(redictContext *c) { char buf[1024*16]; int nread; /* Return early when the context has seen an error. */ if (c->err) return REDICT_ERR; nread = c->funcs->read(c, buf, sizeof(buf)); if (nread < 0) { return REDICT_ERR; } if (nread > 0 && redictReaderFeed(c->reader, buf, nread) != REDICT_OK) { __redictSetError(c, c->reader->err, c->reader->errstr); return REDICT_ERR; } return REDICT_OK; } /* Write the output buffer to the socket. * * Returns REDICT_OK when the buffer is empty, or (a part of) the buffer was * successfully written to the socket. When the buffer is empty after the * write operation, "done" is set to 1 (if given). * * Returns REDICT_ERR if an unrecoverable error occurred in the underlying * c->funcs->write function. */ int redictBufferWrite(redictContext *c, int *done) { /* Return early when the context has seen an error. */ if (c->err) return REDICT_ERR; if (sdslen(c->obuf) > 0) { ssize_t nwritten = c->funcs->write(c); if (nwritten < 0) { return REDICT_ERR; } else if (nwritten > 0) { if (nwritten == (ssize_t)sdslen(c->obuf)) { sdsfree(c->obuf); c->obuf = sdsempty(); if (c->obuf == NULL) goto oom; } else { if (sdsrange(c->obuf,nwritten,-1) < 0) goto oom; } } } if (done != NULL) *done = (sdslen(c->obuf) == 0); return REDICT_OK; oom: __redictSetError(c, REDICT_ERR_OOM, "Out of memory"); return REDICT_ERR; } /* Internal helper that returns 1 if the reply was a RESP3 PUSH * message and we handled it with a user-provided callback. */ static int redictHandledPushReply(redictContext *c, void *reply) { if (reply && c->push_cb && redictIsPushReply(reply)) { c->push_cb(c->privdata, reply); return 1; } return 0; } /* Get a reply from our reader or set an error in the context. */ int redictGetReplyFromReader(redictContext *c, void **reply) { if (redictReaderGetReply(c->reader, reply) == REDICT_ERR) { __redictSetError(c,c->reader->err,c->reader->errstr); return REDICT_ERR; } return REDICT_OK; } /* Internal helper to get the next reply from our reader while handling * any PUSH messages we encounter along the way. This is separate from * redictGetReplyFromReader so as to not change its behavior. */ static int redictNextInBandReplyFromReader(redictContext *c, void **reply) { do { if (redictGetReplyFromReader(c, reply) == REDICT_ERR) return REDICT_ERR; } while (redictHandledPushReply(c, *reply)); return REDICT_OK; } int redictGetReply(redictContext *c, void **reply) { int wdone = 0; void *aux = NULL; /* Try to read pending replies */ if (redictNextInBandReplyFromReader(c,&aux) == REDICT_ERR) return REDICT_ERR; /* For the blocking context, flush output buffer and read reply */ if (aux == NULL && c->flags & REDICT_BLOCK) { /* Write until done */ do { if (redictBufferWrite(c,&wdone) == REDICT_ERR) return REDICT_ERR; } while (!wdone); /* Read until there is a reply */ do { if (redictBufferRead(c) == REDICT_ERR) return REDICT_ERR; if (redictNextInBandReplyFromReader(c,&aux) == REDICT_ERR) return REDICT_ERR; } while (aux == NULL); } /* Set reply or free it if we were passed NULL */ if (reply != NULL) { *reply = aux; } else { freeReplyObject(aux); } return REDICT_OK; } /* Helper function for the redictAppendCommand* family of functions. * * Write a formatted command to the output buffer. When this family * is used, you need to call redictGetReply yourself to retrieve * the reply (or replies in pub/sub). */ int __redictAppendCommand(redictContext *c, const char *cmd, size_t len) { sds newbuf; newbuf = sdscatlen(c->obuf,cmd,len); if (newbuf == NULL) { __redictSetError(c,REDICT_ERR_OOM,"Out of memory"); return REDICT_ERR; } c->obuf = newbuf; return REDICT_OK; } int redictAppendFormattedCommand(redictContext *c, const char *cmd, size_t len) { if (__redictAppendCommand(c, cmd, len) != REDICT_OK) { return REDICT_ERR; } return REDICT_OK; } int redictvAppendCommand(redictContext *c, const char *format, va_list ap) { char *cmd; int len; len = redictvFormatCommand(&cmd,format,ap); if (len == -1) { __redictSetError(c,REDICT_ERR_OOM,"Out of memory"); return REDICT_ERR; } else if (len == -2) { __redictSetError(c,REDICT_ERR_OTHER,"Invalid format string"); return REDICT_ERR; } if (__redictAppendCommand(c,cmd,len) != REDICT_OK) { hi_free(cmd); return REDICT_ERR; } hi_free(cmd); return REDICT_OK; } int redictAppendCommand(redictContext *c, const char *format, ...) { va_list ap; int ret; va_start(ap,format); ret = redictvAppendCommand(c,format,ap); va_end(ap); return ret; } int redictAppendCommandArgv(redictContext *c, int argc, const char **argv, const size_t *argvlen) { sds cmd; long long len; len = redictFormatSdsCommandArgv(&cmd,argc,argv,argvlen); if (len == -1) { __redictSetError(c,REDICT_ERR_OOM,"Out of memory"); return REDICT_ERR; } if (__redictAppendCommand(c,cmd,len) != REDICT_OK) { sdsfree(cmd); return REDICT_ERR; } sdsfree(cmd); return REDICT_OK; } /* Helper function for the redictCommand* family of functions. * * Write a formatted command to the output buffer. If the given context is * blocking, immediately read the reply into the "reply" pointer. When the * context is non-blocking, the "reply" pointer will not be used and the * command is simply appended to the write buffer. * * Returns the reply when a reply was successfully retrieved. Returns NULL * otherwise. When NULL is returned in a blocking context, the error field * in the context will be set. */ static void *__redictBlockForReply(redictContext *c) { void *reply; if (c->flags & REDICT_BLOCK) { if (redictGetReply(c,&reply) != REDICT_OK) return NULL; return reply; } return NULL; } void *redictvCommand(redictContext *c, const char *format, va_list ap) { if (redictvAppendCommand(c,format,ap) != REDICT_OK) return NULL; return __redictBlockForReply(c); } void *redictCommand(redictContext *c, const char *format, ...) { va_list ap; va_start(ap,format); void *reply = redictvCommand(c,format,ap); va_end(ap); return reply; } void *redictCommandArgv(redictContext *c, int argc, const char **argv, const size_t *argvlen) { if (redictAppendCommandArgv(c,argc,argv,argvlen) != REDICT_OK) return NULL; return __redictBlockForReply(c); }