// The MIT License (MIT)

//

// Copyright (c) 2015-2017 Simon Ninon <simon.ninon@gmail.com>

//

// Permission is hereby granted, free of charge, to any person obtaining a copy

// of this software and associated documentation files (the "Software"), to deal

// in the Software without restriction, including without limitation the rights

// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell

// copies of the Software, and to permit persons to whom the Software is

// furnished to do so, subject to the following conditions:

//

// The above copyright notice and this permission notice shall be included in

// all copies or substantial portions of the Software.

//

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR

// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,

// FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE

// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER

// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,

// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE

// SOFTWARE.

#include <cpp_redis/core/client.hpp>

#include <cpp_redis/misc/error.hpp>

#include <cpp_redis/misc/macro.hpp>

#include <cpp_redis/impl/reply.ipp>

namespace cpp_redis {

#ifndef __CPP_REDIS_USE_CUSTOM_TCP_CLIENT

client::client()

: m_reconnecting(false), m_cancel(false), m_callbacks_running(0) {

__CPP_REDIS_LOG(debug, "cpp_redis::client created");

}

#endif // __CPP_REDIS_USE_CUSTOM_TCP_CLIENT

client::client(const std::shared_ptr<network::tcp_client_iface> &tcp_client)

: m_client(tcp_client), m_sentinel(tcp_client), m_reconnecting(false),

m_cancel(false), m_callbacks_running(0) {

__CPP_REDIS_LOG(debug, "cpp_redis::client created");

}

client::~client() {

//!

//! ensure we stopped reconnection attempts

//!

if (!m_cancel) {

cancel_reconnect();

}

//!

//! If for some reason sentinel is connected then disconnect now.

//!

if (m_sentinel.is_connected()) {

m_sentinel.disconnect(true);

}

//!

//! disconnect underlying tcp socket

//!

if (m_client.is_connected()) {

m_client.disconnect(true);

}

__CPP_REDIS_LOG(debug, "cpp_redis::client destroyed");

}

void client::connect(const string_t &name,

const connect_callback_t &connect_callback,

uint_t timeout_ms, int_t max_reconnects,

uint_t reconnect_interval_ms) {

//!

//! Save for auto reconnects

//!

m_master_name = name;

//!

//! We rely on the sentinel to tell us which redis server is currently the

//! master.

//!

if (m_sentinel.get_master_addr_by_name(name, m_redis_server, m_redis_port,

true)) {

connect(m_redis_server, m_redis_port, connect_callback, timeout_ms,

max_reconnects, reconnect_interval_ms);

} else {

throw redis_error(

"cpp_redis::client::connect() could not find master for m_name " +

name);

}

}

void client::connect(const string_t &host, std::size_t port,

const connect_callback_t &connect_callback,

uint_t timeout_ms, int_t max_reconnects,

uint_t reconnect_interval_ms) {

__CPP_REDIS_LOG(debug, "cpp_redis::client attempts to connect");

//!

//! Save for auto reconnects

//!

m_redis_server = host;

m_redis_port = port;

m_connect_callback = connect_callback;

m_max_reconnects = max_reconnects;

m_reconnect_interval_ms = reconnect_interval_ms;

//!

//! notify start

//!

if (m_connect_callback) {

m_connect_callback(host, port, connect_state::start);

}

auto disconnection_handler = std::bind(

&client::connection_disconnection_handler, this, std::placeholders::_1);

auto receive_handler =

std::bind(&client::connection_receive_handler, this,

std::placeholders::_1, std::placeholders::_2);

m_client.connect(host, port, disconnection_handler, receive_handler,

timeout_ms);

__CPP_REDIS_LOG(info, "cpp_redis::client connected");

//!

//! notify end

//!

if (m_connect_callback) {

m_connect_callback(m_redis_server, m_redis_port, connect_state::ok);

}

}

void client::disconnect(bool wait_for_removal) {

__CPP_REDIS_LOG(debug, "cpp_redis::client attempts to disconnect");

//!

//! close connection

//!

m_client.disconnect(wait_for_removal);

//!

//! make sure we clear buffer of unsent commands

//!

clear_callbacks();

__CPP_REDIS_LOG(info, "cpp_redis::client disconnected");

}

bool client::is_connected() const { return m_client.is_connected(); }

void client::cancel_reconnect() { m_cancel = true; }

bool client::is_reconnecting() const { return m_reconnecting; }

void client::add_sentinel(const string_t &host, std::size_t port,

uint_t timeout_ms) {

m_sentinel.add_sentinel(host, port, timeout_ms);

}

const sentinel &client::get_sentinel() const { return m_sentinel; }

sentinel &client::get_sentinel() { return m_sentinel; }

void client::clear_sentinels() { m_sentinel.clear_sentinels(); }

client &client::send(const std::vector<string_t> &redis_cmd,

const reply_callback_t &callback) {

std::lock_guard<std::mutex> lock_callback(m_callbacks_mutex);

__CPP_REDIS_LOG(

info,

"cpp_redis::client attempts to store new command in the send buffer");

unprotected_send(redis_cmd, callback);

__CPP_REDIS_LOG(info,

"cpp_redis::client stored new command in the send buffer");

return *this;

}

void client::unprotected_send(const std::vector<string_t> &redis_cmd,

const reply_callback_t &callback) {

m_client.send(redis_cmd);

m_commands.push({redis_cmd, callback});

}

//!

//! commit pipelined transaction

//!

client &client::commit() {

//!

//! no need to call commit in case of reconnection

//! the reconnection flow will do it for us

//!

if (!is_reconnecting()) {

try_commit();

}

return *this;

}

client &client::sync_commit() {

//!

//! no need to call commit in case of reconnection

//! the reconnection flow will do it for us

//!

if (!is_reconnecting()) {

try_commit();

}

std::unique_lock<std::mutex> lock_callback(m_callbacks_mutex);

__CPP_REDIS_LOG(debug, "cpp_redis::client waiting for callbacks to complete");

m_sync_cond_var.wait(lock_callback, [=] {

return m_callbacks_running == 0 && m_commands.empty();

});

__CPP_REDIS_LOG(debug,

"cpp_redis::client finished waiting for callback completion");

return *this;

}

void client::try_commit() {

try {

__CPP_REDIS_LOG(debug,

"cpp_redis::client attempts to send pipelined commands");

m_client.commit();

__CPP_REDIS_LOG(info, "cpp_redis::client sent pipelined commands");

} catch (const cpp_redis::redis_error &) {

__CPP_REDIS_LOG(error,

"cpp_redis::client could not send pipelined commands");

//!

//! ensure commands are flushed

//!

clear_callbacks();

throw;

}

}

void client::connection_receive_handler(network::redis_connection &,

reply &reply) {

reply_callback_t callback = nullptr;

__CPP_REDIS_LOG(info, "cpp_redis::client received reply");

{

std::lock_guard<std::mutex> lock(m_callbacks_mutex);

m_callbacks_running += 1;

if (!m_commands.empty()) {

callback = m_commands.front().callback;

m_commands.pop();

}

}

if (callback) {

__CPP_REDIS_LOG(debug, "cpp_redis::client executes reply callback");

callback(reply);

}

{

std::lock_guard<std::mutex> lock(m_callbacks_mutex);

m_callbacks_running -= 1;

m_sync_cond_var.notify_all();

}

}

void client::clear_callbacks() {

if (m_commands.empty()) {

return;

}

//!

//! de-queue commands and move them to a local variable

//!

std::queue<command_request> commands = std::move(m_commands);

m_callbacks_running += __CPP_REDIS_LENGTH(commands.size());

std::thread t([=]() mutable {

while (!commands.empty()) {

const auto &callback = commands.front().callback;

if (callback) {

reply r = {"network failure", reply::string_type::error};

callback(r);

}

--m_callbacks_running;

commands.pop();

}

m_sync_cond_var.notify_all();

});

t.detach();

}

void client::resend_failed_commands() {

if (m_commands.empty()) {

return;

}

//!

//! de-queue commands and move them to a local variable

//!

std::queue<command_request> commands = std::move(m_commands);

while (!commands.empty()) {

//!

//! Reissue the pending command and its callback.

//!

unprotected_send(commands.front().command, commands.front().callback);

commands.pop();

}

}

void client::connection_disconnection_handler(network::redis_connection &) {

//!

//! leave right now if we are already dealing with reconnection

//!

if (is_reconnecting()) {

return;

}

//!

//! initiate reconnection process

//!

m_reconnecting = true;

m_current_reconnect_attempts = 0;

__CPP_REDIS_LOG(warn, "cpp_redis::client has been disconnected");

if (m_connect_callback) {

m_connect_callback(m_redis_server, m_redis_port, connect_state::dropped);

}

//!

//! Lock the callbacks mutex of the base class to prevent more client

//! commands from being issued until our reconnect has completed.

//!

std::lock_guard<std::mutex> lock_callback(m_callbacks_mutex);

while (should_reconnect()) {

sleep_before_next_reconnect_attempt();

reconnect();

}

if (!is_connected()) {

clear_callbacks();

//!

//! Tell the user we gave up!

//!

if (m_connect_callback) {

m_connect_callback(m_redis_server, m_redis_port, connect_state::stopped);

}

}

//!

//! terminate reconnection

//!

m_reconnecting = false;

}

void client::sleep_before_next_reconnect_attempt() {

if (m_reconnect_interval_ms <= 0) {

return;

}

if (m_connect_callback) {

m_connect_callback(m_redis_server, m_redis_port, connect_state::sleeping);

}

std::this_thread::sleep_for(

std::chrono::milliseconds(m_reconnect_interval_ms));

}

bool client::should_reconnect() const {

return !is_connected() && !m_cancel &&

(m_max_reconnects == -1 ||

m_current_reconnect_attempts < m_max_reconnects);

}

void client::re_auth() {

if (m_password.empty()) {

return;

}

// In re_auth we do not want to serialize the command to the TCP buffer (i.e.

// call m_client.send) as it will get serialized to the buffer in

// resend_failed_commands after re_auth is called in reconnect. So we just

// push the command into the command queue.

//

// Otherwise we will end up with the AUTH command serialized to

// the buffer twice and in the command queue once, which causes a problem when

// we get 2 responses back. The subsequent responses are off by one in the

// command callbacks.

//

m_commands.push(

{{"AUTH", m_password}, [&](cpp_redis::reply_t &reply) {

if (reply.is_string() && reply.as_string() == "OK") {

__CPP_REDIS_LOG(warn, "client successfully re-authenticated");

} else {

__CPP_REDIS_LOG(warn, string_t("client failed to re-authenticate: " +

reply.as_string())

.c_str());

}

}});

}

void client::re_select() {

if (m_database_index <= 0) {

return;

}

unprotected_select(m_database_index, [&](cpp_redis::reply_t &reply) {

if (reply.is_string() && reply.as_string() == "OK") {

__CPP_REDIS_LOG(warn, "client successfully re-selected redis database");

} else {

__CPP_REDIS_LOG(warn, string_t("client failed to re-select database: " +

reply.as_string())

.c_str());

}

});

}

void client::reconnect() {

//!

//! increase the number of attempts to reconnect

//!

++m_current_reconnect_attempts;

//!

//! We rely on the sentinel to tell us which redis server is currently the

//! master.

//!

if (!m_master_name.empty() &&

!m_sentinel.get_master_addr_by_name(m_master_name, m_redis_server,

m_redis_port, true)) {

if (m_connect_callback) {

m_connect_callback(m_redis_server, m_redis_port,

connect_state::lookup_failed);

}

return;

}

//!

//! Try catch block because the redis client throws an error if connection

//! cannot be made.

//!

try {

connect(m_redis_server, m_redis_port, m_connect_callback,

m_connect_timeout_ms, m_max_reconnects, m_reconnect_interval_ms);

} catch (...) {

}

if (!is_connected()) {

if (m_connect_callback) {

m_connect_callback(m_redis_server, m_redis_port, connect_state::failed);

}

return;

}

//!

//! notify end

//!

if (m_connect_callback) {

m_connect_callback(m_redis_server, m_redis_port, connect_state::ok);

}

__CPP_REDIS_LOG(info, "client reconnected ok");

re_auth();

re_select();

resend_failed_commands();

try_commit();

}

string_t client::aggregate_method_to_string(aggregate_method method) const {

switch (method) {

case aggregate_method::sum:

return "SUM";

case aggregate_method::min:

return "MIN";

case aggregate_method::max:

return "MAX";

default:

return "";

}

}

string_t client::geo_unit_to_string(geo_unit unit) const {

switch (unit) {

case geo_unit::m:

return "m";

case geo_unit::km:

return "km";

case geo_unit::ft:

return "ft";

case geo_unit::mi:

return "mi";

default:

return "";

}

}

string_t client::bitfield_operation_type_to_string(

bitfield_operation_type operation) const {

switch (operation) {

case bitfield_operation_type::get:

return "GET";

case bitfield_operation_type::set:

return "SET";

case bitfield_operation_type::incrby:

return "INCRBY";

default:

return "";

}

}

string_t client::overflow_type_to_string(overflow_type type) const {

switch (type) {

case overflow_type::wrap:

return "WRAP";

case overflow_type::sat:

return "SAT";

case overflow_type::fail:

return "FAIL";

default:

return "";

}

}

client::bitfield_operation

client::bitfield_operation::get(const string_t &type, int offset,

overflow_type overflow) {

return {bitfield_operation_type::get, type, offset, 0, overflow};

}

client::bitfield_operation

client::bitfield_operation::set(const string_t &type, int offset, int value,

overflow_type overflow) {

return {bitfield_operation_type::set, type, offset, value, overflow};

}

client::bitfield_operation

client::bitfield_operation::incrby(const string_t &type, int offset,

int increment, overflow_type overflow) {

return {bitfield_operation_type::incrby, type, offset, increment, overflow};

}

//!

//! Redis commands

//! Callback-based

//!

client &client::append(const string_t &key, const string_t &value,

const reply_callback_t &reply_callback) {

send({"APPEND", key, value}, reply_callback);

return *this;

}

client &client::auth(const string_t &password,

const reply_callback_t &reply_callback) {

std::lock_guard<std::mutex> lock(m_callbacks_mutex);

unprotected_auth(password, reply_callback);

return *this;

}

void client::unprotected_auth(const string_t &password,

const reply_callback_t &reply_callback) {

//!

//! save the password for reconnect attempts.

//!

m_password = password;

//!

//! store command in pipeline

//!

unprotected_send({"AUTH", password}, reply_callback);

}

client &client::bgrewriteaof(const reply_callback_t &reply_callback) {

send({"BGREWRITEAOF"}, reply_callback);

return *this;

}

client &client::bgsave(const reply_callback_t &reply_callback) {

send({"BGSAVE"}, reply_callback);

return *this;

}

client &client::bitcount(const string_t &key,

const reply_callback_t &reply_callback) {

send({"BITCOUNT", key}, reply_callback);

return *this;

}

client &client::bitcount(const string_t &key, int start, int end,

const reply_callback_t &reply_callback) {

send({"BITCOUNT", key, std::to_string(start), std::to_string(end)},

reply_callback);

return *this;

}

client &client::bitfield(const string_t &key,

const std::vector<bitfield_operation> &operations,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"BITFIELD", key};

for (const auto &operation : operations) {

cmd.push_back(bitfield_operation_type_to_string(operation.operation_type));

cmd.push_back(operation.type);

cmd.push_back(std::to_string(operation.offset));

if (operation.operation_type == bitfield_operation_type::set ||

operation.operation_type == bitfield_operation_type::incrby) {

cmd.push_back(std::to_string(operation.value));

}

if (operation.overflow != overflow_type::server_default) {

cmd.emplace_back("OVERFLOW");

cmd.push_back(overflow_type_to_string(operation.overflow));

}

}

send(cmd, reply_callback);

return *this;

}

client &client::bitop(const string_t &operation, const string_t &destkey,

const std::vector<string_t> &keys,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"BITOP", operation, destkey};

cmd.insert(cmd.end(), keys.begin(), keys.end());

send(cmd, reply_callback);

return *this;

}

client &client::bitpos(const string_t &key, int bit,

const reply_callback_t &reply_callback) {

send({"BITPOS", key, std::to_string(bit)}, reply_callback);

return *this;

}

client &client::bitpos(const string_t &key, int bit, int start,

const reply_callback_t &reply_callback) {

send({"BITPOS", key, std::to_string(bit), std::to_string(start)},

reply_callback);

return *this;

}

client &client::bitpos(const string_t &key, int bit, int start, int end,

const reply_callback_t &reply_callback) {

send({"BITPOS", key, std::to_string(bit), std::to_string(start),

std::to_string(end)},

reply_callback);

return *this;

}

client &client::blpop(const std::vector<string_t> &keys, int timeout,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"BLPOP"};

cmd.insert(cmd.end(), keys.begin(), keys.end());

cmd.push_back(std::to_string(timeout));

send(cmd, reply_callback);

return *this;

}

client &client::brpop(const std::vector<string_t> &keys, int timeout,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"BRPOP"};

cmd.insert(cmd.end(), keys.begin(), keys.end());

cmd.push_back(std::to_string(timeout));

send(cmd, reply_callback);

return *this;

}

client &client::brpoplpush(const string_t &src, const string_t &dst,

int timeout,

const reply_callback_t &reply_callback) {

send({"BRPOPLPUSH", src, dst, std::to_string(timeout)}, reply_callback);

return *this;

}

client &client::bzpopmin(const std::vector<string_t> &keys, int timeout,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"BZPOPMIN"};

cmd.insert(cmd.end(), keys.begin(), keys.end());

cmd.push_back(std::to_string(timeout));

send(cmd, reply_callback);

return *this;

}

client &client::bzpopmax(const std::vector<string_t> &keys, int timeout,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"BZPOPMAX"};

cmd.insert(cmd.end(), keys.begin(), keys.end());

cmd.push_back(std::to_string(timeout));

send(cmd, reply_callback);

return *this;

}

client &client::client_id(const reply_callback_t &reply_callback) {

send({"CLIENT", "ID"}, reply_callback);

return *this;

}

client &client::client_list(const reply_callback_t &reply_callback) {

send({"CLIENT", "LIST"}, reply_callback);

return *this;

}

client &

client::client_list_test(const client_list_reply_callback_t &reply_callback) {

send({"CLIENT", "GETNAME"}, [&](const reply_t &repl) {

client_list_payload_t clp(repl);

reply_callback(clp);

});

return *this;

}

client &client::client_getname(const reply_callback_t &reply_callback) {

send({"CLIENT", "GETNAME"}, reply_callback);

return *this;

}

client &client::client_pause(int timeout,

const reply_callback_t &reply_callback) {

send({"CLIENT", "PAUSE", std::to_string(timeout)}, reply_callback);

return *this;

}

client &client::client_reply(const string_t &mode,

const reply_callback_t &reply_callback) {

send({"CLIENT", "REPLY", mode}, reply_callback);

return *this;

}

client &client::client_setname(const string_t &name,

const reply_callback_t &reply_callback) {

send({"CLIENT", "SETNAME", name}, reply_callback);

return *this;

}

client &client::client_unblock(int id, const reply_callback_t &reply_callback) {

send({"CLIENT", "UNBLOCK", std::to_string(id)}, reply_callback);

return *this;

}

client &client::client_unblock(int id, bool witherror,

const reply_callback_t &reply_callback) {

if (witherror)

send({"CLIENT", "UNBLOCK", std::to_string(id), "ERROR"}, reply_callback);

else

send({"CLIENT", "UNBLOCK", std::to_string(id)}, reply_callback);

return *this;

}

client &client::cluster_addslots(const std::vector<string_t> &p_slots,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"CLUSTER", "ADDSLOTS"};

cmd.insert(cmd.end(), p_slots.begin(), p_slots.end());

send(cmd, reply_callback);

return *this;

}

client &

client::cluster_count_failure_reports(const string_t &node_id,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "COUNT-FAILURE-REPORTS", node_id}, reply_callback);

return *this;

}

client &

client::cluster_countkeysinslot(const string_t &slot,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "COUNTKEYSINSLOT", slot}, reply_callback);

return *this;

}

client &client::cluster_delslots(const std::vector<string_t> &p_slots,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"CLUSTER", "DELSLOTS"};

cmd.insert(cmd.end(), p_slots.begin(), p_slots.end());

send(cmd, reply_callback);

return *this;

}

client &client::cluster_failover(const reply_callback_t &reply_callback) {

send({"CLUSTER", "FAILOVER"}, reply_callback);

return *this;

}

client &client::cluster_failover(const string_t &mode,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "FAILOVER", mode}, reply_callback);

return *this;

}

client &client::cluster_forget(const string_t &node_id,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "FORGET", node_id}, reply_callback);

return *this;

}

client &client::cluster_getkeysinslot(const string_t &slot, int count,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "GETKEYSINSLOT", slot, std::to_string(count)},

reply_callback);

return *this;

}

client &client::cluster_info(const reply_callback_t &reply_callback) {

send({"CLUSTER", "INFO"}, reply_callback);

return *this;

}

client &client::cluster_keyslot(const string_t &key,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "KEYSLOT", key}, reply_callback);

return *this;

}

client &client::cluster_meet(const string_t &ip, int port,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "MEET", ip, std::to_string(port)}, reply_callback);

return *this;

}

client &client::cluster_nodes(const reply_callback_t &reply_callback) {

send({"CLUSTER", "NODES"}, reply_callback);

return *this;

}

client &client::cluster_replicate(const string_t &node_id,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "REPLICATE", node_id}, reply_callback);

return *this;

}

client &client::cluster_reset(const reply_callback_t &reply_callback) {

send({"CLUSTER", "RESET"}, reply_callback);

return *this;

}

client &client::cluster_reset(const string_t &mode,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "RESET", mode}, reply_callback);

return *this;

}

client &client::cluster_saveconfig(const reply_callback_t &reply_callback) {

send({"CLUSTER", "SAVECONFIG"}, reply_callback);

return *this;

}

client &

client::cluster_set_config_epoch(const string_t &epoch,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "SET-CONFIG-EPOCH", epoch}, reply_callback);

return *this;

}

client &client::cluster_setslot(const string_t &slot, const string_t &mode,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "SETSLOT", slot, mode}, reply_callback);

return *this;

}

client &client::cluster_setslot(const string_t &slot, const string_t &mode,

const string_t &node_id,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "SETSLOT", slot, mode, node_id}, reply_callback);

return *this;

}

client &client::cluster_slaves(const string_t &node_id,

const reply_callback_t &reply_callback) {

send({"CLUSTER", "SLAVES", node_id}, reply_callback);

return *this;

}

client &client::cluster_slots(const reply_callback_t &reply_callback) {

send({"CLUSTER", "SLOTS"}, reply_callback);

return *this;

}

client &client::command(const reply_callback_t &reply_callback) {

send({"COMMAND"}, reply_callback);

return *this;

}

client &client::command_count(const reply_callback_t &reply_callback) {

send({"COMMAND", "COUNT"}, reply_callback);

return *this;

}

client &client::command_getkeys(const reply_callback_t &reply_callback) {

send({"COMMAND", "GETKEYS"}, reply_callback);

return *this;

}

client &client::command_info(const std::vector<string_t> &command_name,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"COMMAND", "COUNT"};

cmd.insert(cmd.end(), command_name.begin(), command_name.end());

send(cmd, reply_callback);

return *this;

}

client &client::config_get(const string_t &param,

const reply_callback_t &reply_callback) {

send({"CONFIG", "GET", param}, reply_callback);

return *this;

}

client &client::config_rewrite(const reply_callback_t &reply_callback) {

send({"CONFIG", "REWRITE"}, reply_callback);

return *this;

}

client &client::config_set(const string_t &param, const string_t &val,

const reply_callback_t &reply_callback) {

send({"CONFIG", "SET", param, val}, reply_callback);

return *this;

}

client &client::config_resetstat(const reply_callback_t &reply_callback) {

send({"CONFIG", "RESETSTAT"}, reply_callback);

return *this;

}

client &client::dbsize(const reply_callback_t &reply_callback) {

send({"DBSIZE"}, reply_callback);

return *this;

}

client &client::debug_object(const string_t &key,

const reply_callback_t &reply_callback) {

send({"DEBUG", "OBJECT", key}, reply_callback);

return *this;

}

client &client::debug_segfault(const reply_callback_t &reply_callback) {

send({"DEBUG", "SEGFAULT"}, reply_callback);

return *this;

}

client &client::decr(const string_t &key,

const reply_callback_t &reply_callback) {

send({"DECR", key}, reply_callback);

return *this;

}

client &client::decrby(const string_t &key, int val,

const reply_callback_t &reply_callback) {

send({"DECRBY", key, std::to_string(val)}, reply_callback);

return *this;

}

client &client::del(const std::vector<string_t> &key,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"DEL"};

cmd.insert(cmd.end(), key.begin(), key.end());

send(cmd, reply_callback);

return *this;

}

client &client::discard(const reply_callback_t &reply_callback) {

send({"DISCARD"}, reply_callback);

return *this;

}

client &client::dump(const string_t &key,

const reply_callback_t &reply_callback) {

send({"DUMP", key}, reply_callback);

return *this;

}

client &client::echo(const string_t &msg,

const reply_callback_t &reply_callback) {

send({"ECHO", msg}, reply_callback);

return *this;

}

client &client::eval(const string_t &script, int numkeys,

const std::vector<string_t> &keys,

const std::vector<string_t> &args,

const reply_callback_t &reply_callback) {

std::vector<string_t> cmd = {"EVAL", script, std::to_string(numkeys)};

cmd.insert(cmd.end(), keys.begin(), keys.end());

cmd.insert(cmd.end(), args.begin(), args.end());

send(cmd, reply_callback);

return *this;

}

client &client::eval(const string_t &script, const std::vector<string_t> &keys,

const std::vector<string_t> &args,

const reply_callback_t &reply_callback) {