// 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 NONINFRINGEMENT. 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.

#pragma once

#include <atomic>

#include <condition_variable>

#include <queue>

#include <utility>

#include <vector>

#include <cpp_redis/misc/logger.hpp>

#include <cpp_redis/network/redis_connection.hpp>

namespace cpp_redis {

/**

* cpp_redis::sentinel is the class providing sentinel configuration.

* It is meant to be used for sending sentinel-related commands to the remote server and receiving its replies.

* It is also meant to be used with cpp_redis::client and cpp_redis::subscriber for high availability (automatic failover if reconnection is enabled).

*

*/

class sentinel {

public:

/**

* ctor & dtor

*

*/

#ifndef __CPP_REDIS_USE_CUSTOM_TCP_CLIENT

/**

* default ctor

*

*/

sentinel();

#endif /* __CPP_REDIS_USE_CUSTOM_TCP_CLIENT */

/**

* custom ctor to specify custom tcp_client

*

* @param tcp_client tcp client to be used for network communications

*

*/

explicit sentinel(const std::shared_ptr<network::tcp_client_iface> &tcp_client);

/**

* dtor

*

*/

~sentinel();

/**

* copy ctor

*

*/

sentinel(const sentinel &) = delete;

/**

* assignment operator

*

*/

sentinel &operator=(const sentinel &) = delete;

public:

/**

* callback to be called whenever a reply has been received

*

*/

typedef std::function<void(reply &)> reply_callback_t;

/**

* send the given command

* the command is actually pipelined and only buffered, so nothing is sent to the network

* please call commit() to flush the buffer

*

* @param sentinel_cmd command to be sent

* @param callback callback to be called when reply is received for this command

* @return current instance

*

*/

sentinel &send(const std::vector<std::string> &sentinel_cmd, const reply_callback_t &callback = nullptr);

/**

* commit pipelined transaction

* that is, send to the network all commands pipelined by calling send()

*

* @return current instance

*

*/

sentinel &commit();

/**

* same as commit(), but synchronous

* will block until all pending commands have been sent and that a reply has been received for each of them and all underlying callbacks completed

*

* @return current instance

*

*/

sentinel &sync_commit();

/**

* same as sync_commit, but with a timeout

* will simply block until it completes or timeout expires

*

* @return current instance

*

*/

template<class Rep, class Period>

sentinel &

sync_commit(const std::chrono::duration<Rep, Period> &timeout) {

try_commit();

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

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

if (!m_sync_condvar.wait_for(lock_callback, timeout, [=] {

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

})) {

__CPP_REDIS_LOG(debug, "cpp_redis::sentinel finished waiting for callback");

} else {

__CPP_REDIS_LOG(debug, "cpp_redis::sentinel timed out waiting for callback");

}

return *this;

}

public:

/**

* add a sentinel definition. Required for connect() or get_master_addr_by_name() when autoconnect is enabled.

*

* @param host sentinel host

* @param port sentinel port

* @param timeout_ms maximum time to connect

* @return current instance

*

*/

sentinel &add_sentinel(const std::string &host, std::size_t port, std::uint32_t timeout_ms = 0);

/**

* clear all existing sentinels.

*

*/

void clear_sentinels();

public:

/**

* disconnect from redis server

*

* @param wait_for_removal when sets to true, disconnect blocks until the underlying TCP client has been effectively removed from the io_service and that all the underlying callbacks have completed.

*

*/

void disconnect(bool wait_for_removal = false);

/**

* @return whether we are connected to the redis server or not

*

*/

bool is_connected();

/**

* handlers called whenever disconnection occurred

* function takes the sentinel current instance as parameter

*

*/

typedef std::function<void(sentinel &)> sentinel_disconnect_handler_t;

/**

* Connect to 1st active sentinel we find. Requires add_sentinel() to be called first

* will use timeout set for each added sentinel independently

*

* @param disconnect_handler handler to be called whenever disconnection occurs

*

*/

void connect_sentinel(const sentinel_disconnect_handler_t &disconnect_handler = nullptr);

/**

* Connect to named sentinel

*

* @param host host to be connected to

* @param port port to be connected to

* @param timeout_ms maximum time to connect

* @param disconnect_handler handler to be called whenever disconnection occurs

*

*/

void connect(

const std::string &host,

std::size_t port,

const sentinel_disconnect_handler_t &disconnect_handler = nullptr,

std::uint32_t timeout_ms = 0);

/**

* Used to find the current redis master by asking one or more sentinels. Use high availability.

* Handles connect() and disconnect() automatically when autoconnect=true

* This method is synchronous. No need to call sync_commit() or process a reply callback.

* Call add_sentinel() before using when autoconnect==true

*

* @param name sentinel name

* @param host sentinel host

* @param port sentinel port

* @param autoconnect autoconnect we loop through and connect/disconnect as necessary to sentinels that were added using add_sentinel().

* Otherwise we rely on the call to connect to a sentinel before calling this method.

* @return true if a master was found and fills in host and port output parameters, false otherwise

*/

bool get_master_addr_by_name(

const std::string &name,

std::string &host,

std::size_t &port,

bool autoconnect = true);

public:

sentinel &ckquorum(const std::string &name, const reply_callback_t &reply_callback = nullptr);

sentinel &failover(const std::string &name, const reply_callback_t &reply_callback = nullptr);

sentinel &flushconfig(const reply_callback_t &reply_callback = nullptr);

sentinel &master(const std::string &name, const reply_callback_t &reply_callback = nullptr);

sentinel &masters(const reply_callback_t &reply_callback = nullptr);

sentinel &monitor(const std::string &name, const std::string &ip, std::size_t port, std::size_t quorum,

const reply_callback_t &reply_callback = nullptr);

sentinel &ping(const reply_callback_t &reply_callback = nullptr);

sentinel &remove(const std::string &name, const reply_callback_t &reply_callback = nullptr);

sentinel &reset(const std::string &pattern, const reply_callback_t &reply_callback = nullptr);

sentinel &sentinels(const std::string &name, const reply_callback_t &reply_callback = nullptr);

sentinel &set(const std::string &name, const std::string &option, const std::string &value,

const reply_callback_t &reply_callback = nullptr);

sentinel &slaves(const std::string &name, const reply_callback_t &reply_callback = nullptr);

public:

/**

* store informations related to a sentinel

* typically, host, port and connection timeout

*

*/

class sentinel_def {

public:

/**

* ctor

*

*/

sentinel_def(std::string host, std::size_t port, std::uint32_t timeout_ms)

: m_host(std::move(host)), m_port(port), m_timeout_ms(timeout_ms) {}

/**

* dtor

*

*/

~sentinel_def() = default;

public:

/**

* @return sentinel host

*

*/

const std::string &

get_host() const { return m_host; }

/**

* @return sentinel port

*

*/

size_t

get_port() const { return m_port; }

/**

* @return timeout for sentinel

*

*/

std::uint32_t

get_timeout_ms() const { return m_timeout_ms; }

/**

* set connect timeout for sentinel in ms

* @param timeout_ms new value

*

*/

void

set_timeout_ms(std::uint32_t timeout_ms) { m_timeout_ms = timeout_ms; }

private:

/**

* sentinel host

*

*/

std::string m_host;

/**

* sentinel port

*

*/

std::size_t m_port;

/**

* connect timeout config

*

*/

std::uint32_t m_timeout_ms;

};

public:

/**

* @return sentinels

*

*/

const std::vector<sentinel_def> &get_sentinels() const;

/**

* @return sentinels (non-const version)

*

*/

std::vector<sentinel_def> &get_sentinels();

private:

/**

* redis connection receive handler, triggered whenever a reply has been read by the redis connection

*

* @param connection redis_connection instance

* @param reply parsed reply

*

*/

void connection_receive_handler(network::redis_connection &connection, reply &reply);

/**

* redis_connection disconnection handler, triggered whenever a disconnection occurred

*

* @param connection redis_connection instance

*

*/

void connection_disconnect_handler(network::redis_connection &connection);

/**

* Call the user-defined disconnection handler

*

*/

void call_disconnect_handler();

/**

* reset the queue of pending callbacks

*

*/

void clear_callbacks();

/**

* try to commit the pending pipelined

* if client is disconnected, will throw an exception and clear all pending callbacks (call clear_callbacks())

*

*/

void try_commit();

private:

/**

* A pool of 1 or more sentinels we ask to determine which redis server is the master.

*

*/

std::vector<sentinel_def> m_sentinels;

/**

* tcp client for redis sentinel connection

*

*/

network::redis_connection m_client;

/**

* queue of callback to process

*

*/

std::queue<reply_callback_t> m_callbacks;

/**

* user defined disconnection handler to be called on disconnection

*

*/

sentinel_disconnect_handler_t m_disconnect_handler;

/**

* callbacks thread safety

*

*/

std::mutex m_callbacks_mutex;

/**

* condvar for callbacks updates

*

*/

std::condition_variable m_sync_condvar;

/**

* number of callbacks currently being running

*

*/

std::atomic<unsigned int> m_callbacks_running;

};

} // namespace cpp_redis