: 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");

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);

}

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);

}

__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");

const reply_callback_t &callback) {

m_client.send(redis_cmd);

m_commands.push({redis_cmd, callback});

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;

}

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();

}

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();

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();

}

//!

//! 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;

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));

return !is_connected() && !m_cancel &&

(m_max_reconnects == -1 ||

m_current_reconnect_attempts < m_max_reconnects);

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());

}

}});

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());

}

});

//!

//! 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();

switch (method) {

case aggregate_method::sum:

return "SUM";

case aggregate_method::min:

return "MIN";

case aggregate_method::max:

return "MAX";

default:

return "";

}

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 "";

}

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 "";

}

switch (type) {

case overflow_type::wrap:

return "WRAP";

case overflow_type::sat:

return "SAT";

case overflow_type::fail:

return "FAIL";

default:

return "";

}

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);