library.c

#include "php_redis.h"

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "common.h"
#include "php_network.h"
#include <sys/types.h>
#include <errno.h>
#include <limits.h>

#ifdef HAVE_REDIS_IGBINARY
#include "igbinary/igbinary.h"
#endif
#ifdef HAVE_REDIS_MSGPACK
#include "msgpack/php_msgpack.h"
#endif

#ifdef HAVE_REDIS_LZF
#include <lzf.h>

    #ifndef LZF_MARGIN
        #define LZF_MARGIN 128
    #endif
#endif

#ifdef HAVE_REDIS_ZSTD
#include <zstd.h>
#endif

#ifdef HAVE_REDIS_LZ4
#include <lz4.h>
#include <lz4hc.h>

/* uint8_t crf + int length */
#define REDIS_LZ4_HDR_SIZE (sizeof(uint8_t) + sizeof(int))
#if defined(LZ4HC_CLEVEL_MAX)
/* version >= 1.7.5 */
#define REDIS_LZ4_MAX_CLEVEL LZ4HC_CLEVEL_MAX

#elif defined (LZ4HC_MAX_CLEVEL)
/* version >= 1.7.3 */
#define REDIS_LZ4_MAX_CLEVEL LZ4HC_MAX_CLEVEL

#else
/* older versions */
#define REDIS_LZ4_MAX_CLEVEL 12
#endif
#endif

#include <zend_exceptions.h>
#include "library.h"
#include "redis_commands.h"

#ifdef HAVE_REDIS_JSON
#include <ext/json/php_json.h>
#endif

#include <ext/hash/php_hash.h>

#define UNSERIALIZE_NONE 0
#define UNSERIALIZE_KEYS 1
#define UNSERIALIZE_VALS 2
#define UNSERIALIZE_ALL  3

#define SCORE_DECODE_NONE 0
#define SCORE_DECODE_INT  1
#define SCORE_DECODE_DOUBLE 2

#define REDIS_CALLBACKS_INIT_SIZE      8
#define REDIS_CALLBACKS_MAX_DOUBLE 32768
#define REDIS_CALLBACKS_ADD_SIZE    4096

/* PhpRedis often returns either FALSE or NULL depending on whether we have
 * an option set, so this macro just wraps that often repeated logic */
#define REDIS_ZVAL_NULL(sock_, zv_) \
    do { \
        if ((sock_)->null_mbulk_as_null) { \
            ZVAL_NULL((zv_)); \
        } else { \
            ZVAL_FALSE((zv_)); \
        } \
    } while (0)

/** Set return value to false in case of we are in atomic mode or add FALSE to output array in pipeline mode */
#define REDIS_RESPONSE_ERROR(redis_sock, z_tab) \
    do { \
        if (redis_sock_is_atomic(redis_sock)) { \
            RETVAL_FALSE; \
        } else { \
            add_next_index_bool(z_tab, 0); \
        } \
    } while (0)

/** Set return value to `zval` in case of we are in atomic mode or add `zval` to output array in pipeline mode */
#define REDIS_RETURN_ZVAL(redis_sock, z_tab, zval) \
    do { \
        if (redis_sock_is_atomic(redis_sock)) { \
            /* Move value of `zval` to `return_value` */ \
            ZVAL_COPY_VALUE(return_value, &zval); \
        } else { \
            zend_hash_next_index_insert_new(Z_ARRVAL_P(z_tab), &zval); \
        } \
    } while (0)

#ifndef PHP_WIN32
    #include <netinet/tcp.h> /* TCP_NODELAY */
    #include <sys/socket.h>  /* SO_KEEPALIVE */
#else
    #include <winsock.h>
#endif

extern zend_class_entry *redis_ce;
extern zend_class_entry *redis_exception_ce;

extern int le_redis_pconnect;

RedisCmdCtx redis_empty_ctx = {0};

static int redis_mbulk_reply_zipped_raw_variant(RedisSock *redis_sock, zval *zret, int count);

static int redis_bulk_resp_to_zval(RedisSock *redis_sock, zval *zdst,
    int *dstlen) ;

static ConnectionPool *
redis_sock_get_connection_pool(RedisSock *redis_sock)
{
    ConnectionPool *pool;
    zend_resource *le;
    zend_string *persistent_id;

    /* Generate our unique pool id depending on configuration */
    persistent_id = redis_pool_spprintf(redis_sock, zend_ini_string_literal("redis.pconnect.pool_pattern"));

    /* Return early if we can find the pool */
    if ((le = zend_hash_find_ptr(&EG(persistent_list), persistent_id))) {
        zend_string_release(persistent_id);
        return le->ptr;
    }

    /* Create the pool and store it in our persistent list */
    pool = pecalloc(1, sizeof(*pool), 1);
    zend_llist_init(&pool->list, sizeof(php_stream *), NULL, 1);
    zend_register_persistent_resource(ZSTR_VAL(persistent_id),
                                      ZSTR_LEN(persistent_id),
                                      pool, le_redis_pconnect);

    zend_string_release(persistent_id);

    return pool;
}

static int redis_sock_response_ok(RedisSock *redis_sock, char *buf, int buf_size) {
    size_t len;
    if (UNEXPECTED(redis_sock_gets(redis_sock, buf, buf_size - 1, &len) < 0)) {
        return 0;
    }
    if (UNEXPECTED(redis_strncmp(buf, ZEND_STRL("+OK")))) {
        if (buf[0] == '-') {
            redis_sock_set_err(redis_sock, buf + 1, len - 1);
        }
        return 0;
    }
    return 1;
}

/* Helper to select the proper DB number */
static int redis_select_db(RedisSock *redis_sock) {
    char response[4096];
    RedisCmd *cmd;

    cmd = redis_cmd_create_literal(redis_sock, "SELECT");

    redis_cmd_cat_long(cmd, redis_sock->dbNumber);

    if (redis_sock_write(redis_sock, redis_cmd_str(cmd), redis_cmd_len(cmd)) < 0) {
        redis_cmd_free(cmd);
        return -1;
    }

    redis_cmd_free(cmd);

    if (!redis_sock_response_ok(redis_sock, response, sizeof(response))) {
        return -1;
    }

    return 0;
}

/* Append an AUTH command to a smart string if necessary.  This will either
 * append the new style AUTH <user> <password>, old style AUTH <password>, or
 * append no command at all.  Function returns 1 if we appended a command
 * and 0 otherwise. */
static int redis_sock_append_auth(RedisSock *redis_sock, smart_str *str) {
    int argc;

    /* We need a password at least */
    if (redis_sock->pass == NULL)
        return 0;

    argc = !!redis_sock->user + !!redis_sock->pass;

    resp_str_cat_cmd_literal(str, "AUTH", argc);

    if (redis_sock->user)
        resp_str_cat_zstr(str, redis_sock->user);

    resp_str_cat_zstr(str, redis_sock->pass);

    /* We appended a command */
    return 1;
}

PHP_REDIS_API void
redis_sock_set_auth(RedisSock *redis_sock, zend_string *user, zend_string *pass)
{
    /* Release existing user/pass */
    redis_sock_free_auth(redis_sock);

    /* Set new user/pass */
    redis_sock->user = user ? zend_string_copy(user) : NULL;
    redis_sock->pass = pass ? zend_string_copy(pass) : NULL;
}

PHP_REDIS_API void
redis_sock_set_auth_zval(RedisSock *redis_sock, zval *zv) {
    zend_string *user, *pass;

    if (redis_extract_auth_info(zv, &user, &pass) == FAILURE)
        return;

    redis_sock_set_auth(redis_sock, user, pass);

    if (user) zend_string_release(user);
    if (pass) zend_string_release(pass);
}

PHP_REDIS_API void
redis_sock_free_auth(RedisSock *redis_sock) {
    if (redis_sock->user) {
        zend_string_release(redis_sock->user);
        redis_sock->user = NULL;
    }

    if (redis_sock->pass) {
        zend_string_release(redis_sock->pass);
        redis_sock->pass = NULL;
    }
}

#if PHP_VERSION_ID < 80000
static void redis_array_release(HashTable *ht) {
    if (!ht || (GC_FLAGS(ht) & GC_IMMUTABLE)) {
        return;
    }

    if (GC_DELREF(ht) == 0) {
        zend_array_destroy(ht);
    }
}
#else
static inline void redis_array_release(HashTable *ht) {
    zend_array_release(ht);
}
#endif


void redis_sock_free_context(RedisSock *redis_sock) {
    if (redis_sock->context == NULL)
        return;

    redis_array_release(redis_sock->context);
    redis_sock->context = NULL;
}

PHP_REDIS_API zend_string *
redis_sock_auth_cmd(RedisSock *redis_sock) {
    smart_str cmd = {0};

    if (redis_sock_append_auth(redis_sock, &cmd) == 0) {
        return NULL;
    }

    return cmd.s;
}

/* Send Redis AUTH and process response */
PHP_REDIS_API int redis_sock_auth(RedisSock *redis_sock) {
    zend_string *cmd;
    char inbuf[4096];

    if ((cmd = redis_sock_auth_cmd(redis_sock)) == NULL)
        return SUCCESS;

    if (redis_sock_write(redis_sock, ZSTR_VAL(cmd), ZSTR_LEN(cmd)) < 0) {
        zend_string_release(cmd);
        return FAILURE;
    }

    zend_string_release(cmd);

    if (!redis_sock_response_ok(redis_sock, inbuf, sizeof(inbuf))) {
        return FAILURE;
    }

    return SUCCESS;
}

/* Helper function to test a RedisSock error prefix. */
static zend_always_inline zend_bool
redis_sock_errcmp(RedisSock *redis_sock, const char *err, size_t errlen)
{
    return ZSTR_LEN(redis_sock->err) >= errlen &&
           memcmp(ZSTR_VAL(redis_sock->err), err, errlen) == 0;
}

/* Helper function that will throw an exception for a small number of ERR codes
 * returned by Redis.  Typically we just return FALSE to the caller in the event
 * of an ERROR reply, but for the following error types:
 *    1) MASTERDOWN
 *    2) AUTH
 *    3) LOADING
 */
static void
redis_error_throw(RedisSock *redis_sock)
{
    /* Short circuit if we have no redis_sock or any error */
    if (redis_sock == NULL || redis_sock->err == NULL)
        return;

    /* Redis 6 decided to add 'ERR AUTH' which has a normal 'ERR' prefix
     * but is actually an authentication error that we will want to throw
     * an exception for, so just short circuit if this is any other 'ERR'
     * prefixed error. */
    if (redis_sock_errcmp(redis_sock, ZEND_STRL("ERR")) &&
        !redis_sock_errcmp(redis_sock, ZEND_STRL("ERR AUTH"))) return;

    /* We may want to flip this logic and check for MASTERDOWN, AUTH,
     * and LOADING but that may have side effects (esp for things like
     * Disque) */
    if (!redis_sock_errcmp(redis_sock, ZEND_STRL("NOSCRIPT")) &&
        !redis_sock_errcmp(redis_sock, ZEND_STRL("NOQUORUM")) &&
        !redis_sock_errcmp(redis_sock, ZEND_STRL("NOGOODSLAVE")) &&
        !redis_sock_errcmp(redis_sock, ZEND_STRL("WRONGTYPE")) &&
        !redis_sock_errcmp(redis_sock, ZEND_STRL("BUSYGROUP")) &&
        !redis_sock_errcmp(redis_sock, ZEND_STRL("NOGROUP")))
    {
        REDIS_THROW_EXCEPTION(ZSTR_VAL(redis_sock->err), 0);
    }
}

static int
read_mbulk_header(RedisSock *redis_sock, int *nelem)
{
    char line[4096];
    size_t len;

    /* Throws exception on failure */
    if (redis_sock_gets(redis_sock, line, sizeof(line) - 1, &len) < 0) {
        return FAILURE;
    }

    if (*line != TYPE_MULTIBULK) {
        if (*line == TYPE_ERR) {
            redis_sock_set_err(redis_sock, line + 1, len - 1);
        }
        return FAILURE;
    }

    {
        char *endptr;
        long long n;

        errno = 0;
        n = strtoll(line + 1, &endptr, 10);
        if (endptr == line + 1 || errno == ERANGE || n < -1 || n > INT_MAX) {
            return FAILURE;
        }
        *nelem = (int)n;
    }

    return SUCCESS;
}

PHP_REDIS_API int
redis_check_eof(RedisSock *redis_sock, zend_bool no_retry, zend_bool no_throw)
{
    unsigned int retry_index;
    char *errmsg;

    if (!redis_sock || !redis_sock->stream || redis_sock->status == REDIS_SOCK_STATUS_FAILED) {
        if (!no_throw) {
            REDIS_THROW_EXCEPTION( "Connection closed", 0);
        }
        return -1;
    }

    /* NOITCE: set errno = 0 here
     *
     * There is a bug in php socket stream to check liveness of a connection:
     * if (0 >= recv(sock->socket, &buf, sizeof(buf), MSG_PEEK) && php_socket_errno() != EWOULDBLOCK) {
     *    alive = 0;
     * }
     * If last errno is EWOULDBLOCK and recv returns 0 because of connection closed, alive would not be
     * set to 0. However, the connection is close indeed. The php_stream_eof is not reliable. This will
     * cause a "read error on connection" exception when use a closed persistent connection.
     *
     * We work around this by set errno = 0 first.
     *
     * Bug fix of php: https://github.com/php/php-src/pull/1456
     * */
    errno = 0;
    if (php_stream_eof(redis_sock->stream) == 0) {
        /* Success */
        return 0;
    } else if (redis_sock->mode == MULTI || redis_sock->watching) {
        errmsg = "Connection lost and socket is in MULTI/watching mode";
    } else {
        errmsg = "Connection lost";
        redis_backoff_reset(&redis_sock->backoff);
        for (retry_index = 0; !no_retry && retry_index < redis_sock->max_retries; ++retry_index) {
            /* close existing stream before reconnecting */
            if (redis_sock->stream) {
                /* reconnect no need to reset mode, it will cause pipeline mode socket exception */
                redis_sock_disconnect(redis_sock, 1, 0);
            }
            /* Sleep based on our backoff algorithm */
            zend_ulong delay = redis_backoff_compute(&redis_sock->backoff, retry_index);
            if (delay != 0)
                usleep(delay);

            /* reconnect */
            if (redis_sock_connect(redis_sock) == 0) {
                /* check for EOF again. */
                errno = 0;
                if (php_stream_eof(redis_sock->stream) == 0) {
                    if (redis_sock_auth(redis_sock) != SUCCESS) {
                        errmsg = "AUTH failed while reconnecting";
                        break;
                    }
                    redis_sock->status = REDIS_SOCK_STATUS_AUTHENTICATED;

                    /* If we're using a non-zero db, reselect it */
                    if (redis_sock->dbNumber && redis_select_db(redis_sock) != 0) {
                        errmsg = "SELECT failed while reconnecting";
                        break;
                    }
                    redis_sock->status = REDIS_SOCK_STATUS_READY;
                    /* Success */
                    return 0;
                }
            }
        }
    }
    /* close stream and mark socket as failed */
    redis_sock_disconnect(redis_sock, 1, 1);
    redis_sock->status = REDIS_SOCK_STATUS_FAILED;
    if (!no_throw) {
        REDIS_THROW_EXCEPTION( errmsg, 0);
    }
    return -1;
}

PHP_REDIS_API int
redis_sock_read_scan_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                           REDIS_SCAN_TYPE type, uint64_t *cursor)
{
    REDIS_REPLY_TYPE reply_type;
    char err[4096], *p_iter;
    RedisCmdCtx ctx = {0};
    long reply_info;
    size_t errlen;

    /* Our response should have two multibulk replies */
    if(redis_read_reply_type(redis_sock, &reply_type, &reply_info)<0
       || reply_type != TYPE_MULTIBULK || reply_info != 2)
    {
        if (reply_type == TYPE_ERR) {
            if (redis_sock_gets(redis_sock, err, sizeof(err), &errlen) == 0) {
                redis_sock_set_err(redis_sock, err, errlen);
            }
        }

        return -1;
    }

    /* The BULK response iterator */
    if(redis_read_reply_type(redis_sock, &reply_type, &reply_info)<0
       || reply_type != TYPE_BULK)
    {
        return -1;
    }

    /* Attempt to read the iterator */
    if(!(p_iter = redis_sock_read_bulk_reply(redis_sock, reply_info))) {
        return -1;
    }

    /* Push the iterator out to the caller */
    *cursor = strtoull(p_iter, NULL, 10);
    efree(p_iter);

    /* Read our actual keys/members/etc differently depending on what kind of
       scan command this is.  They all come back in slightly different ways */
    switch(type) {
        case TYPE_SCAN:
            return redis_mbulk_reply_raw(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                redis_sock, NULL, ctx);
        case TYPE_SSCAN:
            return redis_sock_read_multibulk_reply(
                INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock, NULL, ctx);
        case TYPE_ZSCAN:
            return redis_mbulk_reply_zipped_keys_dbl(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                redis_sock, NULL, ctx);
        case TYPE_HSCAN:
            return redis_mbulk_reply_zipped_vals(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                redis_sock, NULL, ctx);
        default:
            return -1;
    }
}

PHP_REDIS_API int
redis_pubsub_response(INTERNAL_FUNCTION_PARAMETERS,
                      RedisSock *redis_sock, zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_long_response(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                   z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_read_variant_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                        redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        return redis_mbulk_reply_zipped_keys_int(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                                 redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

static void
ht_free_subs(zval *data)
{
    efree(Z_PTR_P(data));
}

PHP_REDIS_API int
redis_subscribe_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                         zval *z_tab, RedisCmdCtx ctx)
{
    HashTable *subs;
    subscribeCallback *cb;
    subscribeContext *sctx = ctx.ptr;
    zval *z_tmp, z_resp;
    int i;

    ALLOC_HASHTABLE(subs);
    zend_hash_init(subs, 0, NULL, ht_free_subs, 0);
    // Consume response(s) from subscribe, which will vary on argc
    while(sctx->argc--) {
        ZVAL_NULL(&z_resp);
        if (!redis_sock_read_multibulk_reply_zval(redis_sock, &z_resp)) {
            goto error;
        }

        // We'll need to find the command response
        if ((z_tmp = zend_hash_index_find(Z_ARRVAL(z_resp), 0)) == NULL) {
            goto error;
        }

        // Make sure the command response matches the command we called
        if(strcasecmp(Z_STRVAL_P(z_tmp), sctx->kw) !=0) {
            goto error;
        }

        if ((z_tmp = zend_hash_index_find(Z_ARRVAL(z_resp), 1)) == NULL) {
            goto error;
        }

        zend_hash_str_update_mem(subs, Z_STRVAL_P(z_tmp), Z_STRLEN_P(z_tmp),
                                    &sctx->cb, sizeof(sctx->cb));

        zval_ptr_dtor_nogc(&z_resp);
    }

    if (strcasecmp(sctx->kw, "ssubscribe") == 0) {
        i = REDIS_SSUBSCRIBE_IDX;
    } else if (strcasecmp(sctx->kw, "psubscribe") == 0) {
        i = REDIS_PSUBSCRIBE_IDX;
    } else {
        i = REDIS_SUBSCRIBE_IDX;
    }

    if (redis_sock->subs[i]) {
        zend_string *zkey;

        ZEND_HASH_FOREACH_STR_KEY_PTR(subs, zkey, cb) {
            zend_hash_update_mem(redis_sock->subs[i], zkey, cb, sizeof(*cb));
        } ZEND_HASH_FOREACH_END();
        zend_hash_destroy(subs);
        efree(subs);

        RETVAL_TRUE;
        return SUCCESS;
    }

    redis_sock->subs[i] = subs;
    /* Multibulk response, {[pattern], type, channel, payload } */
    while (redis_sock->subs[i]) {
        zval z_ret, z_args[4], *z_type, *z_chan, *z_pat = NULL, *z_data;
        int tab_idx = 1, is_pmsg = 0;
        HashTable *ht_tab;
        zend_string *zs;

        ZVAL_NULL(&z_resp);
        if (!redis_sock_read_multibulk_reply_zval(redis_sock, &z_resp)) {
            goto failure;
        }

        ht_tab = Z_ARRVAL(z_resp);

        if ((z_type = zend_hash_index_find(ht_tab, 0)) == NULL ||
           Z_TYPE_P(z_type) != IS_STRING
        ) {
            goto failure;
        }

        // Check for message or pmessage
        if (zend_string_equals_literal_ci(Z_STR_P(z_type), "message") ||
            zend_string_equals_literal_ci(Z_STR_P(z_type), "pmessage") ||
            zend_string_equals_literal_ci(Z_STR_P(z_type), "smessage")
        ) {
            is_pmsg = *Z_STRVAL_P(z_type)=='p';
        } else {
            zval_ptr_dtor_nogc(&z_resp);
            continue;
        }

        // Extract pattern if it's a pmessage
        if (is_pmsg) {
            z_pat = zend_hash_index_find(ht_tab, tab_idx++);
            if (z_pat == NULL || Z_TYPE_P(z_pat) != IS_STRING)
                goto failure;
        }

        /* Extract channel */
        z_chan = zend_hash_index_find(ht_tab, tab_idx++);
        if (z_chan == NULL || Z_TYPE_P(z_chan) != IS_STRING)
            goto failure;

        /* Finally, extract data */
        z_data = zend_hash_index_find(ht_tab, tab_idx++);
        if (z_data == NULL)
            goto failure;

        /* Find our callback, either by channel or pattern string */
        zs = z_pat != NULL ? Z_STR_P(z_pat) : Z_STR_P(z_chan);
        if ((cb = zend_hash_find_ptr(redis_sock->subs[i], zs)) == NULL)
            goto failure;

        // Different args for SUBSCRIBE and PSUBSCRIBE
        z_args[0] = *getThis();
        if(is_pmsg) {
            z_args[1] = *z_pat;
            z_args[2] = *z_chan;
            z_args[3] = *z_data;
        } else {
            z_args[1] = *z_chan;
            z_args[2] = *z_data;
        }

        // Set arg count
        cb->fci.param_count = tab_idx;
        cb->fci.retval = &z_ret;
        cb->fci.params = z_args;

        // Execute callback
        if (zend_call_function(&cb->fci, &cb->fci_cache) != SUCCESS) {
            goto failure;
        }

        // If we have a return value free it
        zval_ptr_dtor(&z_ret);
        zval_ptr_dtor_nogc(&z_resp);
    }

    RETVAL_TRUE;
    return SUCCESS;

    // This is an error state, clean up
error:
    zend_hash_destroy(subs);
    efree(subs);
failure:
    zval_ptr_dtor_nogc(&z_resp);
    RETVAL_FALSE;
    return FAILURE;
}

PHP_REDIS_API int redis_unsubscribe_response(INTERNAL_FUNCTION_PARAMETERS,
                                      RedisSock *redis_sock, zval *z_tab,
                                      RedisCmdCtx ctx)
{
    subscribeContext *sctx = ctx.ptr;
    zval *z_chan, z_ret, z_resp;
    int i;

    if (strcasecmp(sctx->kw, "sunsubscribe") == 0) {
        i = REDIS_SSUBSCRIBE_IDX;
    } else if (strcasecmp(sctx->kw, "punsubscribe") == 0) {
        i = REDIS_PSUBSCRIBE_IDX;
    } else {
        i = REDIS_SUBSCRIBE_IDX;
    }
    if (!sctx->argc && redis_sock->subs[i]) {
        sctx->argc = zend_hash_num_elements(redis_sock->subs[i]);
    }

    array_init_size(&z_ret, sctx->argc);

    while (sctx->argc--) {
        ZVAL_NULL(&z_resp);
        if (!redis_sock_read_multibulk_reply_zval(redis_sock, &z_resp) ||
            (z_chan = zend_hash_index_find(Z_ARRVAL(z_resp), 1)) == NULL
        ) {
            zval_ptr_dtor_nogc(&z_resp);
            zval_ptr_dtor_nogc(&z_ret);
            RETVAL_FALSE;
            return FAILURE;
        }

        if (!redis_sock->subs[i] ||
            !zend_hash_str_exists(redis_sock->subs[i], Z_STRVAL_P(z_chan), Z_STRLEN_P(z_chan))
        ) {
            add_assoc_bool_ex(&z_ret, Z_STRVAL_P(z_chan), Z_STRLEN_P(z_chan), 0);
        } else {
            zend_hash_str_del(redis_sock->subs[i], Z_STRVAL_P(z_chan), Z_STRLEN_P(z_chan));
            add_assoc_bool_ex(&z_ret, Z_STRVAL_P(z_chan), Z_STRLEN_P(z_chan), 1);
        }

        zval_ptr_dtor_nogc(&z_resp);
    }

    if (redis_sock->subs[i] && !zend_hash_num_elements(redis_sock->subs[i])) {
        zend_hash_destroy(redis_sock->subs[i]);
        efree(redis_sock->subs[i]);
        redis_sock->subs[i] = NULL;
    }

    RETVAL_ZVAL(&z_ret, 0, 1);

    return SUCCESS;
}

PHP_REDIS_API zval *
redis_sock_read_multibulk_reply_zval(RedisSock *redis_sock, zval *z_tab)
{
    int numElems;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        ZVAL_NULL(z_tab);
        return NULL;
    }
    array_init_size(z_tab, numElems > 0 ? numElems : 0);
    redis_mbulk_reply_loop(redis_sock, z_tab, numElems, UNSERIALIZE_ALL);

    return z_tab;
}

/**
 * redis_sock_read_bulk_reply
 */
PHP_REDIS_API char *
redis_sock_read_bulk_reply(RedisSock *redis_sock, int bytes)
{
    int offset = 0, nbytes;
    char *reply;
    ssize_t got;

    if (bytes < -1 || bytes > INT_MAX - 2) {
        zend_throw_exception_ex(redis_exception_ce, 0,
            "protocol error, invalid bulk length: %d", bytes);
        return NULL;
    }

    if (-1 == bytes || -1 == redis_check_eof(redis_sock, 1, 0)) {
        return NULL;
    }

    /* + 2 for \r\n */
    nbytes = bytes + 2;

    /* Allocate memory for string */
    reply = emalloc(nbytes);

    /* Consume bulk string */
    while (offset < nbytes) {
        got = redis_sock_read_raw(redis_sock, reply + offset, nbytes - offset);
        if (got <= 0)
            break;

        offset += got;
    }

    /* Protect against reading too few bytes */
    if (offset < nbytes) {
        /* Error or EOF */
        REDIS_THROW_EXCEPTION("socket error on read socket", 0);
        efree(reply);
        return NULL;
    }

    /* Null terminate reply string */
    reply[bytes] = '\0';

    return reply;
}

/**
 * redis_sock_read
 */
PHP_REDIS_API char *
redis_sock_read(RedisSock *redis_sock, int *buf_len)
{
    char inbuf[4096];
    size_t len;

    *buf_len = 0;
    if (redis_sock_gets(redis_sock, inbuf, sizeof(inbuf) - 1, &len) < 0 || len < 1) {
        return NULL;
    }

    switch(inbuf[0]) {
        case '-':
            redis_sock_set_err(redis_sock, inbuf + 1, len - 1);

            /* Filter our ERROR through the few that should actually throw */
            redis_error_throw(redis_sock);

            return NULL;
        case '$':
            {
                char *endptr;
                long long n;

                errno = 0;
                n = strtoll(inbuf + 1, &endptr, 10);
                if (endptr == inbuf + 1 || errno == ERANGE || n < -1 || n > INT_MAX) {
                    return NULL;
                }
                *buf_len = (int)n;
            }
            return redis_sock_read_bulk_reply(redis_sock, *buf_len);

        case '*':
            /* For null multi-bulk replies (like timeouts from brpoplpush): */
            if(len > 2 && memcmp(inbuf + 1, "-1", 2) == 0) {
                return NULL;
            }
            REDIS_FALLTHROUGH;
        case '+':
        case ':':
            /* Single Line Reply */
            /* +OK or :123 */
            if (len > 1) {
                *buf_len = len;
                return estrndup(inbuf, *buf_len);
            }
            REDIS_FALLTHROUGH;
        default:
            zend_throw_exception_ex(redis_exception_ce, 0,
                "protocol error, got '%c' as reply type byte\n",
                inbuf[0]
            );
    }

    return NULL;
}

static zend_string *redis_hash_auth(zend_string *user, zend_string *pass) {
    zend_string *algo, *hex;
    smart_str salted = {0};
    const php_hash_ops *ops;
    unsigned char *digest;
    void *ctx;

    /* No op if there is not username/password */
    if (user == NULL && pass == NULL)
        return NULL;

    /* Theoretically impossible but check anyway */
    algo = zend_string_init("sha256", sizeof("sha256") - 1, 0);
    if ((ops = redis_hash_fetch_ops(algo)) == NULL) {
        zend_string_release(algo);
        return NULL;
    }

    /* Hash username + password with our salt global */
    smart_str_alloc(&salted, 256, 0);
    if (user) smart_str_append_ex(&salted, user, 0);
    if (pass) smart_str_append_ex(&salted, pass, 0);
    smart_str_appendl_ex(&salted, REDIS_G(salt), sizeof(REDIS_G(salt)), 0);

    ctx = emalloc(ops->context_size);
#if PHP_VERSION_ID >= 80100
    ops->hash_init(ctx,NULL);
#else
    ops->hash_init(ctx);
#endif
    ops->hash_update(ctx, (const unsigned char *)ZSTR_VAL(salted.s), ZSTR_LEN(salted.s));

    digest = emalloc(ops->digest_size);
    ops->hash_final(digest, ctx);
    efree(ctx);

    hex = zend_string_safe_alloc(ops->digest_size, 2, 0, 0);
    zend_bin2hex(ZSTR_VAL(hex), digest, ops->digest_size);
    ZSTR_VAL(hex)[2 * ops->digest_size] = 0;

    efree(digest);
    zend_string_release(algo);
    smart_str_free(&salted);

    return hex;
}

static void append_auth_hash(smart_str *dst, zend_string *user, zend_string *pass) {
    zend_string *s;

    if ((s = redis_hash_auth(user, pass)) != NULL) {
        smart_str_appendc(dst, ':');
        smart_str_append_ex(dst, s, 0);
        zend_string_release(s);
    }
}

/* A printf like function to generate our connection pool hash value. */
PHP_REDIS_API zend_string *
redis_pool_spprintf(RedisSock *redis_sock, const char *fmt, ...) {
    smart_str str = {0};

    smart_str_alloc(&str, 128, 0);

    /* We always include phpredis_<host>:<port> */
    smart_str_appendl(&str, "phpredis_", sizeof("phpredis_") - 1);
    smart_str_append_ex(&str, redis_sock->host, 0);
    smart_str_appendc(&str, ':');
    smart_str_append_long(&str, (zend_long)redis_sock->port);

    /* Short circuit if we don't have a pattern */
    if (fmt == NULL) {
        smart_str_0(&str);
        return str.s;
    }

    while (*fmt) {
        switch (*fmt) {
            case 'i':
                if (redis_sock->persistent_id) {
                    smart_str_appendc(&str, ':');
                    smart_str_append_ex(&str, redis_sock->persistent_id, 0);
                }
                break;
            case 'u':
                smart_str_appendc(&str, ':');
                if (redis_sock->user) {
                    smart_str_append_ex(&str, redis_sock->user, 0);
                }
                break;
            case 'p':
                append_auth_hash(&str, NULL, redis_sock->pass);
                break;
            case 'a':
                append_auth_hash(&str, redis_sock->user, redis_sock->pass);
                break;
            default:
                /* Maybe issue a php_error_docref? */
                break;
        }

        fmt++;
    }

    smart_str_0(&str);
    return str.s;
}

PHP_REDIS_API int
redis_bulk_double_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                           zval *z_tab, RedisCmdCtx ctx)
{

    char *response;
    int response_len;
    double ret;

    if ((response = redis_sock_read(redis_sock, &response_len)) == NULL) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    ret = atof(response);
    efree(response);
    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_DOUBLE(ret);
    } else {
        add_next_index_double(z_tab, ret);
    }

    return SUCCESS;
}

PHP_REDIS_API int
redis_type_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                    zval *z_tab, RedisCmdCtx ctx)
{
    char *response;
    int response_len;
    long l;

    if ((response = redis_sock_read(redis_sock, &response_len)) == NULL) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    if (redis_strncmp(response, ZEND_STRL("+string")) == 0) {
        l = REDIS_STRING;
    } else if (redis_strncmp(response, ZEND_STRL("+set")) == 0){
        l = REDIS_SET;
    } else if (redis_strncmp(response, ZEND_STRL("+list")) == 0){
        l = REDIS_LIST;
    } else if (redis_strncmp(response, ZEND_STRL("+zset")) == 0){
        l = REDIS_ZSET;
    } else if (redis_strncmp(response, ZEND_STRL("+hash")) == 0){
        l = REDIS_HASH;
    } else if (redis_strncmp(response, ZEND_STRL("+stream")) == 0) {
        l = REDIS_STREAM;
    } else if (redis_strncmp(response, ZEND_STRL("+vectorset")) == 0) {
        l = REDIS_VECTORSET;
    } else {
        l = REDIS_NOT_FOUND;
    }

    efree(response);
    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_LONG(l);
    } else {
        add_next_index_long(z_tab, l);
    }

    return SUCCESS;
}

PHP_REDIS_API int
redis_config_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      zval *z_tab, RedisCmdCtx ctx)
{
    FailableResultCallback cb = ctx.ptr;

    ZEND_ASSERT(cb == redis_boolean_response ||
                cb == redis_mbulk_reply_zipped_raw);

    return cb(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock, z_tab, ctx);
}

PHP_REDIS_API int
redis_zrange_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      zval *z_tab, RedisCmdCtx ctx)
{
    FailableResultCallback cb;

    /* Whether or not we have WITHSCORES */
    ZEND_ASSERT(ctx.ptr == NULL || ctx.ptr == PHPREDIS_CTX_PTR);

    cb = ctx.ptr ? redis_mbulk_reply_zipped_keys_dbl :
                   redis_sock_read_multibulk_reply;

    return cb(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock, z_tab, ctx);
}

PHP_REDIS_API int
redis_randmember_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                          zval *z_tab, RedisCmdCtx ctx)
{
    FailableResultCallback cb;

    /* Whether or not we have a COUNT argument */
    ZEND_ASSERT(ctx.ptr == NULL || ctx.ptr == PHPREDIS_CTX_PTR);

    cb = ctx.ptr ? redis_sock_read_multibulk_reply : redis_string_response;

    return cb(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock, z_tab, ctx);
}

PHP_REDIS_API int
redis_info_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                    zval *z_tab, RedisCmdCtx ctx)
{
    char *response;
    int response_len;
    zval z_ret;

    /* Read bulk response */
    if ((response = redis_sock_read(redis_sock, &response_len)) == NULL) {
        RETVAL_FALSE;
        return FAILURE;
    }

    /* Parse it into a zval array */
    ZVAL_UNDEF(&z_ret);
    redis_parse_info_response(response, &z_ret);

    /* Free source response */
    efree(response);

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

/* Count occurrences of a byte in a NUL-terminated string, used to size a
 * result array before tokenizing the string into it. */
static uint32_t redis_count_char(const char *s, char c)
{
    uint32_t n = 0;
    for (; (s = strchr(s, c)) != NULL; s++) n++;
    return n;
}

PHP_REDIS_API void
redis_parse_info_response(char *response, zval *z_ret)
{
    char *p1, *s1 = NULL;
    uint32_t nlines = redis_count_char(response, '\n');

    ZVAL_FALSE(z_ret);
    if ((p1 = php_strtok_r(response, _NL, &s1)) != NULL) {
        array_init_size(z_ret, nlines + 1);
        do {
            if (*p1 == '#') continue;
            char *p;
            zend_uchar type;
            zend_long lval;
            double dval;
            if ((p = strchr(p1, ':')) != NULL) {
                type = is_numeric_string(p + 1, strlen(p + 1), &lval, &dval, 0);
                switch (type) {
                case IS_LONG:
                    add_assoc_long_ex(z_ret, p1, p - p1, lval);
                    break;
                case IS_DOUBLE:
                    if (dval < HUGE_VAL) {
                        add_assoc_double_ex(z_ret, p1, p - p1, dval);
                        break;
                    }
                    // fall through
                default:
                    add_assoc_string_ex(z_ret, p1, p - p1, p + 1);
                }
            } else {
                add_next_index_string(z_ret, p1);
            }
        } while ((p1 = php_strtok_r(NULL, _NL, &s1)) != NULL);
    }
}

static void
redis_parse_client_info(char *info, zval *z_ret)
{
    char *p1, *s1 = NULL;
    uint32_t nfields = redis_count_char(info, ' ');

    ZVAL_FALSE(z_ret);
    if ((p1 = php_strtok_r(info, " ", &s1)) != NULL) {
        array_init_size(z_ret, nfields + 1);
        do {
            char *p;
            zend_uchar type;
            zend_long lval;
            double dval;
            if ((p = strchr(p1, '=')) != NULL) {
                type = is_numeric_string(p + 1, strlen(p + 1), &lval, &dval, 0);
                switch (type) {
                case IS_LONG:
                    add_assoc_long_ex(z_ret, p1, p - p1, lval);
                    break;
                case IS_DOUBLE:
                    add_assoc_double_ex(z_ret, p1, p - p1, dval);
                    break;
                default:
                    add_assoc_string_ex(z_ret, p1, p - p1, p + 1);
                }
            } else {
                add_next_index_string(z_ret, p1);
            }
        } while ((p1 = php_strtok_r(NULL, " ", &s1)) != NULL);
    }
}

int
redis_client_info_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                        zval *z_tab, RedisCmdCtx ctx)
{
    char *resp;
    int resp_len;
    zval z_ret;

    /* Make sure we can read the bulk response from Redis */
    if ((resp = redis_sock_read(redis_sock, &resp_len)) == NULL) {
        RETVAL_FALSE;
        return FAILURE;
    }

    /* Parse it out */
    redis_parse_client_info(resp, &z_ret);

    /* Free our response */
    efree(resp);

    /* Return or append depending if we're atomic */
    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

/*
 * Specialized handling of the CLIENT LIST output so it comes out in a simple way for PHP userland code
 * to handle.
 */
PHP_REDIS_API int
redis_client_list_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                        zval *z_tab, RedisCmdCtx ctx)
{
    char *resp;
    int resp_len;
    zval z_ret;

    /* Make sure we can read the bulk response from Redis */
    if ((resp = redis_sock_read(redis_sock, &resp_len)) == NULL) {
        RETVAL_FALSE;
        return FAILURE;
    } else if (resp_len > 0) {
        /* Parse it out */
        redis_parse_client_list_response(resp, &z_ret);
    } else {
        ZVAL_EMPTY_ARRAY(&z_ret);
    }

    /* Free our response */
    efree(resp);

    /* Return or append depending if we're atomic */
    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

PHP_REDIS_API void
redis_parse_client_list_response(char *response, zval *z_ret)
{
    char *p, *s = NULL;
    uint32_t nlines = redis_count_char(response, '\n');

    ZVAL_FALSE(z_ret);
    if ((p = php_strtok_r(response, _NL, &s)) != NULL) {
        array_init_size(z_ret, nlines + 1);
        do {
            zval z_sub;
            redis_parse_client_info(p, &z_sub);
            add_next_index_zval(z_ret, &z_sub);
        } while ((p = php_strtok_r(NULL, _NL, &s)) != NULL);
    }
}

PHP_REDIS_API int
redis_zadd_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                    zval *z_tab, RedisCmdCtx ctx)
{
    FailableResultCallback cb;

    ZEND_ASSERT(ctx.ptr == NULL || ctx.ptr == PHPREDIS_CTX_PTR);

    cb = ctx.ptr ? redis_bulk_double_response : redis_long_response;

    return cb(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock, z_tab, redis_empty_ctx);
}

PHP_REDIS_API int
redis_zrandmember_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                           zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_mbulk_reply_raw(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        return redis_mbulk_reply_zipped_keys_dbl(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                                 redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

PHP_REDIS_API int
redis_zdiff_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                     zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_mbulk_reply_raw(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                     z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_mbulk_reply_zipped_keys_dbl(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                                 redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

PHP_REDIS_API int
redis_set_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                   zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_boolean_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                      redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

PHP_REDIS_API int
redis_hrandfield_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                          zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_mbulk_reply_raw(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        return redis_mbulk_reply_zipped_raw(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                            redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

PHP_REDIS_API int
redis_pop_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                   zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_sock_read_multibulk_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                               redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

PHP_REDIS_API int
redis_object_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      zval *z_tab, RedisCmdCtx ctx)
{
    ZEND_ASSERT(ctx.ptr == PHPREDIS_CTX_PTR || ctx.ptr == PHPREDIS_CTX_PTR + 1);

    if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_long_response(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                   z_tab, redis_empty_ctx);
    } else {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    }
}

PHP_REDIS_API int
redis_read_lpos_response(zval *zdst, RedisSock *redis_sock, char reply_type,
                         long long elements, RedisCmdCtx ctx)
{
    char inbuf[4096];
    size_t len;
    int i;

    if (ctx.ptr == NULL) {
        if (reply_type != TYPE_INT && reply_type != TYPE_BULK)
            return FAILURE;

        if (elements > -1) {
            ZVAL_LONG(zdst, elements);
        } else {
            REDIS_ZVAL_NULL(redis_sock, zdst);
        }
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        if (reply_type != TYPE_MULTIBULK)
            return FAILURE;

        array_init_size(zdst, elements > 0 ? elements : 0);

        for (i = 0;  i < elements; ++i) {
            if (redis_sock_gets(redis_sock, inbuf, sizeof(inbuf), &len) < 0) {
                zval_ptr_dtor_nogc(zdst);
                return FAILURE;
            }
            add_next_index_long(zdst, atol(inbuf + 1));
        }
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }

    return SUCCESS;
}


PHP_REDIS_API int
redis_lpos_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                    zval *z_tab, RedisCmdCtx ctx)
{
    char inbuf[1024] = {0};
    int res = SUCCESS;
    zval zdst = {0};
    size_t len;

    /* Attempt to read the LPOS response */
    if (redis_sock_gets(redis_sock, inbuf, sizeof(inbuf), &len) < 0 ||
        redis_read_lpos_response(&zdst, redis_sock, *inbuf, atoll(inbuf+1), ctx) < 0)
    {
        ZVAL_FALSE(&zdst);
        res = FAILURE;
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, zdst);

    return res;
}

PHP_REDIS_API int
redis_select_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      zval *z_tab, RedisCmdCtx ctx)
{
    if (redis_boolean_response(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                               z_tab, redis_empty_ctx) < 0)
        return FAILURE;

    redis_sock->dbNumber = (long)(uintptr_t)ctx.ptr;
    return SUCCESS;
}

PHP_REDIS_API int
redis_boolean_response_impl(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                            zval *z_tab, RedisCmdCtx ctx,
                            SuccessCallback success_callback)
{

    char *response;
    int response_len;
    zend_bool ret = 0;

    if ((response = redis_sock_read(redis_sock, &response_len)) != NULL) {
        ret = (*response == '+');
        efree(response);
    }

    if (ret && success_callback != NULL) {
        success_callback(redis_sock);
    }
    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_BOOL(ret);
    } else {
        add_next_index_bool(z_tab, ret);
    }

    return ret ? SUCCESS : FAILURE;
}

PHP_REDIS_API int redis_boolean_response(INTERNAL_FUNCTION_PARAMETERS,
                                         RedisSock *redis_sock, zval *z_tab,
                                         RedisCmdCtx ctx)
{
    return redis_boolean_response_impl(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                       z_tab, redis_empty_ctx, NULL);
}

PHP_REDIS_API int redis_long_response(INTERNAL_FUNCTION_PARAMETERS,
                                      RedisSock *redis_sock, zval * z_tab,
                                      RedisCmdCtx ctx)
{

    char *response;
    int response_len;

    if ((response = redis_sock_read(redis_sock, &response_len)) == NULL ||
        *response != TYPE_INT)
    {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        if (response) efree(response);
        return FAILURE;
    }

    int64_t ret = phpredis_atoi64(response + 1);

    if (redis_sock_is_atomic(redis_sock)) {
        if (ret > LONG_MAX) { /* overflow */
            RETVAL_STRINGL(response + 1, response_len - 1);
        } else {
            RETVAL_LONG((long)ret);
        }
    } else {
        if (ret > LONG_MAX) { /* overflow */
            add_next_index_stringl(z_tab, response + 1, response_len - 1);
        } else {
            add_next_index_long(z_tab, (long)ret);
        }
    }

    efree(response);
    return SUCCESS;
}

/* Helper method to convert [key, value, key, value] into [key => value,
 * key => value] when returning data to the caller.  Depending on our decode
 * flag we'll convert the value data types */
static void array_zip_values_and_scores(RedisSock *redis_sock, zval *z_tab,
                                        int decode)
{

    zval z_ret, zele, *zv, *zkey = NULL;
    zend_string *key, *aux, *val;
    HashTable *ht;

    ht = Z_ARRVAL_P(z_tab);

    array_init_size(&z_ret, zend_hash_num_elements(ht) / 2);

    ZEND_HASH_FOREACH_VAL(ht, zv) {
        if (zkey == NULL) {
            zkey = zv;
            continue;
        }

        if (decode == SCORE_DECODE_INT || decode == SCORE_DECODE_DOUBLE) {
            val = zval_get_tmp_string(zv, &aux);

            if (decode == SCORE_DECODE_INT && ZSTR_LEN(val) > 0) {
                ZVAL_LONG(&zele, atoi(ZSTR_VAL(val)+1));
            } else if (decode == SCORE_DECODE_DOUBLE) {
                ZVAL_DOUBLE(&zele, atof(ZSTR_VAL(val)));
            } else {
                ZVAL_ZVAL(&zele, zv, 1, 0);
            }

            zend_tmp_string_release(aux);
        } else {
            ZVAL_ZVAL(&zele, zv, 1, 0);
        }

        key = zval_get_tmp_string(zkey, &aux);
        zend_symtable_update(Z_ARRVAL_P(&z_ret), key, &zele);
        zend_tmp_string_release(aux);

        zkey = NULL;
    } ZEND_HASH_FOREACH_END();

    /* Replace */
    zval_ptr_dtor_nogc(z_tab);
    ZVAL_ZVAL(z_tab, &z_ret, 0, 0);
}

static int array_zip_values_recursive(zval *z_tab) {
    zend_string *zkey = NULL;
    zval z_ret, z_sub, *zv;

    array_init_size(&z_ret, zend_hash_num_elements(Z_ARRVAL_P(z_tab)));
    ZEND_HASH_FOREACH_VAL(Z_ARRVAL_P(z_tab), zv) {
        if (zkey != NULL) {
            if (Z_TYPE_P(zv) == IS_ARRAY && array_zip_values_recursive(zv) != SUCCESS) {
                zend_string_release(zkey);
                zval_ptr_dtor_nogc(&z_ret);
                return FAILURE;
            }
            ZVAL_ZVAL(&z_sub, zv, 1, 0);
            add_assoc_zval_ex(&z_ret, ZSTR_VAL(zkey), ZSTR_LEN(zkey), &z_sub);
            zend_string_release(zkey);
            zkey = NULL;
        } else if (Z_TYPE_P(zv) == IS_STRING) {
            zkey = zval_get_string(zv);
        } else {
            if (Z_TYPE_P(zv) == IS_ARRAY && array_zip_values_recursive(zv) != SUCCESS) {
                zval_ptr_dtor_nogc(&z_ret);
                return FAILURE;
            }
            ZVAL_ZVAL(&z_sub, zv, 1, 0);
            add_next_index_zval(&z_ret, &z_sub);
        }
    } ZEND_HASH_FOREACH_END();

    if (zkey != NULL) {
        zend_string_release(zkey);
        zval_ptr_dtor_nogc(&z_ret);
        return FAILURE;
    }

    zval_ptr_dtor_nogc(z_tab);
    ZVAL_ZVAL(z_tab, &z_ret, 0, 0);

    return SUCCESS;
}

static int
redis_mbulk_reply_zipped(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                         zval *z_tab, int unserialize, int decode)
{
    int numElems;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }
    zval z_multi_result;

    if (numElems < 1) {
        ZVAL_EMPTY_ARRAY(&z_multi_result);
    } else {
        array_init_size(&z_multi_result, numElems); /* pre-allocate array for multi's results. */

        /* Grab our key, value, key, value array */
        redis_mbulk_reply_loop(redis_sock, &z_multi_result, numElems, unserialize);

        /* Zip keys and values */
        array_zip_values_and_scores(redis_sock, &z_multi_result, decode);
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_multi_result);

    return 0;
}

static int
geosearch_cast(zval *zv)
{
    if (Z_TYPE_P(zv) == IS_ARRAY) {
        zend_hash_apply(Z_ARRVAL_P(zv), geosearch_cast);
    } else if (Z_TYPE_P(zv) == IS_STRING) {
        convert_to_double(zv);
    }
    return SUCCESS;
}

PHP_REDIS_API int
redis_read_mpop_response(RedisSock *redis_sock, zval *zdst, int elements,
                         RedisCmdCtx ctx)
{
    int subele, keylen;
    zval zele = {0};
    char *key;

    ZEND_ASSERT(ctx.ptr == NULL || ctx.ptr == PHPREDIS_CTX_PTR);

    if (elements < 0) {
        REDIS_ZVAL_NULL(redis_sock, zdst);
        return SUCCESS;
    }

    if (UNEXPECTED(elements != 2)) {
        php_error_docref(NULL, E_WARNING, "Expected 2 elements in response, got %d", elements);
        return FAILURE;
    }

    array_init_size(zdst, 2);

    /* Key name and number of entries */
    if ((key = redis_sock_read(redis_sock, &keylen)) == NULL ||
        read_mbulk_header(redis_sock, &elements) < 0 || elements < 0)
    {
        if (key) efree(key);
        goto fail;
    }

    add_next_index_stringl(zdst, key, keylen);
    efree(key);

    array_init_size(&zele, elements);

    if (ctx.ptr == PHPREDIS_CTX_PTR) {
        int i;
        for (i = 0; i < elements; i++) {
            if (read_mbulk_header(redis_sock, &subele) < 0 || subele != 2) {
                zval_ptr_dtor_nogc(&zele);
                goto fail;
            }
            redis_mbulk_reply_loop(redis_sock, &zele, subele, UNSERIALIZE_KEYS);
        }

        array_zip_values_and_scores(redis_sock, &zele, SCORE_DECODE_DOUBLE);
    } else {
        redis_mbulk_reply_loop(redis_sock, &zele, elements, UNSERIALIZE_ALL);
    }

    add_next_index_zval(zdst, &zele);

    return SUCCESS;

fail:
    zval_ptr_dtor_nogc(zdst);
    ZVAL_FALSE(zdst);

    return FAILURE;
}

PHP_REDIS_API int
redis_mpop_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                    zval *z_tab, RedisCmdCtx ctx)
{
    int elements, res = SUCCESS;
    zval zret = {0};

    if (read_mbulk_header(redis_sock, &elements) == FAILURE ||
        redis_read_mpop_response(redis_sock, &zret, elements, ctx) == FAILURE)
    {
        res = FAILURE;
        ZVAL_FALSE(&zret);
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, zret);

    return res;
}

PHP_REDIS_API int
redis_read_geosearch_response(zval *zdst, RedisSock *redis_sock,
                              long long elements, int with_aux_data)
{
    zval z_multi_result, z_sub, *z_ele, *zv;
    zend_string *zkey;

    /* Handle the trivial "empty" result first */
    if (elements < 0 && redis_sock->null_mbulk_as_null) {
        ZVAL_NULL(zdst);
        return SUCCESS;
    }

    array_init_size(zdst, elements > 0 ? elements : 0);

    if (with_aux_data == 0) {
        redis_mbulk_reply_loop(redis_sock, zdst, elements, UNSERIALIZE_NONE);
    } else {
        array_init_size(&z_multi_result, elements > 0 ? elements : 0);

        redis_read_multibulk_recursive(redis_sock, elements, 0, &z_multi_result);

        ZEND_HASH_FOREACH_VAL(Z_ARRVAL(z_multi_result), z_ele) {
            // The first item in the sub-array is always the name of the returned item
            zv = zend_hash_index_find(Z_ARRVAL_P(z_ele), 0);
            zkey = zval_get_string(zv);

            zend_hash_index_del(Z_ARRVAL_P(z_ele), 0);

            // The other information is returned in the following order as successive
            // elements of the sub-array: distance, geohash, coordinates
            zend_hash_apply(Z_ARRVAL_P(z_ele), geosearch_cast);

            // Reindex elements so they start at zero */
            ZVAL_ARR(&z_sub, zend_array_to_list(Z_ARRVAL_P(z_ele)));

            add_assoc_zval_ex(zdst, ZSTR_VAL(zkey), ZSTR_LEN(zkey), &z_sub);
            zend_string_release(zkey);
        } ZEND_HASH_FOREACH_END();

        // Cleanup
        zval_ptr_dtor_nogc(&z_multi_result);
    }

    return SUCCESS;
}

PHP_REDIS_API int
redis_geosearch_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                         zval *z_tab, RedisCmdCtx ctx)
{
    zval zret = {0};
    int elements;

    if (read_mbulk_header(redis_sock, &elements) < 0 ||
        redis_read_geosearch_response(&zret, redis_sock, elements,
                                      ctx.ptr != NULL) < 0)
    {
        ZVAL_FALSE(&zret);
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, zret);

    return SUCCESS;
}

int redis_client_trackinginfo_reply(INTERNAL_FUNCTION_PARAMETERS,
                                    RedisSock *redis_sock, zval *z_tab,
                                    RedisCmdCtx ctx)
{
    int numElems;
    zval z_ret;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    array_init_size(&z_ret, numElems > 0 ? numElems : 0);
    redis_read_multibulk_recursive(redis_sock, numElems, 0, &z_ret);
    array_zip_values_and_scores(redis_sock, &z_ret, 0);

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

PHP_REDIS_API int
redis_client_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_client_info_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                       redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_client_list_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                       redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        return redis_boolean_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                      redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 2) {
        return redis_long_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                   redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 3) {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 4) {
        return redis_client_trackinginfo_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                               redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

static int
redis_hello_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                     zval *z_tab, RedisCmdCtx ctx)
{
    zval z_ret, *zv;
    int numElems;

    if (read_mbulk_header(redis_sock, &numElems) < 0)
        goto fail;

    array_init_size(&z_ret, numElems > 0 ? numElems : 0);

    if (redis_read_multibulk_recursive(redis_sock, numElems, 0, &z_ret) != SUCCESS ||
        array_zip_values_recursive(&z_ret) != SUCCESS)
    {
        zval_ptr_dtor_nogc(&z_ret);
        goto fail;
    }

    if (redis_sock->hello.server) {
        zend_string_release(redis_sock->hello.server);
    }

    if ((zv = zend_hash_str_find(Z_ARRVAL(z_ret), ZEND_STRL("dragonfly_version")))) {
        redis_sock->hello.server = zend_string_init(ZEND_STRL("dragonfly"), 0);
    } else {
        zv = zend_hash_str_find(Z_ARRVAL(z_ret), ZEND_STRL("server"));
        redis_sock->hello.server = zv ? zval_get_string(zv) : ZSTR_EMPTY_ALLOC();
    }

    if (redis_sock->hello.version) {
        zend_string_release(redis_sock->hello.version);
    }
    zv = zend_hash_str_find(Z_ARRVAL(z_ret), ZEND_STRL("version"));
    redis_sock->hello.version = zv ? zval_get_string(zv) : ZSTR_EMPTY_ALLOC();

    zval_ptr_dtor_nogc(&z_ret);

    if (ctx.ptr == PHPREDIS_CTX_PTR) {
        ZVAL_STR_COPY(&z_ret, redis_sock->hello.server);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        ZVAL_STR_COPY(&z_ret, redis_sock->hello.version);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }

    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_ZVAL(&z_ret, 0, 1);
    } else {
        add_next_index_zval(z_tab, &z_ret);
    }

    return SUCCESS;

fail:
    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_FALSE;
    } else {
        add_next_index_bool(z_tab, 0);
    }
    return FAILURE;
}


PHP_REDIS_API int
redis_hello_server_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                            zval *z_tab, RedisCmdCtx ctx)
{
    static RedisCmdCtx hello_ctx = { .ptr = PHPREDIS_CTX_PTR };

    return redis_hello_response(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                z_tab, hello_ctx);
}

PHP_REDIS_API int
redis_hello_version_response(INTERNAL_FUNCTION_PARAMETERS,
                             RedisSock *redis_sock, zval *z_tab,
                             RedisCmdCtx ctx)
{
    static RedisCmdCtx hello_ctx = { .ptr = PHPREDIS_CTX_PTR + 1 };

    return redis_hello_response(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                z_tab, hello_ctx);
}

int
redis_function_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                     zval *z_tab, RedisCmdCtx ctx)
{
    int numElems;
    zval z_ret;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    array_init_size(&z_ret, numElems > 0 ? numElems : 0);
    redis_read_multibulk_recursive(redis_sock, numElems, 0, &z_ret);
    array_zip_values_recursive(&z_ret);

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

PHP_REDIS_API int
redis_function_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                        zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_boolean_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                      redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        return redis_function_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                    redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

int
redis_command_info_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                         zval *z_tab, RedisCmdCtx ctx)
{
    int numElems;
    zval z_ret;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    array_init_size(&z_ret, numElems > 0 ? numElems : 0);
    redis_read_multibulk_recursive(redis_sock, numElems, 0, &z_ret);
    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

PHP_REDIS_API int
redis_command_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                       zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_command_info_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                        redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_long_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                   redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        return redis_mbulk_reply_raw(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 2) {
        return redis_read_variant_reply_strings(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                                redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

/* Helper function to consume Redis stream message data.  This is useful for
 * multiple stream callers (e.g. XREAD[GROUP], and X[REV]RANGE handlers). */
PHP_REDIS_API int
redis_read_stream_messages(RedisSock *redis_sock, int count, zval *z_ret) {
    zval z_message;
    int i, mhdr, fields;
    char *id = NULL;
    int idlen;

    /* Iterate over each message */
    for (i = 0; i < count; i++) {
        id = NULL;

        /* Consume inner multi-bulk header, message ID itself and finally
         * the multi-bulk header for field and values */
        if ((read_mbulk_header(redis_sock, &mhdr) < 0 || mhdr != 2) ||
            ((id = redis_sock_read(redis_sock, &idlen)) == NULL) ||
            (read_mbulk_header(redis_sock, &fields) < 0 ||
            (fields > 0 && fields % 2 != 0)))
        {
            if (id) efree(id);
            return -1;
        }

        if (fields < 0) {
            add_assoc_null_ex(z_ret, id, idlen);
        } else {
            array_init_size(&z_message, fields > 0 ? fields : 0);
            if (redis_mbulk_reply_loop(redis_sock, &z_message, fields, UNSERIALIZE_VALS) < 0) {
                zval_ptr_dtor_nogc(&z_message);
                efree(id);
                return -1;
            }
            array_zip_values_and_scores(redis_sock, &z_message, SCORE_DECODE_NONE);
            add_assoc_zval_ex(z_ret, id, idlen, &z_message);
        }
        efree(id);
    }

    return 0;
}

PHP_REDIS_API int
redis_xrange_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                   zval *z_tab, RedisCmdCtx ctx)
{
    zval z_messages;
    int messages;

    if (read_mbulk_header(redis_sock, &messages) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return -1;
    }

    array_init_size(&z_messages, messages > 0 ? messages : 0);

    if (redis_read_stream_messages(redis_sock, messages, &z_messages) < 0)
    {
        zval_ptr_dtor_nogc(&z_messages);
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return -1;
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_messages);

    return 0;
}

PHP_REDIS_API int
redis_read_stream_messages_multi(RedisSock *redis_sock, int count,
                                 zval *z_streams)
{
    zval z_messages;
    int i, shdr, messages;
    char *id = NULL;
    int idlen;

    for (i = 0; i < count; i++) {
        id = NULL;

        if ((read_mbulk_header(redis_sock, &shdr) < 0 || shdr != 2) ||
            (id = redis_sock_read(redis_sock, &idlen)) == NULL ||
            read_mbulk_header(redis_sock, &messages) < 0)
        {
            if (id) efree(id);
            return -1;
        }

        array_init_size(&z_messages, messages > 0 ? messages : 0);

        if (redis_read_stream_messages(redis_sock, messages, &z_messages) < 0)
            goto failure;

        add_assoc_zval_ex(z_streams, id, idlen, &z_messages);
        efree(id);
    }

    return 0;
failure:
    efree(id);
    zval_ptr_dtor_nogc(&z_messages);
    return -1;
}

PHP_REDIS_API int
redis_xread_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                  zval *z_tab, RedisCmdCtx ctx)
{
    zval z_rv;
    int streams;

    if (read_mbulk_header(redis_sock, &streams) < 0)
        goto failure;

    if (streams == -1 && redis_sock->null_mbulk_as_null) {
        ZVAL_NULL(&z_rv);
    } else {
        array_init_size(&z_rv, streams > 0 ? streams : 0);
        if (redis_read_stream_messages_multi(redis_sock, streams, &z_rv) < 0)
            goto cleanup;
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_rv);
    return 0;

cleanup:
    zval_ptr_dtor_nogc(&z_rv);
failure:
    REDIS_RESPONSE_ERROR(redis_sock, z_tab);
    return -1;
}

/* A helper method to read X[AUTO]CLAIM messages into an array.  */
static int
redis_read_xclaim_ids(RedisSock *redis_sock, int count, zval *rv) {
    zval z_msg;
    REDIS_REPLY_TYPE type;
    char *id = NULL;
    int i, fields, idlen;
    long li;

    for (i = 0; i < count; i++) {
        id = NULL;

        /* Consume inner reply type */
        if (redis_read_reply_type(redis_sock, &type, &li) < 0 ||
            (type != TYPE_BULK && type != TYPE_MULTIBULK) ||
            (type == TYPE_BULK && li <= 0)) return -1;

        /* TYPE_BULK is the JUSTID variant, otherwise it's standard xclaim response */
        if (type == TYPE_BULK) {
            if ((id = redis_sock_read_bulk_reply(redis_sock, (size_t)li)) == NULL)
                return -1;

            add_next_index_stringl(rv, id, li);
            efree(id);
        } else {
            if ((li != 2 || (id = redis_sock_read(redis_sock, &idlen)) == NULL) ||
                (read_mbulk_header(redis_sock, &fields) < 0 || fields % 2 != 0))
            {
                if (id) efree(id);
                return -1;
            }

            array_init_size(&z_msg, fields > 0 ? fields : 0);

            if (redis_mbulk_reply_loop(redis_sock, &z_msg, fields, UNSERIALIZE_VALS) < 0) {
                zval_ptr_dtor_nogc(&z_msg);
                efree(id);
                return -1;
            }
            array_zip_values_and_scores(redis_sock, &z_msg, SCORE_DECODE_NONE);
            add_assoc_zval_ex(rv, id, idlen, &z_msg);
            efree(id);
        }
    }

    return 0;
}

/* Read an X[AUTO]CLAIM reply having already consumed the reply-type byte. */
PHP_REDIS_API int
redis_read_xclaim_reply(RedisSock *redis_sock, int count, int is_xautoclaim,
                        zval *rv)
{
    REDIS_REPLY_TYPE type;
    zval z_msgs = {0};
    char *id = NULL;
    long id_len = 0;
    int messages = 0;

    if (is_xautoclaim && (count != 2 && count != 3)) {
        return -1;
    }

    ZVAL_UNDEF(rv);

    /* If this is XAUTOCLAIM consume the BULK ID and then the actual number of IDs.
     * Otherwise, our 'count' argument is the number of IDs. */
    if (is_xautoclaim) {
        if (redis_read_reply_type(redis_sock, &type, &id_len) < 0 || type != TYPE_BULK)
            goto failure;
        if ((id = redis_sock_read_bulk_reply(redis_sock, id_len)) == NULL)
            goto failure;
        if (read_mbulk_header(redis_sock, &messages) < 0)
            goto failure;
    } else {
        messages = count;
    }

    array_init_size(&z_msgs, messages > 0 ? messages : 0);

    if (redis_read_xclaim_ids(redis_sock, messages, &z_msgs) < 0)
        goto failure;

    /* If XAUTOCLAIM we now need to consume the final array of message IDs */
    if (is_xautoclaim) {
        zval z_deleted = {0};

        if (count == 3 && redis_sock_read_multibulk_reply_zval(redis_sock, &z_deleted) == NULL)
            goto failure;

        array_init_size(rv, count);

        // Package up ID and message
        add_next_index_stringl(rv, id, id_len);
        add_next_index_zval(rv, &z_msgs);

        // Add deleted messages if they exist
        if (count == 3)
            add_next_index_zval(rv, &z_deleted);

        efree(id);
    } else {
        // We just want the messages
        ZVAL_COPY_VALUE(rv, &z_msgs);
    }

    return 0;

failure:
    zval_ptr_dtor_nogc(&z_msgs);
    zval_ptr_dtor_nogc(rv);
    if (id) efree(id);

    return -1;
}

PHP_REDIS_API int
redis_xclaim_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                   zval *z_tab, RedisCmdCtx ctx)
{
    zval z_ret = {0};
    int count;

    ZEND_ASSERT(ctx.ptr == NULL || ctx.ptr == PHPREDIS_CTX_PTR);

    if (read_mbulk_header(redis_sock, &count) < 0)
        goto failure;

    if (redis_read_xclaim_reply(redis_sock, count, ctx.ptr == PHPREDIS_CTX_PTR, &z_ret) < 0)
        goto failure;

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return 0;

failure:
    REDIS_RESPONSE_ERROR(redis_sock, z_tab);
    return -1;
}

PHP_REDIS_API int
redis_read_xinfo_response(RedisSock *redis_sock, zval *z_ret, int elements)
{
    zval zv;
    int i, len = 0;
    char *key = NULL, *data;
    REDIS_REPLY_TYPE type;
    long li;

    for (i = 0; i < elements; ++i) {
        if (redis_read_reply_type(redis_sock, &type, &li) < 0) {
            goto failure;
        }
        switch (type) {
        case TYPE_BULK:
            if ((data = redis_sock_read_bulk_reply(redis_sock, li)) == NULL) {
                if (!key) goto failure;
                add_assoc_null_ex(z_ret, key, len);
                efree(key);
                key = NULL;
            } else if (key) {
                add_assoc_stringl_ex(z_ret, key, len, data, li);
                efree(data);
                efree(key);
                key = NULL;
            } else {
                key = data;
                len = li;
            }
            break;
        case TYPE_INT:
            if (key) {
                add_assoc_long_ex(z_ret, key, len, li);
                efree(key);
                key = NULL;
            } else {
                len = spprintf(&key, 0, "%ld", li);
            }
            break;
        case TYPE_MULTIBULK:
            array_init_size(&zv, li > 0 ? li : 0);
            if (redis_read_xinfo_response(redis_sock, &zv, li) != SUCCESS) {
                zval_ptr_dtor_nogc(&zv);
                goto failure;
            }
            if (key) {
                add_assoc_zval_ex(z_ret, key, len, &zv);
                efree(key);
                key = NULL;
            } else {
                add_next_index_zval(z_ret, &zv);
            }
            break;
        default:
            goto failure;
        }
    }

    return SUCCESS;

failure:
    if (key) efree(key);
    return FAILURE;
}

PHP_REDIS_API int
redis_read_vinfo_response(RedisSock *redis_sock, zval *z_ret, long long count) {
    char kbuf[256], vbuf[256];
    REDIS_REPLY_TYPE type;
    size_t klen, vlen;
    long lval;

    if (count < 0 || count % 2 != 0 || Z_TYPE_P(z_ret) != IS_ARRAY) {
        zend_error_noreturn(E_ERROR, "Internal vinfo handler error");
    }

    for (long long i = 0; i < count; i += 2) {
        if (redis_read_reply_type(redis_sock, &type, &lval) < 0 ||
            type != TYPE_LINE ||
            redis_sock_gets(redis_sock, kbuf, sizeof(kbuf), &klen) < 0)
        {
            return FAILURE;
        }

        if (redis_read_reply_type(redis_sock, &type, &lval) < 0) {
            return FAILURE;
        }

        switch (type) {
            case TYPE_LINE:
                if (redis_sock_gets(redis_sock, vbuf, sizeof(vbuf), &vlen) < 0) {
                    return FAILURE;
                }
                add_assoc_stringl_ex(z_ret, kbuf, klen, vbuf, vlen);
                break;
            case TYPE_INT:
                add_assoc_long_ex(z_ret, kbuf, klen, lval);
                break;
            default:
                return FAILURE;
       }
    }

    return SUCCESS;
}


PHP_REDIS_API int
redis_vinfo_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                  zval *z_tab, RedisCmdCtx ctx)
{
    zval z_ret;
    int count;

    if (read_mbulk_header(redis_sock, &count) < 0 ||
        count < 0 || count % 2 != 0)
    {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    array_init_size(&z_ret, count / 2);

    if (redis_read_vinfo_response(redis_sock, &z_ret, count) != SUCCESS) {
        zval_ptr_dtor_nogc(&z_ret);
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

/* Unfortunately VEMB decided to use +string\r\n for the encoding when RAW is
 * sent which PhpRedis will parse as `(true)` so we need a specific handler for
 * it */
PHP_REDIS_API int
redis_read_vemb_response(RedisSock *redis_sock, zval *z_ret, long long count) {
    REDIS_REPLY_TYPE type;
    char kbuf[256], *str;
    size_t klen;
    double dval;
    long tlen;

    if (count < 0 || Z_TYPE_P(z_ret) != IS_ARRAY) {
        zend_error_noreturn(E_ERROR, "Internal vemb handler error");
    }

    for (long long i = 0; i < count; i++) {
        if (redis_read_reply_type(redis_sock, &type, &tlen) < 0) {
            return FAILURE;
        }

        if (type == TYPE_LINE) {
            if (redis_sock_gets(redis_sock, kbuf, sizeof(kbuf), &klen) < 0)
                return FAILURE;
            add_next_index_stringl(z_ret, kbuf, klen);
        } else if (type == TYPE_BULK) {
            if ((str = redis_sock_read_bulk_reply(redis_sock, tlen)) == NULL)
                return FAILURE;

            if (is_numeric_string(str, tlen, NULL, &dval, 0) == IS_DOUBLE) {
                add_next_index_double(z_ret, dval);
            } else {
                add_next_index_stringl(z_ret, str, tlen);
            }

            efree(str);
        } else {
            return FAILURE;
        }
    }

    return SUCCESS;
}

PHP_REDIS_API int
redis_vemb_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                 zval *z_tab, RedisCmdCtx ctx)
{
    zval z_ret;
    int count;

    if (read_mbulk_header(redis_sock, &count) < 0 || count < 1) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    array_init_size(&z_ret, count);

    if (redis_read_vemb_response(redis_sock, &z_ret, count) != SUCCESS) {
        zval_ptr_dtor_nogc(&z_ret);
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

PHP_REDIS_API int
redis_read_vlinks_response(RedisSock *redis_sock, zval *z_ret,
                           long long elements, RedisCmdCtx ctx)
{
    long long i;
    zval z_ele;
    int links;

    array_init_size(z_ret, elements);

    for (i = 0; i < elements; i++) {
        if (read_mbulk_header(redis_sock, &links) < 0)
            return FAILURE;

        array_init_size(&z_ele, links > 0 ? links : 0);
        redis_mbulk_reply_loop(redis_sock, &z_ele, links, UNSERIALIZE_KEYS);

        if (ctx.ptr == PHPREDIS_CTX_PTR) {
            array_zip_values_and_scores(redis_sock, &z_ele, SCORE_DECODE_DOUBLE);
        }

        add_next_index_zval(z_ret, &z_ele);
    }

    return SUCCESS;
}

PHP_REDIS_API int
redis_vlinks_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                   zval *z_tab, RedisCmdCtx ctx)
{
    int elements;
    zval z_ret;

    ZVAL_FALSE(&z_ret);

    if (read_mbulk_header(redis_sock, &elements) < 0)
        goto fail;

    /* TODO: Figure out if we want `false` or `[]` here */
    if (elements < 0) {
        REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);
        return  SUCCESS;
    }

    if (redis_read_vlinks_response(redis_sock, &z_ret, elements, ctx) != SUCCESS)
        goto fail;

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);
    return SUCCESS;

fail:
    zval_ptr_dtor_nogc(&z_ret);
    REDIS_RESPONSE_ERROR(redis_sock, z_tab);
    return FAILURE;
}

PHP_REDIS_API int
redis_deserialize_vgetattr_reply(zval *zret, const char *str, size_t len) {
#ifdef HAVE_REDIS_JSON
    #if PHP_VERSION_ID < 80000
        #define PHP_JSON_IN(s) ((char*)s)
    #else
        #define PHP_JSON_IN(s) (s)
    #endif

    if (php_json_decode(zret, PHP_JSON_IN(str), len, 1,
                        PHP_JSON_PARSER_DEFAULT_DEPTH) == FAILURE)
    {
        php_error_docref(NULL, E_WARNING, "Failed to deserialize attributes");
        return FAILURE;
    }

    return SUCCESS;

    #undef PHP_JSON_IN
#else
    php_error_docref(NULL, E_WARNING,
        "PhpRedis is not compiled with JSON support");
    return FAILURE;
#endif
}

PHP_REDIS_API int
redis_vgetattr_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                     zval *z_tab, RedisCmdCtx ctx)
{
    zval z_ret;
    char *attr;
    int len;

    attr = redis_sock_read(redis_sock, &len);
    if (attr == NULL) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    if (len == 0) {
        ZVAL_FALSE(&z_ret);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        if (redis_deserialize_vgetattr_reply(&z_ret, attr, len) != SUCCESS) {
            ZVAL_STRINGL(&z_ret, attr, len);
        }
    } else {
        ZVAL_STRINGL(&z_ret, attr, len);
    }

    efree(attr);

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    return SUCCESS;
}

PHP_REDIS_API int
redis_xinfo_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                  zval *z_tab, RedisCmdCtx ctx)
{
    zval z_ret;
    int elements;

    if (read_mbulk_header(redis_sock, &elements) == SUCCESS) {
        array_init_size(&z_ret, elements > 0 ? elements : 0);
        if (redis_read_xinfo_response(redis_sock, &z_ret, elements) == SUCCESS) {
            REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);
            return SUCCESS;
        }
        zval_ptr_dtor_nogc(&z_ret);
    }

    REDIS_RESPONSE_ERROR(redis_sock, z_tab);
    return FAILURE;
}

PHP_REDIS_API int
redis_acl_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                   zval *z_tab, RedisCmdCtx ctx)
{
    if (ctx.ptr == NULL) {
        return redis_read_variant_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                        redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR) {
        return redis_boolean_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                      redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 1) {
        return redis_string_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                     redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 2) {
        return redis_long_response(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                   redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 3) {
        return redis_acl_getuser_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                       redis_sock, z_tab, redis_empty_ctx);
    } else if (ctx.ptr == PHPREDIS_CTX_PTR + 4) {
        return redis_acl_log_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                   redis_sock, z_tab, redis_empty_ctx);
    } else {
        ZEND_ASSERT(!"memory corruption?");
        return FAILURE;
    }
}

PHP_REDIS_API int
redis_read_acl_log_reply(RedisSock *redis_sock, zval *zret, long count) {
    zval zsub;
    int i, nsub;

    for (i = 0; i < count; i++) {
        if (read_mbulk_header(redis_sock, &nsub) < 0 || nsub % 2 != 0)
            return FAILURE;

        array_init_size(&zsub, nsub > 0 ? nsub : 0);
        if (redis_mbulk_reply_zipped_raw_variant(redis_sock, &zsub, nsub) == FAILURE)
            return FAILURE;

        add_next_index_zval(zret, &zsub);
    }

    return SUCCESS;
}

PHP_REDIS_API int
redis_read_acl_getuser_reply(RedisSock *redis_sock, zval *zret, long count) {
    REDIS_REPLY_TYPE type;
    zval zv;
    char *key, *val;
    long vlen;
    int klen, i;

    for (i = 0; i < count; i += 2) {
        if (!(key = redis_sock_read(redis_sock, &klen)) ||
            redis_read_reply_type(redis_sock, &type, &vlen) < 0 ||
            (type != TYPE_BULK && type != TYPE_MULTIBULK) ||
            vlen > INT_MAX)
        {
            if (key) efree(key);
            return FAILURE;
        }

        if (type == TYPE_BULK) {
            if (!(val = redis_sock_read_bulk_reply(redis_sock, (int)vlen)))
                return FAILURE;
            add_assoc_stringl_ex(zret, key, klen, val, vlen);
            efree(val);
        } else {
            array_init_size(&zv, vlen > 0 ? vlen : 0);
            redis_mbulk_reply_loop(redis_sock, &zv, (int)vlen, UNSERIALIZE_NONE);
            add_assoc_zval_ex(zret, key, klen, &zv);
        }

        efree(key);
    }

    return SUCCESS;
}

int redis_acl_custom_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                           zval *z_tab, RedisCmdCtx ctx,
                           int (*cb)(RedisSock*, zval*, long))
{
    REDIS_REPLY_TYPE type;
    int res = FAILURE;
    zval zret;
    long len;

    if (redis_read_reply_type(redis_sock, &type, &len) == 0 && type == TYPE_MULTIBULK) {
        array_init_size(&zret, len > 0 ? len : 0);

        res = cb(redis_sock, &zret, len);
        if (res == FAILURE) {
            zval_ptr_dtor_nogc(&zret);
            ZVAL_FALSE(&zret);
        }
    } else {
        ZVAL_FALSE(&zret);
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, zret);

    return res;
}

int redis_acl_getuser_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                            zval *z_tab, RedisCmdCtx ctx)
{
    return redis_acl_custom_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                  z_tab, ctx, redis_read_acl_getuser_reply);
}

int redis_acl_log_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                        zval *z_tab, RedisCmdCtx ctx)
{
    return redis_acl_custom_reply(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                  z_tab, ctx, redis_read_acl_log_reply);
}

/* Zipped key => value reply but we don't touch anything (e.g. CONFIG GET) */
PHP_REDIS_API int
redis_mbulk_reply_zipped_raw(INTERNAL_FUNCTION_PARAMETERS,
                             RedisSock *redis_sock, zval *z_tab,
                             RedisCmdCtx ctx)
{
    return redis_mbulk_reply_zipped(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                    z_tab, UNSERIALIZE_NONE, SCORE_DECODE_NONE);
}

/* Zipped key => value reply unserializing keys and decoding the score as an integer (PUBSUB) */
PHP_REDIS_API int
redis_mbulk_reply_zipped_keys_int(INTERNAL_FUNCTION_PARAMETERS,
                                  RedisSock *redis_sock, zval *z_tab,
                                  RedisCmdCtx ctx)
{
    return redis_mbulk_reply_zipped(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                    redis_sock, z_tab, UNSERIALIZE_KEYS,
                                    SCORE_DECODE_INT);
}

/* Zipped key => value reply unserializing keys and decoding the score as a
 * double (ZSET commands) */
PHP_REDIS_API int
redis_mbulk_reply_zipped_keys_dbl(INTERNAL_FUNCTION_PARAMETERS,
                                  RedisSock *redis_sock, zval *z_tab,
                                  RedisCmdCtx ctx)
{
    return redis_mbulk_reply_zipped(INTERNAL_FUNCTION_PARAM_PASSTHRU,
                                    redis_sock, z_tab, UNSERIALIZE_KEYS,
                                    SCORE_DECODE_DOUBLE);
}

/* Zipped key => value reply where only the values are unserialized (e.g.
 * HMGET) */
PHP_REDIS_API int
redis_mbulk_reply_zipped_vals(INTERNAL_FUNCTION_PARAMETERS,
                              RedisSock *redis_sock, zval *z_tab,
                              RedisCmdCtx ctx)
{
    return redis_mbulk_reply_zipped(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                    z_tab, UNSERIALIZE_VALS, SCORE_DECODE_NONE);
}

PHP_REDIS_API int
redis_1_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                 zval *z_tab, RedisCmdCtx ctx)
{
    char *response;
    int response_len;
    zend_bool ret = 0;

    if ((response = redis_sock_read(redis_sock, &response_len)) != NULL) {
        ret = (response[1] == '1');
        efree(response);
    }

    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_BOOL(ret);
    } else {
        add_next_index_bool(z_tab, ret);
    }

    return ret ? SUCCESS : FAILURE;
}

static int redis_bulk_resp_to_zval(RedisSock *redis_sock, zval *zdst, int *dstlen) {
    char *resp;
    int len;

    resp = redis_sock_read(redis_sock, &len);
    if (dstlen) *dstlen = len;

    if (resp == NULL) {
        ZVAL_FALSE(zdst);
        return FAILURE;
    }

    redis_unpack(redis_sock, resp, len, zdst);

    efree(resp);
    return SUCCESS;
}

PHP_REDIS_API int
redis_string_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      zval *z_tab, RedisCmdCtx ctx)
{
    zval zret;
    int ret;

    ret = redis_bulk_resp_to_zval(redis_sock, &zret, NULL);

    REDIS_RETURN_ZVAL(redis_sock, z_tab, zret);

    return ret;
}

PHP_REDIS_API int
redis_bulk_withmeta_response(INTERNAL_FUNCTION_PARAMETERS,
                             RedisSock *redis_sock, zval *z_tab,
                             RedisCmdCtx ctx)
{
    zval zret, zbulk;
    int len, ret;

    ret = redis_bulk_resp_to_zval(redis_sock, &zbulk, &len);

    redis_with_metadata(&zret, &zbulk, len);

    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_ZVAL(&zret, 0, 1);
    } else {
        add_next_index_zval(z_tab, &zret);
    }

    return ret;
}

/* like string response, but never unserialized. */
PHP_REDIS_API int
redis_ping_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                    zval *z_tab, RedisCmdCtx ctx)
{

    char *response;
    int response_len;

    if ((response = redis_sock_read(redis_sock, &response_len))
                                    == NULL)
    {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }
    if (redis_sock_is_atomic(redis_sock)) {
        RETVAL_STRINGL(response, response_len);
    } else {
        add_next_index_stringl(z_tab, response, response_len);
    }

    efree(response);
    return SUCCESS;
}

PHP_REDIS_API int
redis_sock_configure(RedisSock *redis_sock, HashTable *opts)
{
    zend_string *zkey;
    zval *val;

    ZEND_HASH_FOREACH_STR_KEY_VAL(opts, zkey, val) {
        if (zkey == NULL) {
            continue;
        }
        ZVAL_DEREF(val);
        if (zend_string_equals_literal_ci(zkey, "host")) {
            if (Z_TYPE_P(val) != IS_STRING) {
                REDIS_VALUE_EXCEPTION("Invalid host");
                return FAILURE;
            }
            if (redis_sock->host) zend_string_release(redis_sock->host);
            redis_sock->host = zval_get_string(val);
        } else if (zend_string_equals_literal_ci(zkey, "port")) {
            if (Z_TYPE_P(val) != IS_LONG) {
                REDIS_VALUE_EXCEPTION("Invalid port");
                return FAILURE;
            }
            redis_sock->port = zval_get_long(val);
        } else if (zend_string_equals_literal_ci(zkey, "connectTimeout")) {
            if (Z_TYPE_P(val) != IS_LONG && Z_TYPE_P(val) != IS_DOUBLE) {
                REDIS_VALUE_EXCEPTION("Invalid connect timeout");
                return FAILURE;
            }
            redis_sock->timeout = zval_get_double(val);
        } else if (zend_string_equals_literal_ci(zkey, "readTimeout")) {
            if (Z_TYPE_P(val) != IS_LONG && Z_TYPE_P(val) != IS_DOUBLE) {
                REDIS_VALUE_EXCEPTION("Invalid read timeout");
                return FAILURE;
            }
            redis_sock->read_timeout = zval_get_double(val);
        } else if (zend_string_equals_literal_ci(zkey, "persistent")) {
            if (Z_TYPE_P(val) == IS_STRING) {
                if (redis_sock->persistent_id) zend_string_release(redis_sock->persistent_id);
                redis_sock->persistent_id = zval_get_string(val);
                redis_sock->persistent = 1;
            } else {
                redis_sock->persistent = zend_is_true(val);
            }
        } else if (zend_string_equals_literal_ci(zkey, "maxRetries")) {
            if (Z_TYPE_P(val) != IS_LONG || Z_LVAL_P(val) < 0) {
                REDIS_VALUE_EXCEPTION("Max retries must be a non-negative integer");
                return FAILURE;
            }
            redis_sock->max_retries = zval_get_long(val);
        } else if (zend_string_equals_literal_ci(zkey, "retryInterval")) {
            if (Z_TYPE_P(val) != IS_LONG && Z_TYPE_P(val) != IS_DOUBLE) {
                REDIS_VALUE_EXCEPTION("Invalid retry interval");
                return FAILURE;
            }
            redis_sock->retry_interval = zval_get_long(val);
        } else if (zend_string_equals_literal_ci(zkey, "ssl")) {
            if (redis_sock_set_context_zval(redis_sock, val) != SUCCESS) {
                REDIS_VALUE_EXCEPTION("Invalid SSL context options");
                return FAILURE;
            }
        } else if (zend_string_equals_literal_ci(zkey, "auth")) {
            if (Z_TYPE_P(val) != IS_STRING && Z_TYPE_P(val) != IS_ARRAY) {
                REDIS_VALUE_EXCEPTION("Invalid auth credentials");
                return FAILURE;
            }
            redis_sock_set_auth_zval(redis_sock, val);
        } else if (zend_string_equals_literal_ci(zkey, "database")) {
            if (Z_TYPE_P(val) != IS_LONG || Z_LVAL_P(val) < 0 || Z_LVAL_P(val) > INT_MAX) {
                REDIS_VALUE_EXCEPTION("Invalid database number");
                return FAILURE;
            }
            redis_sock->dbNumber = Z_LVAL_P(val);
        } else if (zend_string_equals_literal_ci(zkey, "backoff")) {
            if (redis_sock_set_backoff(redis_sock, val) != SUCCESS) {
                REDIS_VALUE_EXCEPTION("Invalid backoff options");
                return FAILURE;
            }
        } else {
             php_error_docref(NULL, E_WARNING, "Skip unknown option '%s'", ZSTR_VAL(zkey));
        }
    } ZEND_HASH_FOREACH_END();

    return SUCCESS;
}

/**
 * redis_sock_create
 */
PHP_REDIS_API RedisSock*
redis_sock_create(RedisSockType type, const char *host, int host_len, int port,
                  double timeout, double read_timeout, int persistent,
                  const char *persistent_id, long retry_interval)
{
    RedisSock *redis_sock;

    redis_sock = ecalloc(1, sizeof(RedisSock));
    redis_sock->type = type;
    redis_sock->host = zend_string_init(host, host_len, 0);
    redis_sock->status = REDIS_SOCK_STATUS_DISCONNECTED;
    redis_sock->retry_interval = retry_interval * 1000;
    redis_sock->max_retries = 10;
    redis_initialize_backoff(&redis_sock->backoff, redis_sock->retry_interval);
    redis_sock->persistent = persistent;

    if (persistent && persistent_id != NULL) {
        redis_sock->persistent_id = zend_string_init(persistent_id, strlen(persistent_id), 0);
    }

    redis_sock->port    = port;
    redis_sock->timeout = timeout;
    redis_sock->read_timeout = read_timeout;

    redis_sock->serializer = REDIS_SERIALIZER_NONE;
    redis_sock->compression = REDIS_COMPRESSION_NONE;
    redis_sock->mode = ATOMIC;

    return redis_sock;
}

static int redis_uniqid(char *buf, size_t buflen) {
    static unsigned long counter = 0;
    struct timeval tv;

    gettimeofday(&tv, NULL);

    return snprintf(buf, buflen, "phpredis:%08lx%05lx:%08lx",
                    (long)tv.tv_sec, (long)tv.tv_usec, counter++);
}

static int redis_stream_liveness_check(php_stream *stream) {
    return php_stream_set_option(stream, PHP_STREAM_OPTION_CHECK_LIVENESS,
                                 0, NULL) == PHP_STREAM_OPTION_RETURN_OK ?
                                 SUCCESS : FAILURE;
}

/* Try to get the underlying socket FD for use with poll/select.
 * Returns -1 on failure. */
static php_socket_t redis_stream_fd_for_select(php_stream *stream) {
    php_socket_t fd;
    int flags;

    flags = PHP_STREAM_AS_FD_FOR_SELECT | PHP_STREAM_CAST_INTERNAL;
    if (php_stream_cast(stream, flags, (void*)&fd, 1) == FAILURE)
        return -1;

    return fd;
}

static int redis_detect_dirty_config(void) {
    int val = zend_ini_long_literal("redis.pconnect.pool_detect_dirty");

    if (val >= 0 && val <= 2)
        return val;
    else if (val > 2)
        return 2;
    else
        return 0;
}

static int redis_pool_poll_timeout(void) {
    int val = zend_ini_long_literal("redis.pconnect.pool_poll_timeout");
    if (val >= 0)
        return val;

    return 0;
}

static zend_always_inline void
redis_poll_fd_set(php_pollfd *pfd, php_socket_t fd, int events)
{
    pfd->fd = fd;
    pfd->events = events;
    pfd->revents = 0;
}

/* Try to determine if the socket is out of sync (has unconsumed replies) */
static int redis_stream_detect_dirty(php_stream *stream) {
    php_socket_t fd;
    php_pollfd pfd;
    int rv, action;

    /* Short circuit if this is disabled */
    if ((action = redis_detect_dirty_config()) == 0)
        return SUCCESS;

    /* Seek past unconsumed bytes if we detect them */
    if (stream->readpos < stream->writepos) {
        redisDbgFmt("%s on unconsumed buffer (%ld < %ld)",
                    action > 1 ? "Aborting" : "Seeking",
                    (long)stream->readpos, (long)stream->writepos);

        /* Abort if we are configured to immediately fail */
        if (action == 1)
            return FAILURE;

        /* Seek to the end of buffered data */
        zend_off_t offset = stream->writepos - stream->readpos;
        if (php_stream_seek(stream, offset, SEEK_CUR) == FAILURE)
            return FAILURE;
    }

    /* Get the underlying FD */
    if ((fd = redis_stream_fd_for_select(stream)) == -1)
        return FAILURE;

    /* We want to detect a readable socket (it shouldn't be) */
    redis_poll_fd_set(&pfd, fd, PHP_POLLREADABLE);
    rv = php_poll2(&pfd, 1, redis_pool_poll_timeout());

    /* If we detect the socket is readable, it's dirty which is
     * a failure.  Otherwise as best we can tell it's good.
     * TODO:  We could attempt to consume up to N bytes */
    redisDbgFmt("Detected %s socket", rv > 0 ? "readable" : "unreadable");
    return rv == 0 ? SUCCESS : FAILURE;
}

static inline zend_bool
redis_check_echo_response(RedisSock *redis_sock, char *hdr, const char *id,
                          size_t idlen)
{
    char buf[256] = {0};
    size_t len;

    /* Sentinel doesn't have ECHO but it will contain the ID in the error
     * message so just check that */
    if (redis_sock->type == REDIS_SOCK_SENTINEL) {
        return redis_strncmp(hdr, ZEND_STRL("-ERR unknown command")) == 0 &&
               strstr(hdr, id) != NULL;
    }

    /* Non-sentinel: Read and verify the ID */
    return *hdr == TYPE_BULK && atoi(hdr + 1) == idlen &&
           redis_sock_gets(redis_sock, buf, sizeof(buf) - 1, &len) == 0 &&
           redis_strncmp(buf, id, idlen) == 0;
}

static int
redis_sock_check_liveness(RedisSock *redis_sock)
{
    char id[64], inbuf[256] = {0};
    smart_str cmd = {0};
    int idlen, auth;
    size_t len;

    /* Short circuit if PHP detects the stream isn't live */
    if (redis_stream_liveness_check(redis_sock->stream) != SUCCESS)
        goto failure;

    /* Short circuit if we detect the stream is "dirty", can't or are
       configured not to try and fix it */
    if (redis_stream_detect_dirty(redis_sock->stream) != SUCCESS)
        goto failure;

    redis_sock->status = REDIS_SOCK_STATUS_CONNECTED;
    if (!zend_ini_long_literal("redis.pconnect.echo_check_liveness")) {
        return SUCCESS;
    }

    /* AUTH (if we need it) */
    auth = redis_sock_append_auth(redis_sock, &cmd);

    /* ECHO challenge/response */
    idlen = redis_uniqid(id, sizeof(id));
    resp_str_cat_cmd_literal(&cmd, "ECHO", 1);

    resp_str_cat_str(&cmd, id, idlen);

    /* Send command(s) and make sure we can consume reply(ies) */
    if (redis_sock_write(redis_sock, ZSTR_VAL(cmd.s), ZSTR_LEN(cmd.s)) < 0) {
        smart_str_free(&cmd);
        goto failure;
    }

    smart_str_free(&cmd);

    if (redis_sock_gets(redis_sock, inbuf, sizeof(inbuf) - 1, &len) < 0) {
        goto failure;
    }

    if (auth) {
        if (redis_strncmp(inbuf, ZEND_STRL("+OK")) == 0 ||
            redis_strncmp(inbuf, ZEND_STRL("-ERR Client sent AUTH")) == 0)
        {
            /* successfully authenticated or authentication isn't required */
            if (redis_sock_gets(redis_sock, inbuf, sizeof(inbuf) - 1, &len) < 0) {
                goto failure;
            }
        } else if (redis_strncmp(inbuf, ZEND_STRL("-NOAUTH")) == 0) {
            /* connection is fine but authentication failed, next command must
             * fail too */
            if (redis_sock_gets(redis_sock, inbuf, sizeof(inbuf) - 1, &len) < 0
                || redis_strncmp(inbuf, ZEND_STRL("-NOAUTH")) != 0)
            {
                goto failure;
            }
            return SUCCESS;
        } else {
            goto failure;
        }
        redis_sock->status = REDIS_SOCK_STATUS_AUTHENTICATED;
    } else {
        if (redis_strncmp(inbuf, ZEND_STRL("-NOAUTH")) == 0) {
            /* connection is fine but authentication required */
            return SUCCESS;
        }
    }

    if (redis_check_echo_response(redis_sock, inbuf, id, idlen)) {
        return SUCCESS;
    }

failure:
    redis_sock->status = REDIS_SOCK_STATUS_DISCONNECTED;
    if (redis_sock->stream) {
        php_stream_pclose(redis_sock->stream);
        redis_sock->stream = NULL;
    }
    return FAILURE;
}

static php_stream_context *
alloc_stream_context(RedisSock *redis_sock) {
    php_stream_context *ctx;
    zend_string *zkey;
    zval *z_ele;

    if (redis_sock->context == NULL)
        return NULL;

    ctx = php_stream_context_alloc();

    ZEND_HASH_FOREACH_STR_KEY_VAL(redis_sock->context, zkey, z_ele) {
        if (zkey != NULL) {
            php_stream_context_set_option(ctx, "ssl", ZSTR_VAL(zkey), z_ele);
        }
    } ZEND_HASH_FOREACH_END();

    return ctx;
}

/**
 * redis_sock_connect
 */
PHP_REDIS_API int redis_sock_connect(RedisSock *redis_sock)
{
    struct timeval tv, read_tv, *tv_ptr = NULL;
    zend_string *persistent_id = NULL, *estr = NULL;
    char host[1024], scheme[8], *pos, *address;
    const char *fmtstr = "%s://%s:%d";
    int host_len, usocket = 0, err = 0, tcp_flag = 1;
    ConnectionPool *p = NULL;
    php_stream_context *ctx;

    // Monotonically incrementing persistent id counter
    static unsigned long long counter = 0;

    if (redis_sock->stream != NULL) {
        redis_sock_disconnect(redis_sock, 0, 1);
    }

    address = ZSTR_VAL(redis_sock->host);
    if ((pos = strstr(address, "://")) == NULL) {
        strcpy(scheme, redis_sock->context ? "ssl" : "tcp");
    } else {
        snprintf(scheme, sizeof(scheme), "%.*s", (int)(pos - address), address);
        address = pos + sizeof("://") - 1;
    }
    if (address[0] == '/' && redis_sock->port < 1) {
        host_len = snprintf(host, sizeof(host), "unix://%s", address);
        usocket = 1;
    } else {
        if(redis_sock->port == 0)
            redis_sock->port = 6379;

#ifdef HAVE_IPV6
        /* If we've got IPv6 and find a colon in our address, convert to proper
         * IPv6 [host]:port format */
        if (strchr(address, ':') != NULL && strchr(address, '[') == NULL) {
            fmtstr = "%s://[%s]:%d";
        }
#endif
        host_len = snprintf(host, sizeof(host), fmtstr, scheme, address, redis_sock->port);
    }

    if (redis_sock->persistent) {
        if (zend_ini_long_literal("redis.pconnect.pooling_enabled")) {
            p = redis_sock_get_connection_pool(redis_sock);
            if (zend_llist_count(&p->list) > 0) {
                redis_sock->stream = *(php_stream **)zend_llist_get_last(&p->list);
                zend_llist_remove_tail(&p->list);

                if (redis_sock_check_liveness(redis_sock) == SUCCESS) {
                    return SUCCESS;
                }

                p->nb_active--;
            }

            int limit = zend_ini_long_literal("redis.pconnect.connection_limit");
            if (limit > 0 && p->nb_active >= limit) {
                redis_sock_set_err(redis_sock, ZEND_STRL("Connection limit reached"));
                return FAILURE;
            }

            gettimeofday(&tv, NULL);
            persistent_id = strpprintf(0, "phpredis_%ld%ld:%llu", tv.tv_sec,
                                       (long)tv.tv_usec, ++counter);
        } else {
            if (redis_sock->persistent_id) {
                persistent_id = strpprintf(0, "phpredis:%s:%s", host, ZSTR_VAL(redis_sock->persistent_id));
            } else {
                persistent_id = strpprintf(0, "phpredis:%s:%f", host, redis_sock->timeout);
            }
        }
    }

    tv.tv_sec  = (time_t)redis_sock->timeout;
    tv.tv_usec = (int)((redis_sock->timeout - tv.tv_sec) * 1000000);
    if (tv.tv_sec != 0 || tv.tv_usec != 0) {
        tv_ptr = &tv;
    }

    ctx = alloc_stream_context(redis_sock);

    redis_sock->stream = php_stream_xport_create(host, host_len,
        0, STREAM_XPORT_CLIENT | STREAM_XPORT_CONNECT,
        persistent_id ? ZSTR_VAL(persistent_id) : NULL,
        tv_ptr, ctx, &estr, &err);

    /* On successful connection, the context is moved into the stream so we
     * can disown it. If the connection failed, clean it up ourselves. */
    if (ctx && redis_sock->stream) {
        ZEND_ASSERT(GC_REFCOUNT(ctx->res) == 2);
        GC_DELREF(ctx->res);
    } else if (ctx) {
        ZEND_ASSERT(GC_REFCOUNT(ctx->res) == 1);
        zend_list_delete(ctx->res);
    }

    if (persistent_id) {
        zend_string_release(persistent_id);
    }

    if (!redis_sock->stream) {
        if (estr) {
            redis_sock_set_err(redis_sock, ZSTR_VAL(estr), ZSTR_LEN(estr));
            zend_string_release(estr);
        }
        return FAILURE;
    }

    if (p) p->nb_active++;

    /* Attempt to set TCP_NODELAY/TCP_KEEPALIVE if we're not using a unix socket. */
    if (!usocket) {
        php_netstream_data_t *sock = (php_netstream_data_t*)redis_sock->stream->abstract;
        int tcp_keepalive = redis_sock->tcp_keepalive;
        err = setsockopt(sock->socket, IPPROTO_TCP, TCP_NODELAY, (char*) &tcp_flag, sizeof(tcp_flag));
        PHPREDIS_NOTUSED(err);
        err = setsockopt(sock->socket, SOL_SOCKET, SO_KEEPALIVE, (char*) &tcp_keepalive, sizeof(tcp_keepalive));
        PHPREDIS_NOTUSED(err);
    }

    php_stream_auto_cleanup(redis_sock->stream);

    read_tv.tv_sec  = (time_t)redis_sock->read_timeout;
    read_tv.tv_usec = (int)((redis_sock->read_timeout - read_tv.tv_sec) * 1000000);

    if (read_tv.tv_sec != 0 || read_tv.tv_usec != 0) {
        php_stream_set_option(redis_sock->stream,PHP_STREAM_OPTION_READ_TIMEOUT,
            0, &read_tv);
    }
    php_stream_set_option(redis_sock->stream,
        PHP_STREAM_OPTION_WRITE_BUFFER, PHP_STREAM_BUFFER_NONE, NULL);

    redis_sock->status = REDIS_SOCK_STATUS_CONNECTED;

    return SUCCESS;
}

/**
 * redis_sock_server_open
 */
PHP_REDIS_API int
redis_sock_server_open(RedisSock *redis_sock)
{
    if (redis_sock) {
        switch (redis_sock->status) {
        case REDIS_SOCK_STATUS_DISCONNECTED:
            if (redis_sock_connect(redis_sock) != SUCCESS) {
                break;
            }
            redis_sock->status = REDIS_SOCK_STATUS_CONNECTED;
            // fall through
        case REDIS_SOCK_STATUS_CONNECTED:
            if (redis_sock_auth(redis_sock) != SUCCESS) {
                break;
            }
            redis_sock->status = REDIS_SOCK_STATUS_AUTHENTICATED;
            // fall through
        case REDIS_SOCK_STATUS_AUTHENTICATED:
            if (redis_sock->dbNumber && redis_select_db(redis_sock) != SUCCESS) {
                break;
            }
            redis_sock->status = REDIS_SOCK_STATUS_READY;
            // fall through
        case REDIS_SOCK_STATUS_READY:
            return SUCCESS;
        default:
            return FAILURE;
        }
    }
    return FAILURE;
}

/**
 * redis_sock_disconnect
 */
PHP_REDIS_API int
redis_sock_disconnect(RedisSock *redis_sock, int force, int is_reset_mode)
{
    if (redis_sock == NULL) {
        return FAILURE;
    } else if (redis_sock->stream) {
        if (redis_sock->persistent) {
            ConnectionPool *p = NULL;
            if (zend_ini_long_literal("redis.pconnect.pooling_enabled")) {
                p = redis_sock_get_connection_pool(redis_sock);
            }
            if (force || !redis_sock_is_atomic(redis_sock)) {
                php_stream_pclose(redis_sock->stream);
                redis_free_reply_callbacks(redis_sock);
                if (p) p->nb_active--;
            } else if (p) {
                zend_llist_prepend_element(&p->list, &redis_sock->stream);
            }
        } else {
            php_stream_close(redis_sock->stream);
        }
        redis_sock->stream = NULL;
    }
    if (is_reset_mode) {
        redis_sock->mode = ATOMIC;
    }
    redis_sock->status = REDIS_SOCK_STATUS_DISCONNECTED;
    redis_sock->watching = 0;

    return SUCCESS;
}

/**
 * redis_sock_set_err
 */
PHP_REDIS_API void
redis_sock_set_err(RedisSock *redis_sock, const char *msg, int msg_len)
{
    redis_sock_clear_err(redis_sock);

    if (msg != NULL && msg_len > 0) {
        // Copy in our new error message
        redis_sock->err = zend_string_init(msg, msg_len, 0);
    }
}

PHP_REDIS_API void redis_sock_clear_err(RedisSock *redis_sock) {
    if (redis_sock->err == NULL)
        return;

    zend_string_release(redis_sock->err);
    redis_sock->err = NULL;
}

static zend_always_inline void
redis_gc_try_addref(HashTable *ht)
{
#if PHP_VERSION_ID < 80000
    if (ht && !(GC_FLAGS(ht) & GC_IMMUTABLE)) {
        GC_ADDREF(ht);
    }
#else
    GC_TRY_ADDREF(ht);
#endif
}


void redis_sock_set_context(RedisSock *redis_sock, HashTable *ht) {
    redis_sock_free_context(redis_sock);

    if (ht == NULL)
        return;

    redis_gc_try_addref(ht);
    redis_sock->context = ht;
}

int redis_sock_set_context_zval(RedisSock *redis_sock, zval *zv) {
    HashTable *ht;

    ht = zv && Z_TYPE_P(zv) == IS_ARRAY ? Z_ARRVAL_P(zv) : NULL;

    redis_sock_set_context(redis_sock, ht);

    return ht ? SUCCESS : FAILURE;
}

PHP_REDIS_API int
redis_sock_set_backoff(RedisSock *redis_sock, zval *options)
{
    zend_string *zkey;
    zend_long val;
    zval *z_ele;

    if (!redis_sock || Z_TYPE_P(options) != IS_ARRAY) {
        return FAILURE;
    }

    ZEND_HASH_FOREACH_STR_KEY_VAL(Z_ARRVAL_P(options), zkey, z_ele) {
        if (zkey != NULL) {
            ZVAL_DEREF(z_ele);
            if (zend_string_equals_literal_ci(zkey, "algorithm")) {
                if ((val = zval_get_long(z_ele)) < 0 || val >= REDIS_BACKOFF_ALGORITHMS) {
                    return FAILURE;
                }
                redis_sock->backoff.algorithm = val;
            } else if (zend_string_equals_literal_ci(zkey, "base")) {
                if ((val = zval_get_long(z_ele)) < 0) {
                    return FAILURE;
                }
                redis_sock->backoff.base = val * 1000;
            } else if (zend_string_equals_literal_ci(zkey, "cap")) {
                if ((val = zval_get_long(z_ele)) < 0) {
                    return FAILURE;
                }
                redis_sock->backoff.cap = val * 1000;
            } else {
                php_error_docref(NULL, E_WARNING, "Skip unknown backoff option '%s'", ZSTR_VAL(zkey));
            }
        }
    } ZEND_HASH_FOREACH_END();

    return SUCCESS;
}

/**
 * redis_sock_read_multibulk_reply
 */
PHP_REDIS_API int redis_sock_read_multibulk_reply(INTERNAL_FUNCTION_PARAMETERS,
                                           RedisSock *redis_sock, zval *z_tab,
                                           RedisCmdCtx ctx)
{
    zval z_multi_result;
    int numElems;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }
    if (numElems == -1 && redis_sock->null_mbulk_as_null) {
        ZVAL_NULL(&z_multi_result);
    } else if (numElems < 1) {
        ZVAL_EMPTY_ARRAY(&z_multi_result);
    } else {
        array_init_size(&z_multi_result, numElems);
        redis_mbulk_reply_loop(redis_sock, &z_multi_result, numElems, UNSERIALIZE_ALL);
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_multi_result);

    return 0;
}

/* Like multibulk reply, but don't touch the values, they won't be unserialized
 * (this is used by HKEYS). */
PHP_REDIS_API int
redis_mbulk_reply_raw(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      zval *z_tab, RedisCmdCtx ctx)
{
    int numElems;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    zval z_multi_result;

    if (numElems < 1) {
        ZVAL_EMPTY_ARRAY(&z_multi_result);
    } else {
        array_init_size(&z_multi_result, numElems); /* pre-allocate array for multi's results. */
        redis_mbulk_reply_loop(redis_sock, &z_multi_result, numElems, UNSERIALIZE_NONE);
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_multi_result);

    return SUCCESS;
}

PHP_REDIS_API int
redis_mbulk_reply_double(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                         zval *z_tab, RedisCmdCtx ctx)
{
    char *line;
    int i, numElems, len;
    zval z_multi_result;

    if (read_mbulk_header(redis_sock, &numElems) < 0) {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    if (numElems < 1) {
        ZVAL_EMPTY_ARRAY(&z_multi_result);
    } else {
        array_init_size(&z_multi_result, numElems);
        for (i = 0; i < numElems; ++i) {
            if ((line = redis_sock_read(redis_sock, &len)) == NULL) {
                add_next_index_bool(&z_multi_result, 0);
                continue;
            }
            add_next_index_double(&z_multi_result, atof(line));
            efree(line);
        }
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_multi_result);

    return SUCCESS;
}

/* Bulk-string reader that materializes the body directly into a zend_string,
 * avoiding the extra emalloc+memcpy a caller would otherwise pay to turn a
 * raw char* reply into a PHP string. The caller must have already handled the
 * null-bulk ($-1) case; bytes is the non-negative payload length. */
static zend_string *
redis_sock_read_bulk_zstr(RedisSock *redis_sock, int bytes)
{
    size_t offset = 0, nbytes;
    zend_string *zstr;
    ssize_t got;

    if (bytes < 0 || bytes > INT_MAX - 2) {
        zend_throw_exception_ex(redis_exception_ce, 0,
            "protocol error, invalid bulk length: %d", bytes);
        return NULL;
    }

    if (-1 == redis_check_eof(redis_sock, 1, 0)) {
        return NULL;
    }

    /* Over-allocate by two so the trailing \r\n lands in the same buffer as
     * the body; the reported length is trimmed back to the payload below. */
    nbytes = (size_t)bytes + 2;
    zstr = zend_string_alloc(nbytes, 0);

    while (offset < nbytes) {
        got = redis_sock_read_raw(redis_sock, ZSTR_VAL(zstr) + offset, nbytes - offset);
        if (got <= 0)
            break;

        offset += got;
    }

    if (offset < nbytes) {
        REDIS_THROW_EXCEPTION("socket error on read socket", 0);
        zend_string_release(zstr);
        return NULL;
    }

    ZSTR_LEN(zstr) = bytes;
    ZSTR_VAL(zstr)[bytes] = '\0';

    return zstr;
}

/* A zend_string-returning counterpart to redis_sock_read, used by the
 * multibulk loop so element strings can be moved into the result array with
 * no intermediate copy. Returns NULL for null/error replies, mirroring
 * redis_sock_read's NULL semantics. */
static zend_string *
redis_sock_read_zstr(RedisSock *redis_sock)
{
    char inbuf[4096];
    size_t len;

    if (redis_sock_gets(redis_sock, inbuf, sizeof(inbuf) - 1, &len) < 0 || len < 1) {
        return NULL;
    }

    switch (inbuf[0]) {
        case '-':
            redis_sock_set_err(redis_sock, inbuf + 1, len - 1);
            redis_error_throw(redis_sock);
            return NULL;
        case '$':
            {
                char *endptr;
                long long n;

                errno = 0;
                n = strtoll(inbuf + 1, &endptr, 10);
                if (endptr == inbuf + 1 || errno == ERANGE || n < -1 || n > INT_MAX) {
                    return NULL;
                }
                if (n == -1) {
                    return NULL;
                }
                return redis_sock_read_bulk_zstr(redis_sock, (int)n);
            }
        case '*':
            /* For null multi-bulk replies (like timeouts from brpoplpush): */
            if (len > 2 && memcmp(inbuf + 1, "-1", 2) == 0) {
                return NULL;
            }
            REDIS_FALLTHROUGH;
        case '+':
        case ':':
            /* Single line reply (+OK or :123), kept verbatim like
             * redis_sock_read does, including the leading type byte. */
            if (len > 1) {
                return zend_string_init(inbuf, len, 0);
            }
            REDIS_FALLTHROUGH;
        default:
            zend_throw_exception_ex(redis_exception_ce, 0,
                "protocol error, got '%c' as reply type byte\n",
                inbuf[0]
            );
    }

    return NULL;
}

PHP_REDIS_API int
redis_mbulk_reply_loop(RedisSock *redis_sock, zval *z_tab, int count,
                       int unserialize)
{
    zval z_value;
    zend_string *zstr;
    int i;

    /* When neither a serializer nor compression is active, redis_unpack just
     * copies the bytes into a new string, so we can move our zend_string into
     * the array directly and skip the unpack entirely. */
    int repack = redis_sock->serializer != REDIS_SERIALIZER_NONE ||
                 redis_sock->compression != REDIS_COMPRESSION_NONE;

    for (i = 0; i < count; ++i) {
        if ((zstr = redis_sock_read_zstr(redis_sock)) == NULL) {
            add_next_index_bool(z_tab, 0);
            if (EG(exception) || redis_sock->stream == NULL ||
                redis_sock->status == REDIS_SOCK_STATUS_FAILED
            ) {
                return FAILURE;
            }
            continue;
        }

        /* We will attempt unserialization, if we're unserializing everything,
         * or if we're unserializing keys and we're on a key, or we're
         * unserializing values and we're on a value! */
        int unwrap = (
            (unserialize == UNSERIALIZE_ALL) ||
            (unserialize == UNSERIALIZE_KEYS && i % 2 == 0) ||
            (unserialize == UNSERIALIZE_VALS && i % 2 != 0)
        );

        if (unwrap && repack) {
            redis_unpack(redis_sock, ZSTR_VAL(zstr), ZSTR_LEN(zstr), &z_value);
            zend_string_release(zstr);
        } else {
            ZVAL_STR(&z_value, zstr);
        }
        zend_hash_next_index_insert_new(Z_ARRVAL_P(z_tab), &z_value);
    }

    return SUCCESS;
}

static int
redis_mbulk_reply_zipped_raw_variant(RedisSock *redis_sock, zval *zret, int count) {
    REDIS_REPLY_TYPE type;
    char *key, *val;
    int keylen, i;
    zend_long lval;
    double dval;
    long vallen;

    for (i = 0; i < count; i+= 2) {
        /* Keys should always be bulk strings */
        if ((key = redis_sock_read(redis_sock, &keylen)) == NULL)
            return FAILURE;

        /* This can vary */
        if (redis_read_reply_type(redis_sock, &type, &vallen) < 0) {
            efree(key);
            return FAILURE;
        }

        if (type == TYPE_BULK) {
            if (vallen > INT_MAX || (val = redis_sock_read_bulk_reply(redis_sock, (int)vallen)) == NULL) {
                efree(key);
                return FAILURE;
            }

            /* Possibly overkill, but provides really nice types */
            switch (is_numeric_string(val, vallen, &lval, &dval, 0)) {
                case IS_LONG:
                    add_assoc_long_ex(zret, key, keylen, lval);
                    break;
                case IS_DOUBLE:
                    add_assoc_double_ex(zret, key, keylen, dval);
                    break;
                default:
                    add_assoc_stringl_ex(zret, key, keylen, val, vallen);
            }

            efree(val);
        } else if (type == TYPE_INT) {
            add_assoc_long_ex(zret, key, keylen, (zend_long)vallen);
        } else {
            add_assoc_null_ex(zret, key, keylen);
        }

        efree(key);
    }

    return SUCCESS;
}

/* Specialized multibulk processing for HMGET where we need to pair requested
 * keys with their returned values */
PHP_REDIS_API int
redis_mbulk_reply_assoc(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                        zval *z_tab, RedisCmdCtx ctx)
{
    HashTable *htctx = ctx.ptr;
    int numElems;
    zval *zfield;
    char *rresp;
    int len;

    ZEND_ASSERT(htctx != NULL);

    if (read_mbulk_header(redis_sock, &numElems) < 0 ||
        zend_hash_num_elements(htctx) != numElems)
    {
        REDIS_RESPONSE_ERROR(redis_sock, z_tab);
        return FAILURE;
    }

    zval z_multi_result, zunpacked;

    GC_ADDREF(htctx);

    ZEND_HASH_FOREACH_VAL(htctx, zfield) {
        rresp = redis_sock_read(redis_sock, &len);

        if (rresp != NULL) {
            redis_unpack(redis_sock, rresp, len, &zunpacked);
            efree(rresp);
        } else {
            ZVAL_FALSE(&zunpacked);
        }

        ZVAL_COPY_VALUE(zfield, &zunpacked);
    } ZEND_HASH_FOREACH_END();

    ZVAL_ARR(&z_multi_result, htctx);
    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_multi_result);

    return SUCCESS;
}

/**
 * redis_sock_write
 */
PHP_REDIS_API int
redis_sock_write(RedisSock *redis_sock, const char *cmd, size_t sz)
{
    if (redis_check_eof(redis_sock, 0, 0) == 0 &&
        redis_sock_write_raw(redis_sock, cmd, sz) == sz)
    {
        return sz;
    }

    return -1;
}

/* Grow array to double size if we need more space */
fold_item*
redis_add_reply_callback(RedisSock *redis_sock) {
    if (UNEXPECTED(redis_sock->reply_callback_count == redis_sock->reply_callback_capacity)) {
        if (redis_sock->reply_callback_capacity == 0) {
            redis_sock->reply_callback_capacity = REDIS_CALLBACKS_INIT_SIZE;
        } else if (redis_sock->reply_callback_capacity < REDIS_CALLBACKS_MAX_DOUBLE) {
            redis_sock->reply_callback_capacity *= 2;
        } else {
            redis_sock->reply_callback_capacity += REDIS_CALLBACKS_ADD_SIZE;
        }
        redis_sock->reply_callback = erealloc(redis_sock->reply_callback, redis_sock->reply_callback_capacity * sizeof(fold_item));
    }
    return &redis_sock->reply_callback[redis_sock->reply_callback_count++];
}

void
redis_free_reply_callbacks(RedisSock *redis_sock)
{
    if (redis_sock->reply_callback != NULL) {
        size_t i;

        for (i = 0; i < redis_sock->reply_callback_count; i++) {
            redis_cmd_ctx_free(redis_sock->reply_callback[i].ctx);
        }

        efree(redis_sock->reply_callback);
        redis_sock->reply_callback = NULL;
        redis_sock->reply_callback_count = 0;
        redis_sock->reply_callback_capacity = 0;
    }
}

static void
redis_sock_release_hello(struct RedisHello *hello) {
    if (hello->server) {
        zend_string_release(hello->server);
        hello->server = NULL;
    }

    if (hello->version) {
        zend_string_release(hello->version);
        hello->version = NULL;
    }
}

/**
 * redis_free_socket
 */
PHP_REDIS_API void redis_free_socket(RedisSock *redis_sock)
{
    int i;

    if (redis_sock->prefix) {
        zend_string_release(redis_sock->prefix);
    }
    smart_string_free(&redis_sock->pipeline_cmd);
    if (redis_sock->err) {
        zend_string_release(redis_sock->err);
    }
    if (redis_sock->persistent_id) {
        zend_string_release(redis_sock->persistent_id);
    }
    if (redis_sock->host) {
        zend_string_release(redis_sock->host);
    }
    for (i = 0; i < REDIS_SUBS_BUCKETS; ++i) {
        if (redis_sock->subs[i]) {
            zend_hash_destroy(redis_sock->subs[i]);
            efree(redis_sock->subs[i]);
            redis_sock->subs[i] = NULL;
        }
    }
    redis_sock_free_auth(redis_sock);
    redis_free_reply_callbacks(redis_sock);
    redis_sock_release_hello(&redis_sock->hello);
    redis_sock_free_context(redis_sock);

    efree(redis_sock);
}

#ifdef HAVE_REDIS_LZ4
/* Implementation of CRC8 for our LZ4 checksum value */
static uint8_t crc8(unsigned char *input, size_t len) {
    size_t i;
    uint8_t crc = 0xFF;

    while (len--) {
        crc ^= *input++;
        for (i = 0; i < 8; i++) {
            if (crc & 0x80)
                crc = (uint8_t)(crc << 1) ^ 0x31;
            else
                crc <<= 1;
        }
    }

    return crc;
}
#endif

PHP_REDIS_API int
redis_compress(RedisSock *redis_sock, char **dst, size_t *dstlen, char *buf, size_t len) {
    switch (redis_sock->compression) {
        case REDIS_COMPRESSION_LZF:
#ifdef HAVE_REDIS_LZF
            {
                char *data;
                uint32_t res;
                double size;

                /* preserve compatibility with PECL lzf_compress margin (greater of 4% and LZF_MARGIN) */
                size = len + MIN(UINT_MAX - len, MAX(LZF_MARGIN, len / 25));
                data = emalloc(size);
                if ((res = lzf_compress(buf, len, data, size)) > 0) {
                    *dst = data;
                    *dstlen = res;
                    return 1;
                }
                efree(data);
            }
#endif
            break;
        case REDIS_COMPRESSION_ZSTD:
#ifdef HAVE_REDIS_ZSTD
            {
                char *data;
                size_t size;
                int level;

                if (redis_sock->compression_level < 1) {
#ifdef ZSTD_CLEVEL_DEFAULT
                    level = ZSTD_CLEVEL_DEFAULT;
#else
                    level = 3;
#endif
                } else if (redis_sock->compression_level > ZSTD_maxCLevel()) {
                    level = ZSTD_maxCLevel();
                } else {
                    level = redis_sock->compression_level;
                }

                size = ZSTD_compressBound(len);
                data = emalloc(size);
                size = ZSTD_compress(data, size, buf, len, level);
                if (!ZSTD_isError(size)) {
                    *dst = erealloc(data, size);
                    *dstlen = size;
                    return 1;
                }
                efree(data);
            }
#endif
            break;
        case REDIS_COMPRESSION_LZ4:
#ifdef HAVE_REDIS_LZ4
            {
                /* Compressing empty data is pointless */
                if (len < 1)
                    break;

                /* Compressing more than INT_MAX bytes would require multiple blocks */
                if (len > INT_MAX) {
                    php_error_docref(NULL, E_WARNING,
                        "LZ4: compressing > %d bytes not supported", INT_MAX);
                    break;
                }

                int old_len = len, lz4len, lz4bound;
                uint8_t crc = crc8((unsigned char*)&old_len, sizeof(old_len));
                char *lz4buf, *lz4pos;

                lz4bound = LZ4_compressBound(len);
                lz4buf = emalloc(REDIS_LZ4_HDR_SIZE + lz4bound);
                lz4pos = lz4buf;

                /* Copy and move past crc8 length checksum */
                memcpy(lz4pos, &crc, sizeof(crc));
                lz4pos += sizeof(crc);

                /* Copy and advance past length */
                memcpy(lz4pos, &old_len, sizeof(old_len));
                lz4pos += sizeof(old_len);

                if (redis_sock->compression_level <= 0 || redis_sock->compression_level > REDIS_LZ4_MAX_CLEVEL) {
                    lz4len = LZ4_compress_default(buf, lz4pos, old_len, lz4bound);
                } else {
                    lz4len = LZ4_compress_HC(buf, lz4pos, old_len, lz4bound, redis_sock->compression_level);
                }

                if (lz4len <= 0) {
                    efree(lz4buf);
                    break;
                }

                *dst = lz4buf;
                *dstlen = lz4len + REDIS_LZ4_HDR_SIZE;
                return 1;
            }
#endif
            break;
    }

    *dst = buf;
    *dstlen = len;
    return 0;
}

PHP_REDIS_API int
redis_uncompress(RedisSock *redis_sock, char **dst, size_t *dstlen, const char *src, size_t len) {
    switch (redis_sock->compression) {
        case REDIS_COMPRESSION_LZF:
#ifdef HAVE_REDIS_LZF
            {
                char *data = NULL;
                uint32_t res;
                int i;

                if (len == 0)
                    break;

                /* Grow our buffer until we succeed or get a non E2BIG error */
                errno = E2BIG;
                for (i = 2; errno == E2BIG; i *= 2) {
                    data = erealloc(data, len * i);
                    if ((res = lzf_decompress(src, len, data, len * i)) > 0) {
                        *dst = data;
                        *dstlen = res;
                        return 1;
                    }
                }

                efree(data);
                break;
            }
#endif
            break;
        case REDIS_COMPRESSION_ZSTD:
#ifdef HAVE_REDIS_ZSTD
            {
                char *data;
                unsigned long long zlen;

                zlen = ZSTD_getFrameContentSize(src, len);
                if (zlen == ZSTD_CONTENTSIZE_ERROR || zlen == ZSTD_CONTENTSIZE_UNKNOWN || zlen > INT_MAX)
                    break;

                data = emalloc(zlen);
                *dstlen = ZSTD_decompress(data, zlen, src, len);
                if (ZSTD_isError(*dstlen) || *dstlen != zlen) {
                    efree(data);
                    break;
                }

                *dst = data;
                return 1;
            }
#endif
            break;
        case REDIS_COMPRESSION_LZ4:
#ifdef HAVE_REDIS_LZ4
            {
                char *data;
                int datalen, res;
                uint8_t lz4crc;

                /* We must have at least enough bytes for our header, and can't have more than
                 * INT_MAX + our header size. */
                if (len < REDIS_LZ4_HDR_SIZE || len > INT_MAX + REDIS_LZ4_HDR_SIZE)
                    break;

                /* Operate on copies in case our CRC fails */
                const char *copy = src;
                size_t copylen = len;

                /* Read in our header bytes */
                memcpy(&lz4crc, copy, sizeof(uint8_t));
                copy += sizeof(uint8_t); copylen -= sizeof(uint8_t);
                memcpy(&datalen, copy, sizeof(int));
                copy += sizeof(int); copylen -= sizeof(int);

                if (datalen <= 0)
                    break;

                /* Make sure our CRC matches (TODO:  Maybe issue a docref error?) */
                if (crc8((unsigned char*)&datalen, sizeof(datalen)) != lz4crc)
                    break;

                /* Finally attempt decompression */
                data = emalloc(datalen);
                res = LZ4_decompress_safe(copy, data, copylen, datalen);
                if (res == datalen) {
                    *dst = data;
                    *dstlen = res;
                    return 1;
                }

                efree(data);
            }
#endif
            break;
    }

    *dst = (char*)src;
    *dstlen = len;
    return 0;
}

static int serialize_generic_zval(char **dst, size_t *len, zval *zsrc) {
    zend_string *zstr;

    zstr = zval_get_string_func(zsrc);
    if (ZSTR_IS_INTERNED(zstr)) {
        *dst = ZSTR_VAL(zstr);
        *len = ZSTR_LEN(zstr);
        return 0;
    }

    *dst = estrndup(ZSTR_VAL(zstr), ZSTR_LEN(zstr));
    *len = ZSTR_LEN(zstr);

    zend_string_release(zstr);

    return 1;
}


PHP_REDIS_API int
redis_pack(RedisSock *redis_sock, zval *z, char **val, size_t *val_len) {
    size_t tmplen;
    int tmpfree;
    char *tmp;

    /* Don't pack actual numbers if the user asked us not to */
    if (UNEXPECTED(redis_sock->pack_ignore_numbers &&
                   (Z_TYPE_P(z) == IS_LONG || Z_TYPE_P(z) == IS_DOUBLE)))
    {
        return serialize_generic_zval(val, val_len, z);
    }

    /* First serialize */
    tmpfree = redis_serialize(redis_sock, z, &tmp, &tmplen);

    /* Now attempt compression */
    if (redis_compress(redis_sock, val, val_len, tmp, tmplen)) {
        if (tmpfree) efree(tmp);
        return 1;
    }

    return tmpfree;
}

PHP_REDIS_API int
redis_unpack(RedisSock *redis_sock, const char *src, int srclen, zval *zdst) {
    zend_long lval;
    double dval;
    size_t len;
    char *buf;

    if (UNEXPECTED((redis_sock->serializer != REDIS_SERIALIZER_NONE ||
                    redis_sock->compression != REDIS_COMPRESSION_NONE) &&
                    redis_sock->pack_ignore_numbers) &&
                    srclen > 0 && srclen < 512)
    {
        switch (is_numeric_string(src, srclen, &lval, &dval, 0)) {
            case IS_LONG:
                ZVAL_LONG(zdst, lval);
                return 1;
            case IS_DOUBLE:
                ZVAL_DOUBLE(zdst, dval);
                return 1;
            default:
                /* Fallthrough */
                break;
        }
    }

    /* Input string is empty */
    if (srclen == 0) {
        ZVAL_STR(zdst, ZSTR_EMPTY_ALLOC());
        return 1;
    }

    /* Uncompress, then unserialize */
    if (redis_uncompress(redis_sock, &buf, &len, src, srclen)) {
        if (!redis_unserialize(redis_sock, buf, len, zdst)) {
            ZVAL_STRINGL_FAST(zdst, buf, len);
        }
        efree(buf);
        return 1;
    }

    if (!redis_unserialize(redis_sock, src, srclen, zdst)) {
        ZVAL_STRINGL_FAST(zdst, src, srclen);
    }

    return 1;
}

PHP_REDIS_API int

redis_serialize(RedisSock *redis_sock, zval *z, char **val, size_t *val_len)
{
    php_serialize_data_t ht;

    smart_str sstr = {0};
#ifdef HAVE_REDIS_IGBINARY
    size_t sz;
    uint8_t *val8;
#endif

    *val = "";
    *val_len = 0;
    switch(redis_sock->serializer) {
        case REDIS_SERIALIZER_NONE:
            switch(Z_TYPE_P(z)) {
                case IS_STRING:
                    *val = Z_STRVAL_P(z);
                    *val_len = Z_STRLEN_P(z);
                    break;

                case IS_OBJECT:
                    *val = "Object";
                    *val_len = 6;
                    break;

                case IS_ARRAY:
                    *val = "Array";
                    *val_len = 5;
                    break;

                default:
                    return serialize_generic_zval(val, val_len, z);
            }
            break;
        case REDIS_SERIALIZER_PHP:
            PHP_VAR_SERIALIZE_INIT(ht);
            php_var_serialize(&sstr, z, &ht);
            PHP_VAR_SERIALIZE_DESTROY(ht);

            if (!sstr.s) {
                break;
            }
            *val = estrndup(ZSTR_VAL(sstr.s), ZSTR_LEN(sstr.s));
            *val_len = ZSTR_LEN(sstr.s);

            smart_str_free(&sstr);

            return 1;

        case REDIS_SERIALIZER_MSGPACK:
#ifdef HAVE_REDIS_MSGPACK
            php_msgpack_serialize(&sstr, z);
            *val = estrndup(ZSTR_VAL(sstr.s), ZSTR_LEN(sstr.s));
            *val_len = ZSTR_LEN(sstr.s);
            smart_str_free(&sstr);

            return 1;
#endif
            break;
        case REDIS_SERIALIZER_IGBINARY:
#ifdef HAVE_REDIS_IGBINARY
            if(igbinary_serialize(&val8, (size_t *)&sz, z) == 0) {
                *val = (char*)val8;
                *val_len = sz;
                return 1;
            }
#endif
            break;
        case REDIS_SERIALIZER_JSON:
#ifdef HAVE_REDIS_JSON
            php_json_encode(&sstr, z, PHP_JSON_OBJECT_AS_ARRAY);
            *val = estrndup(ZSTR_VAL(sstr.s), ZSTR_LEN(sstr.s));
            *val_len = ZSTR_LEN(sstr.s);
            smart_str_free(&sstr);
            return 1;
#endif
            break;
        default:
            ZEND_UNREACHABLE();
    }

    return 0;
}

PHP_REDIS_API int
redis_unserialize(RedisSock* redis_sock, const char *val, int val_len,
                  zval *z_ret)
{

    php_unserialize_data_t var_hash;
    int ret = 0;

    switch(redis_sock->serializer) {
        case REDIS_SERIALIZER_NONE:
            /* Nothing to do */
            break;
        case REDIS_SERIALIZER_PHP:
            PHP_VAR_UNSERIALIZE_INIT(var_hash);

            ret = php_var_unserialize(z_ret, (const unsigned char **)&val,
                                      (const unsigned char *)val + val_len,
                                       &var_hash);

            PHP_VAR_UNSERIALIZE_DESTROY(var_hash);
            break;

        case REDIS_SERIALIZER_MSGPACK:
#ifdef HAVE_REDIS_MSGPACK
            ret = !php_msgpack_unserialize(z_ret, (char *)val, (size_t)val_len);
#endif
            break;

        case REDIS_SERIALIZER_IGBINARY:
#ifdef HAVE_REDIS_IGBINARY
            /*
             * Check if the given string starts with an igbinary header.
             *
             * A modern igbinary string consists of the following format:
             *
             * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-
             * | header (4) | type (1) | ... (n) |  NUL (1) |
             * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-
             *
             * With header being three zero bytes and one non-zero version
             * specifier.  At the time of this comment, there is only version
             * 0x01 and 0x02, but newer versions will use subsequent
             * values.
             *
             * Not all versions contain a trailing \x00 so don't check for that.
             */
            if (val_len < 5 || memcmp(val, "\x00\x00\x00", 3) || val[3] < '\x01' || val[3] > '\x04')
            {
                /* This is most definitely not an igbinary string, so do
                   not try to unserialize this as one. */
                break;
            }

            ret = !igbinary_unserialize((const uint8_t *)val, (size_t)val_len, z_ret);
#endif
            break;
        case REDIS_SERIALIZER_JSON:
#ifdef HAVE_REDIS_JSON
            ret = !php_json_decode(z_ret, (char *)val, val_len, 1, PHP_JSON_PARSER_DEFAULT_DEPTH);
#endif
            break;
        default:
            ZEND_UNREACHABLE();
    }

    return ret;
}

PHP_REDIS_API int
redis_key_prefix(RedisSock *redis_sock, char **key, size_t *key_len) {
    int ret_len;
    char *ret;

    if (redis_sock->prefix == NULL) {
        return 0;
    }

    ret_len = ZSTR_LEN(redis_sock->prefix) + *key_len;
    ret = ecalloc(1 + ret_len, 1);
    memcpy(ret, ZSTR_VAL(redis_sock->prefix), ZSTR_LEN(redis_sock->prefix));
    memcpy(ret + ZSTR_LEN(redis_sock->prefix), *key, *key_len);

    *key = ret;
    *key_len = ret_len;
    return 1;
}

/* This is very similar to PHP >= 7.4 zend_string_concat2 only we are taking
 * two zend_string arguments rather than two char*, size_t pairs */
static zend_string *redis_zstr_concat(zend_string *prefix, zend_string *suffix) {
    zend_string *res;
    size_t len;

    ZEND_ASSERT(prefix != NULL && suffix != NULL);

    len = ZSTR_LEN(prefix) + ZSTR_LEN(suffix);
    res = zend_string_alloc(len, 0);

    memcpy(ZSTR_VAL(res), ZSTR_VAL(prefix), ZSTR_LEN(prefix));
    memcpy(ZSTR_VAL(res) + ZSTR_LEN(prefix), ZSTR_VAL(suffix), ZSTR_LEN(suffix));
    ZSTR_VAL(res)[len] = '\0';

    return res;
}

PHP_REDIS_API zend_string *
redis_key_prefix_zstr(RedisSock *redis_sock, zend_string *key) {
    if (redis_sock->prefix == NULL)
        return zend_string_copy(key);

    return redis_zstr_concat(redis_sock->prefix, key);
}

/*
 * Processing for variant reply types (think EVAL)
 */

PHP_REDIS_API int
redis_sock_gets(RedisSock *redis_sock, char *buf, int buf_size, size_t *line_size) {
    // Handle EOF
    if(-1 == redis_check_eof(redis_sock, 1, 0)) {
        return -1;
    }

    if(redis_sock_get_line(redis_sock, buf, buf_size, line_size) == NULL ||
       *line_size < 2 || memcmp(buf + *line_size - 2, ZEND_STRL("\r\n")) != 0)
    {
        if (redis_sock->port < 0) {
            snprintf(buf, buf_size, "read error on connection to %s", ZSTR_VAL(redis_sock->host));
        } else {
            snprintf(buf, buf_size, "read error on connection to %s:%d", ZSTR_VAL(redis_sock->host), redis_sock->port);
        }
        // Close our socket
        redis_sock_disconnect(redis_sock, 1, 1);

        // Throw a read error exception
        REDIS_THROW_EXCEPTION(buf, 0);
        return FAILURE;
    }

    /* We don't need \r\n */
    *line_size -= 2;
    buf[*line_size] = '\0';

    /* Success! */
    return 0;
}

PHP_REDIS_API int
redis_read_reply_type(RedisSock *redis_sock, REDIS_REPLY_TYPE *reply_type,
                      long *reply_info)
{
    size_t nread;

    // Make sure we haven't lost the connection, even trying to reconnect
    if(-1 == redis_check_eof(redis_sock, 1, 0)) {
        // Failure
        *reply_type = EOF;
        return -1;
    }

    // Attempt to read the reply-type byte
    if((*reply_type = redis_sock_getc(redis_sock)) == EOF) {
        REDIS_THROW_EXCEPTION( "socket error on read socket", 0);
        return -1;
    }

    // If this is a BULK, MULTI BULK, or simply an INTEGER response, we can
    // extract the value or size info here
    if(*reply_type == TYPE_INT || *reply_type == TYPE_BULK ||
       *reply_type == TYPE_MULTIBULK)
    {
        // Buffer to hold size information
        char inbuf[255];

        /* Read up to our newline */
        if (redis_sock_get_line(redis_sock, inbuf, sizeof(inbuf), &nread) == NULL) {
            return -1;
        }

        /* Set our size response */
        {
            char *endptr;

            errno = 0;
            *reply_info = strtol(inbuf, &endptr, 10);
            if (endptr == inbuf || errno == ERANGE) {
                return -1;
            }
        }
    } else {
        /* Always initialize to prevent UB */
        *reply_info = 0;
    }

    /* Success! */
    return 0;
}

/*
 * Read a single line response, having already consumed the reply-type byte
 */
static int
redis_read_variant_line(RedisSock *redis_sock, REDIS_REPLY_TYPE reply_type,
                        int as_string, zval *z_ret)
{
    // Buffer to read our single line reply
    char inbuf[4096];
    size_t len;

    /* Attempt to read our single line reply */
    if(redis_sock_gets(redis_sock, inbuf, sizeof(inbuf), &len) < 0) {
        return -1;
    }

    /* Throw exception on SYNC error otherwise just set error string */
    if(reply_type == TYPE_ERR) {
        redis_sock_set_err(redis_sock, inbuf, len);
        redis_error_throw(redis_sock);
        ZVAL_FALSE(z_ret);
    } else if (as_string) {
        ZVAL_STRINGL(z_ret, inbuf, len);
    } else {
        ZVAL_TRUE(z_ret);
    }

    return 0;
}

PHP_REDIS_API int
redis_read_variant_bulk(RedisSock *redis_sock, long size, zval *z_ret
                       )
{
    int bytes;

    if (size < -1 || size > INT_MAX - 2) {
        zend_throw_exception_ex(redis_exception_ce, 0,
            "protocol error, invalid bulk length: %ld", size);
        ZVAL_FALSE(z_ret);
        return -1;
    }

    bytes = size;

    // Attempt to read the bulk reply
    char *bulk_resp = redis_sock_read_bulk_reply(redis_sock, bytes);

    /* Set our reply to FALSE on failure, and the string on success */
    if(bulk_resp == NULL) {
        ZVAL_FALSE(z_ret);
        return -1;
    }
    ZVAL_STRINGL(z_ret, bulk_resp, bytes);
    efree(bulk_resp);
    return 0;
}

PHP_REDIS_API int
redis_read_multibulk_recursive(RedisSock *redis_sock, long long elements, int status_strings,
                               zval *z_ret)
{
    long reply_info;
    REDIS_REPLY_TYPE reply_type;
    zval z_subelem;

    // Iterate while we have elements
    while(elements > 0) {
        // Attempt to read our reply type
        if(redis_read_reply_type(redis_sock, &reply_type, &reply_info
                                ) < 0)
        {
            zend_throw_exception_ex(redis_exception_ce, 0,
                "protocol error, couldn't parse MULTI-BULK response\n");
            return FAILURE;
        }

        // Switch on our reply-type byte
        switch(reply_type) {
            case TYPE_ERR:
            case TYPE_LINE:
                redis_read_variant_line(redis_sock, reply_type, status_strings,
                                        &z_subelem);
                add_next_index_zval(z_ret, &z_subelem);
                break;
            case TYPE_INT:
                // Add our long value
                add_next_index_long(z_ret, reply_info);
                break;
            case TYPE_BULK:
                // Init a zval for our bulk response, read and add it
                redis_read_variant_bulk(redis_sock, reply_info, &z_subelem);
                add_next_index_zval(z_ret, &z_subelem);
                break;
            case TYPE_MULTIBULK:
                if (reply_info < 0 && redis_sock->null_mbulk_as_null) {
                    add_next_index_null(z_ret);
                } else {
                    array_init_size(&z_subelem, reply_info > 0 ? reply_info : 0);
                    if (reply_info > 0) {
                        redis_read_multibulk_recursive(redis_sock, reply_info, status_strings, &z_subelem);
                    }
                    add_next_index_zval(z_ret, &z_subelem);
                }
                break;
            default:
                // Stop the compiler from whinging
                break;
        }

        /* Decrement our element counter */
        elements--;
    }

    return SUCCESS;
}

static int
variant_reply_generic(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                      int status_strings, int null_mbulk_as_null,
                      zval *z_tab, RedisCmdCtx ctx)
{
    // Reply type, and reply size vars
    REDIS_REPLY_TYPE reply_type;
    long reply_info;
    zval z_ret;

    // Attempt to read our header
    if(redis_read_reply_type(redis_sock,&reply_type,&reply_info) < 0) {
        return -1;
    }

    /* Switch based on our top level reply type */
    switch(reply_type) {
        case TYPE_ERR:
        case TYPE_LINE:
            redis_read_variant_line(redis_sock, reply_type, status_strings, &z_ret);
            break;
        case TYPE_INT:
            ZVAL_LONG(&z_ret, reply_info);
            break;
        case TYPE_BULK:
            redis_read_variant_bulk(redis_sock, reply_info, &z_ret);
            break;
        case TYPE_MULTIBULK:
            if (reply_info > -1) {
                array_init_size(&z_ret, reply_info);
                redis_read_multibulk_recursive(redis_sock, reply_info, status_strings, &z_ret);
            } else {
                if (null_mbulk_as_null) {
                    ZVAL_NULL(&z_ret);
                } else {
                    ZVAL_EMPTY_ARRAY(&z_ret);
                }
            }
            break;
        default:
            zend_throw_exception_ex(redis_exception_ce, 0,
                "protocol error, got '%c' as reply-type byte\n", reply_type);
            return FAILURE;
    }

    REDIS_RETURN_ZVAL(redis_sock, z_tab, z_ret);

    /* Success */
    return 0;
}

PHP_REDIS_API int
redis_read_raw_variant_reply(INTERNAL_FUNCTION_PARAMETERS,
                             RedisSock *redis_sock, zval *z_tab,
                             RedisCmdCtx ctx)
{
    return variant_reply_generic(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                 redis_sock->reply_literal,
                                 redis_sock->null_mbulk_as_null,
                                 z_tab, ctx);
}

PHP_REDIS_API int
redis_read_variant_reply(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                         zval *z_tab, RedisCmdCtx ctx)
{
    return variant_reply_generic(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock, 0,
                                 redis_sock->null_mbulk_as_null, z_tab, ctx);
}

PHP_REDIS_API int
redis_read_variant_reply_strings(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock,
                                 zval *z_tab, RedisCmdCtx ctx)
{
    return variant_reply_generic(INTERNAL_FUNCTION_PARAM_PASSTHRU, redis_sock,
                                 1, 0, z_tab, ctx);
}

/* The user may wish to send us something like [NULL, 'password'] or
 * [false, 'password'] so don't convert NULL or FALSE into "". */
static int redisTrySetAuthArg(zend_string **dst, zval *zsrc) {
    if (Z_TYPE_P(zsrc) == IS_NULL || Z_TYPE_P(zsrc) == IS_FALSE)
        return FAILURE;

    *dst = zval_get_string(zsrc);

    return SUCCESS;
}

PHP_REDIS_API
int redis_extract_auth_info(zval *ztest, zend_string **user, zend_string **pass) {
    zval *zv;
    HashTable *ht;
    int num;

    /* Null out user and password */
    *user = *pass = NULL;

    /* User passed nothing */
    if (ztest == NULL)
        return FAILURE;

    /* Handle a non-array first */
    if (Z_TYPE_P(ztest) != IS_ARRAY) {
        return redisTrySetAuthArg(pass, ztest);
    }

    /* Handle the array case */
    ht = Z_ARRVAL_P(ztest);
    num = zend_hash_num_elements(ht);

    /* Something other than one or two entries makes no sense */
    if (num != 1 && num != 2) {
        php_error_docref(NULL, E_WARNING, "When passing an array as auth it must have one or two elements!");
        return FAILURE;
    }

    if (num == 2) {
        if ((zv = REDIS_HASH_STR_FIND_STATIC(ht, "user")) ||
            (zv = zend_hash_index_find(ht, 0)))
        {
            redisTrySetAuthArg(user, zv);
        }

        if ((zv = REDIS_HASH_STR_FIND_STATIC(ht, "pass")) ||
            (zv = zend_hash_index_find(ht, 1)))
        {
            redisTrySetAuthArg(pass, zv);
        }
    } else if ((zv = REDIS_HASH_STR_FIND_STATIC(ht, "pass")) ||
               (zv = zend_hash_index_find(ht, 0)))
    {
        redisTrySetAuthArg(pass, zv);
    }

    /* If we at least have a password, we're good */
    if (*pass != NULL)
        return SUCCESS;

    /* Failure, clean everything up so caller doesn't need to care */
    if (*user) zend_string_release(*user);
    *user = NULL;

    return FAILURE;
}

PHP_REDIS_API void redis_with_metadata(zval *zdst, zval *zsrc, zend_long length) {
    zval z_sub;

    array_init_size(zdst, 2);
    add_next_index_zval(zdst, zsrc);

    array_init_size(&z_sub, 1);
    add_assoc_long_ex(&z_sub, ZEND_STRL("length"), length);
    add_next_index_zval(zdst, &z_sub);
}

/* Helper methods to extract configuration settings from a hash table */

zval *redis_hash_str_find_type(HashTable *ht, const char *key, int keylen, int type) {
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL || Z_TYPE_P(zv) != type)
        return NULL;

    return zv;
}

void redis_conf_double(HashTable *ht, const char *key, int keylen, double *dval) {
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL)
        return;

    *dval = zval_get_double(zv);
}

void redis_conf_bool(HashTable *ht, const char *key, int keylen, int *ival) {
    zend_string *zstr = NULL;

    redis_conf_string(ht, key, keylen, &zstr);
    if (zstr == NULL)
        return;

    *ival = zend_string_equals_literal_ci(zstr, "true") ||
            zend_string_equals_literal_ci(zstr, "yes") ||
            zend_string_equals_literal_ci(zstr, "1");

    zend_string_release(zstr);
}

void redis_conf_zend_bool(HashTable *ht, const char *key, int keylen, zend_bool *bval) {
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL)
        return;

    *bval = zend_is_true(zv);
}

void redis_conf_long(HashTable *ht, const char *key, int keylen, zend_long *lval) {
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL)
        return;

    *lval = zval_get_long(zv);
}

void redis_conf_int(HashTable *ht, const char *key, int keylen, int *ival) {
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL)
        return;

    *ival = zval_get_long(zv);
}

void redis_conf_string(HashTable *ht, const char *key, size_t keylen,
                       zend_string **sval)
{
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL)
        return;

    *sval = zval_get_string(zv);
}

void redis_conf_zval(HashTable *ht, const char *key, size_t keylen, zval *zret,
                     int copy, int dtor)
{
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL)
        return;

    ZVAL_ZVAL(zret, zv, copy, dtor);
}

void redis_conf_auth(HashTable *ht, const char *key, size_t keylen,
                     zend_string **user, zend_string **pass)
{
    zval *zv = zend_hash_str_find(ht, key, keylen);
    if (zv == NULL)
        return;

    redis_extract_auth_info(zv, user, pass);
}

/* Update a zval with the current 64 bit scan cursor.  This presents a problem
 * because we can only represent up to 63 bits in a PHP integer.  So depending
 * on the cursor value, we may need to represent it as a string. */
void redisSetScanCursor(zval *zv, uint64_t cursor) {
    char tmp[21];
    size_t len;

    ZEND_ASSERT(zv != NULL && (Z_TYPE_P(zv) == IS_LONG ||
                               Z_TYPE_P(zv) == IS_STRING));

    if (Z_TYPE_P(zv) == IS_STRING)
        zend_string_release(Z_STR_P(zv));

    if (cursor > ZEND_LONG_MAX) {
        len = snprintf(tmp, sizeof(tmp), "%llu", (unsigned long long)cursor);
        ZVAL_STRINGL(zv, tmp, len);
    } else {
        ZVAL_LONG(zv, cursor);
    }
}

/* Get a Redis SCAN cursor value out of a zval.  These are always taken as a
 * reference argument that that must be `null`, `int`, or `string`.  */
uint64_t redisGetScanCursor(zval *zv, zend_bool *was_zero) {
    ZEND_ASSERT(zv != NULL && (Z_TYPE_P(zv) == IS_LONG ||
                               Z_TYPE_P(zv) == IS_STRING ||
                               Z_TYPE_P(zv) == IS_NULL));

    if (Z_TYPE_P(zv) == IS_NULL) {
        convert_to_long(zv);
        *was_zero = 0;
        return 0;
    } else if (Z_TYPE_P(zv) == IS_STRING) {
        *was_zero = Z_STRLEN_P(zv) == 1 && Z_STRVAL_P(zv)[0] == '0';
        return strtoull(Z_STRVAL_P(zv), NULL, 10);
    } else {
        *was_zero = Z_LVAL_P(zv) == 0;
        return Z_LVAL_P(zv);
    }
}

/* vim: set tabstop=4 softtabstop=4 expandtab shiftwidth=4: */