include/private/redis/asynchirediscommanddispatcher.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_ASYNCHIREDISCOMMANDDISPATCHER_HPP_
  18 #define SHAREDDATALAYER_REDIS_ASYNCHIREDISCOMMANDDISPATCHER_HPP_
  19
  20 #include "private/redis/asynccommanddispatcher.hpp"
  21 #include <string>
  22 #include <list>
  23 #include <vector>
  24 #include <map>
  25 #include <memory>
  26 #include <queue>
  27 #include "private/logger.hpp"
  28 #include "private/timer.hpp"
  29
  30 extern "C"
  31 {
  32     struct redisReply;
  33     struct redisAsyncContext;
  34 }
  35
  36 namespace shareddatalayer
  37 {
  38     class Engine;
  39
  40     namespace redis
  41     {
  42         class HiredisSystem;
  43         class HiredisEpollAdapter;
  44         class Reply;
  45
  46         class AsyncHiredisCommandDispatcher: public AsyncCommandDispatcher
  47         {
  48         public:
  49             AsyncHiredisCommandDispatcher(const AsyncHiredisCommandDispatcher&) = delete;
  50
  51             AsyncHiredisCommandDispatcher& operator = (const AsyncHiredisCommandDispatcher&) = delete;
  52
  53             AsyncHiredisCommandDispatcher(Engine& engine,
  54                                           const std::string& address,
  55                                           uint16_t port,
  56                                           std::shared_ptr<ContentsBuilder> contentsBuilder,
  57                                           bool usePermanentCommandCallbacks,
  58                                           std::shared_ptr<Logger> logger,
  59                                           bool usedForSentinel);
  60
  61             AsyncHiredisCommandDispatcher(Engine& engine,
  62                                           const std::string& address,
  63                                           uint16_t port,
  64                                           std::shared_ptr<ContentsBuilder> contentsBuilder,
  65                                           bool usePermanentCommandCallbacks,
  66                                           HiredisSystem& hiredisSystem,
  67                                           std::shared_ptr<HiredisEpollAdapter> adapter,
  68                                           std::shared_ptr<Logger> logger,
  69                                           bool usedForSentinel);
  70
  71             ~AsyncHiredisCommandDispatcher() override;
  72
  73             void waitConnectedAsync(const ConnectAck& connectAck) override;
  74
  75             void registerDisconnectCb(const DisconnectCb& disconnectCb) override;
  76
  77             void dispatchAsync(const CommandCb& commandCb, const AsyncConnection::Namespace& ns,
  78                                const Contents& contents) override;
  79
  80             void disableCommandCallbacks() override;
  81
  82             void setConnected();
  83
  84             void setDisconnected();
  85
  86             void handleReply(const CommandCb& commandCb,
  87                              const std::error_code& error,
  88                              const redisReply* rr);
  89
  90             bool isClientCallbacksEnabled() const;
  91
  92             void verifyConnection();
  93
  94             void disconnectHiredis();
  95
  96             void armConnectionRetryTimer(Timer::Duration duration,
  97                                          std::function<void()> retryAction);
  98
  99         private:
 100             enum class ServiceState
 101             {   DISCONNECTED,
 102                 CONNECTION_VERIFICATION,
 103                 CONNECTED
 104             };
 105
 106             using Callback = std::function<void(const Reply&)>;
 107
 108             Engine& engine;
 109             std::string address;
 110             uint16_t port;
 111             std::shared_ptr<ContentsBuilder> contentsBuilder;
 112             bool usePermanentCommandCallbacks;
 113             HiredisSystem& hiredisSystem;
 114             std::shared_ptr<HiredisEpollAdapter> adapter;
 115             redisAsyncContext* ac;
 116             ConnectAck connectAck;
 117             DisconnectCb disconnectCallback;
 118             ServiceState serviceState;
 119             std::list<CommandCb> cbs;
 120             bool clientCallbacksEnabled;
 121             Timer connectionRetryTimer;
 122             Timer::Duration connectionRetryTimerDuration;
 123             Timer::Duration connectionVerificationRetryTimerDuration;
 124             std::shared_ptr<Logger> logger;
 125             bool usedForSentinel;
 126
 127             void connect();
 128
 129             bool isValidCb(const CommandCb& commandCb);
 130
 131             void removeCb(const CommandCb& commandCb);
 132
 133             void callCommandCbWithError(const CommandCb& commandCb, const std::error_code& error);
 134
 135             void dispatchAsync(const CommandCb& commandCb, const Contents& contents, bool checkConnectionState);
 136
 137             void verifyConnectionReply(const std::error_code& error, const redis::Reply& reply);
 138         };
 139     }
 140 }
 141
 142 #endif