sentinel.hpp

// 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 <vector>

#include <cpp_redis/misc/logger.hpp>
#include <cpp_redis/network/redis_connection.hpp>

namespace cpp_redis {

class sentinel {
public:
#ifndef __CPP_REDIS_USE_CUSTOM_TCP_CLIENT
  sentinel(void);
#endif /* __CPP_REDIS_USE_CUSTOM_TCP_CLIENT */

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

  ~sentinel(void);

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

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

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

  sentinel& commit(void);

  sentinel& sync_commit(void);

  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:
  sentinel& add_sentinel(const std::string& host, std::size_t port, std::uint32_t timeout_msecs = 0);

  void clear_sentinels(void);

public:
  void disconnect(bool wait_for_removal = false);

  bool is_connected(void);

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

  void connect_sentinel(const sentinel_disconnect_handler_t& disconnect_handler = nullptr);

  void connect(
    const std::string& host,
    std::size_t port,
    const sentinel_disconnect_handler_t& disconnect_handler = nullptr,
    std::uint32_t timeout_msecs                             = 0);

  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:
  class sentinel_def {
  public:
    sentinel_def(const std::string& host, std::size_t port, std::uint32_t timeout_msecs)
    : m_host(host), m_port(port), m_timeout_msecs(timeout_msecs) {}

    ~sentinel_def(void) = default;

  public:
    const std::string&
    get_host(void) const { return m_host; }

    size_t
    get_port(void) const { return m_port; }

    std::uint32_t
    get_timeout_msecs(void) const { return m_timeout_msecs; }

    void
    set_timeout_msecs(std::uint32_t timeout_msecs) { m_timeout_msecs = timeout_msecs; }

  private:
    std::string m_host;

    std::size_t m_port;

    std::uint32_t m_timeout_msecs;
  };

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

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

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

  void connection_disconnect_handler(network::redis_connection& connection);

  void call_disconnect_handler(void);

  void clear_callbacks(void);

  void try_commit(void);

private:
  std::vector<sentinel_def> m_sentinels;

  network::redis_connection m_client;

  std::queue<reply_callback_t> m_callbacks;

  sentinel_disconnect_handler_t m_disconnect_handler;

  std::mutex m_callbacks_mutex;

  std::condition_variable m_sync_condvar;

  std::atomic<unsigned int> m_callbacks_running;
};

}; // namespace cpp_redis