include/private/redis/asynchiredisclustercommanddispatcher.hpp

   1 /*
   2    Copyright (c) 2018-2019 Nokia.
   3
   4    Licensed under the Apache License, Version 2.0 (the "License");
   5    you may not use this file except in compliance with the License.
   6    You may obtain a copy of the License at
   7
   8        http://www.apache.org/licenses/LICENSE-2.0
   9
  10    Unless required by applicable law or agreed to in writing, software
  11    distributed under the License is distributed on an "AS IS" BASIS,
  12    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13    See the License for the specific language governing permissions and
  14    limitations under the License.
  15 */
  16
  17 #ifndef SHAREDDATALAYER_REDIS_ASYNCHIREDISCLUSTERCOMMANDDISPATCHER_HPP_
  18 #define SHAREDDATALAYER_REDIS_ASYNCHIREDISCLUSTERCOMMANDDISPATCHER_HPP_
  19
  20 #include "private/redis/asynccommanddispatcher.hpp"
  21 #include "private/databaseconfiguration.hpp"
  22 #include "private/logger.hpp"
  23 #include "private/timer.hpp"
  24 #include <string>
  25 #include <set>
  26 #include <list>
  27 #include <vector>
  28 #include <map>
  29 #include <memory>
  30 #include <queue>
  31 #include <boost/optional.hpp>
  32
  33 extern "C"
  34 {
  35     struct redisReply;
  36     struct redisClusterAsyncContext;
  37     struct redisAsyncContext;
  38 }
  39
  40 namespace shareddatalayer
  41 {
  42     class Engine;
  43
  44     namespace redis
  45     {
  46         class HiredisClusterSystem;
  47         class HiredisClusterEpollAdapter;
  48         class Reply;
  49
  50         class AsyncHiredisClusterCommandDispatcher: public AsyncCommandDispatcher
  51         {
  52         public:
  53             AsyncHiredisClusterCommandDispatcher(const AsyncHiredisClusterCommandDispatcher&) = delete;
  54
  55             AsyncHiredisClusterCommandDispatcher& operator = (const AsyncHiredisClusterCommandDispatcher&) = delete;
  56
  57             AsyncHiredisClusterCommandDispatcher(Engine& engine,
  58                                                  const boost::optional<std::string>& ns,
  59                                                  const DatabaseConfiguration::Addresses& addresses,
  60                                                  std::shared_ptr<ContentsBuilder> contentsBuilder,
  61                                                  bool usePermanentCommandCallbacks,
  62                                                  std::shared_ptr<Logger> logger);
  63
  64             AsyncHiredisClusterCommandDispatcher(Engine& engine,
  65                                                  const boost::optional<std::string>& ns,
  66                                                  const DatabaseConfiguration::Addresses& addresses,
  67                                                  std::shared_ptr<ContentsBuilder> contentsBuilder,
  68                                                  bool usePermanentCommandCallbacks,
  69                                                  HiredisClusterSystem& hiredisClusterSystem,
  70                                                  std::shared_ptr<HiredisClusterEpollAdapter> adapter,
  71                                                  std::shared_ptr<Logger> logger);
  72
  73             ~AsyncHiredisClusterCommandDispatcher() override;
  74
  75             void waitConnectedAsync(const ConnectAck& connectAck) override;
  76
  77             void registerDisconnectCb(const DisconnectCb& disconnectCb) override;
  78
  79             void dispatchAsync(const CommandCb& commandCb, const AsyncConnection::Namespace& ns, const Contents& contents) override;
  80
  81             void disableCommandCallbacks() override;
  82
  83             void handleReply(const CommandCb& commandCb, const std::error_code& error, const redisReply* rr);
  84
  85             bool isClientCallbacksEnabled() const;
  86
  87             void handleDisconnect(const redisAsyncContext* ac);
  88
  89         private:
  90             enum class ServiceState
  91             {
  92                 DISCONNECTED,
  93                 CONNECTED
  94             };
  95
  96             using Callback = std::function<void(const Reply&)>;
  97
  98             Engine& engine;
  99             const boost::optional<std::string> initialNamespace;
 100             const DatabaseConfiguration::Addresses addresses;
 101             std::shared_ptr<ContentsBuilder> contentsBuilder;
 102             bool usePermanentCommandCallbacks;
 103             HiredisClusterSystem& hiredisClusterSystem;
 104             std::shared_ptr<HiredisClusterEpollAdapter> adapter;
 105             redisClusterAsyncContext* acc;
 106             ConnectAck connectAck;
 107             DisconnectCb disconnectCallback;
 108             ServiceState serviceState;
 109             std::list<CommandCb> cbs;
 110             bool clientCallbacksEnabled;
 111             Timer connectionRetryTimer;
 112             Timer::Duration connectionRetryTimerDuration;
 113             std::shared_ptr<Logger> logger;
 114
 115             void connect();
 116
 117             bool isValidCb(const CommandCb& commandCb);
 118
 119             void removeCb(const CommandCb& commandCb);
 120
 121             void callCommandCbWithError(const CommandCb& commandCb, const std::error_code& error);
 122
 123             void dispatchAsync(const CommandCb& commandCb, const AsyncConnection::Namespace& ns, const Contents& contents, bool checkConnectionState);
 124
 125             void verifyConnection();
 126
 127             void verifyConnectionReply(const std::error_code& error, const redis::Reply& reply);
 128
 129             void setConnected();
 130
 131             void armConnectionRetryTimer();
 132
 133             void disconnectHiredisCluster();
 134         };
 135     }
 136 }
 137
 138 #endif