cpp_redis/include/cpp_redis/core/sentinel.hpp at develop · cpp-redis/cpp_redis

450 lines (387 loc) · 11.7 KB
// 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.
#ifndef CPP_REDIS_CORE_SENTINEL_HPP_
#define CPP_REDIS_CORE_SENTINEL_HPP_
#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 {
  //!  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 shared_ptr<tcp_client_iface_t> &tcp_client);
  //!  dtor
  ~sentinel();
  //!  copy ctor
  sentinel(const sentinel &) = delete;
  //!  assignment operator
  sentinel &operator=(const sentinel &) = delete;
  //!  callback to be called whenever a reply has been received
  using reply_callback_t = std::function<void(reply_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 vector<string_t> &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<mutex_t> 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;
  //!  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 string_t &host, size_t port,
                         uint_t timeout_ms = 0);
  //!  clear all existing sentinels.
  void clear_sentinels();
  //!  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
  using sentinel_disconnect_handler_t = std::function<void(sentinel &)>;
  //!  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
  connect(const string_t &host, size_t port,
          const sentinel_disconnect_handler_t &disconnect_handler = nullptr,
          uint_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 string_t &name, string_t &host,
                               size_t &port, bool autoconnect = true);
  sentinel &ckquorum(const string_t &name,
                     const reply_callback_t &reply_callback = nullptr);
  sentinel &failover(const string_t &name,
                     const reply_callback_t &reply_callback = nullptr);
  sentinel &flushconfig(const reply_callback_t &reply_callback = nullptr);
  sentinel &master(const string_t &name,
                   const reply_callback_t &reply_callback = nullptr);
  sentinel &masters(const reply_callback_t &reply_callback = nullptr);
  sentinel &monitor(const string_t &name, const string_t &ip, size_t port,
                    size_t quorum,
                    const reply_callback_t &reply_callback = nullptr);
  sentinel &ping(const reply_callback_t &reply_callback = nullptr);
  sentinel &remove(const string_t &name,
                   const reply_callback_t &reply_callback = nullptr);
  sentinel &reset(const string_t &pattern,
                  const reply_callback_t &reply_callback = nullptr);
  sentinel &sentinels(const string_t &name,
                      const reply_callback_t &reply_callback = nullptr);
  sentinel &set(const string_t &name, const string_t &option,
                const string_t &value,
                const reply_callback_t &reply_callback = nullptr);
  sentinel &slaves(const string_t &name,
                   const reply_callback_t &reply_callback = nullptr);
  //!  store informations related to a sentinel
  //!  typically, host, port and connection timeout
  class sentinel_def {
    //!  ctor
    sentinel_def(string_t host, size_t port, uint_t timeout_ms)
        : m_host(std::move(host)), m_port(port), m_timeout_ms(timeout_ms) {}
    //!  dtor
    ~sentinel_def() = default;
    //!  @return sentinel host
    const string_t &get_host() const { return m_host; }
    //!  @return sentinel port
    size_t get_port() const { return m_port; }
    //!  @return timeout for sentinel
    uint_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(uint_t timeout_ms) { m_timeout_ms = timeout_ms; }
    //!  sentinel host
    string_t m_host;
    //!  sentinel port
    size_t m_port;
    //!  connect timeout config
    uint_t m_timeout_ms;
  //!  @return sentinels
  const vector<sentinel_def> &get_sentinels() const;
  //!  @return sentinels (non-const version)
  vector<sentinel_def> &get_sentinels();
  //!  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(redis_connection_t &connection,
                                  reply_t &reply);
  //!  redis_connection disconnection handler, triggered whenever a
  //!  disconnection occurred
  //!  @param connection redis_connection instance
  void connection_disconnect_handler(redis_connection_t &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();
  //!  A pool of 1 or more sentinels we ask to determine which redis server is
  //!  the master.
  vector<sentinel_def> m_sentinels;
  //!  tcp client for redis sentinel connection
  redis_connection_t m_client;
  //!  queue of callback to process
  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
  mutex_t m_callbacks_mutex;
  //!  condvar for callbacks updates
  condition_variable_t m_sync_condvar;
  //!  number of callbacks currently being running
  atomic_uint m_callbacks_running;
using sentinel_t = sentinel;
} // namespace cpp_redis
#endif // !CPP_REDIS_CORE_SENTINEL_HPP_
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

sentinel.hpp

Latest commit

History

sentinel.hpp

File metadata and controls
