1 /* 2 * Copyright (c) 2021 Huawei Device Co., Ltd. 3 * Licensed under the Apache License, Version 2.0 (the "License"); 4 * you may not use this file except in compliance with the License. 5 * You may obtain a copy of the License at 6 * 7 * http://www.apache.org/licenses/LICENSE-2.0 8 * 9 * Unless required by applicable law or agreed to in writing, software 10 * distributed under the License is distributed on an "AS IS" BASIS, 11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 * See the License for the specific language governing permissions and 13 * limitations under the License. 14 */ 15 16 #ifndef VIRTUAL_COMMUNICATOR_H 17 #define VIRTUAL_COMMUNICATOR_H 18 19 #include <chrono> 20 #include <condition_variable> 21 #include <cstdint> 22 #include <functional> 23 #include <map> 24 #include <mutex> 25 #include <string> 26 27 #include "icommunicator.h" 28 #include "ref_object.h" 29 #include "serial_buffer.h" 30 #include "sync_types.h" 31 32 namespace DistributedDB { 33 class VirtualCommunicatorAggregator; 34 35 class VirtualCommunicator : public ICommunicator { 36 public: 37 VirtualCommunicator(const std::string &deviceId, VirtualCommunicatorAggregator *communicatorAggregator); 38 ~VirtualCommunicator() override; 39 40 DISABLE_COPY_ASSIGN_MOVE(VirtualCommunicator); 41 42 int RegOnMessageCallback(const OnMessageCallback &onMessage, const Finalizer &inOper) override; 43 int RegOnConnectCallback(const OnConnectCallback &onConnect, const Finalizer &inOper) override; 44 int RegOnSendableCallback(const std::function<void(void)> &onSendable, const Finalizer &inOper) override; 45 46 void Activate() override; 47 48 uint32_t GetCommunicatorMtuSize() const override; 49 uint32_t GetCommunicatorMtuSize(const std::string &target) const override; 50 void SetCommunicatorMtuSize(uint32_t mtuSize); 51 52 uint32_t GetTimeout() const override; 53 uint32_t GetTimeout(const std::string &target) const override; 54 void SetTimeout(uint32_t timeout); 55 56 int GetLocalIdentity(std::string &outTarget) const override; 57 58 int SendMessage(const std::string &dstTarget, const Message *inMsg, const SendConfig &config) override; 59 int SendMessage(const std::string &dstTarget, const Message *inMsg, const SendConfig &config, 60 const OnSendEnd &onEnd) override; 61 62 int GetRemoteCommunicatorVersion(const std::string &deviceId, uint16_t &version) const override; 63 64 void CallbackOnMessage(const std::string &srcTarget, Message *inMsg); 65 66 void CallbackOnConnect(const std::string &target, bool isConnect) const; 67 68 int GeneralVirtualSyncId(); 69 70 void Disable(); 71 72 void Enable(); 73 74 void SetDeviceId(const std::string &deviceId); 75 76 std::string GetDeviceId() const; 77 78 bool IsEnabled() const; 79 80 bool IsDeviceOnline(const std::string &device) const override; 81 82 void SetDropMessageTypeByDevice(MessageId msgid, uint32_t dropTimes = 1); 83 84 private: 85 int TimeSync(); 86 int DataSync(); 87 int WaterMarkSync(); 88 static int TranslateMsg(const Message *inMsg, Message *&outMsg); 89 90 mutable std::mutex onMessageLock_; 91 OnMessageCallback onMessage_; 92 93 mutable std::mutex onConnectLock_; 94 OnConnectCallback onConnect_; 95 mutable std::mutex devicesMapLock_; 96 mutable std::map<std::string, bool> onlineDevicesMap_; 97 98 std::string remoteDeviceId_ = "real_device"; 99 std::mutex syncIdLock_; 100 int currentSyncId_ = 1000; 101 bool isEnable_ = true; 102 std::string deviceId_; 103 104 std::mutex onAggregatorLock_; 105 VirtualCommunicatorAggregator *communicatorAggregator_; 106 107 uint32_t timeout_ = 5 * 1000; // 5 * 1000ms 108 MessageId dropMsgId_ = MessageId::UNKNOW_MESSAGE; 109 uint32_t dropMsgTimes_ = 0; 110 uint32_t mtuSize_ = 5 * 1024 * 1024; // 5 * 1024 * 1024B 111 }; 112 } // namespace DistributedDB 113 114 #endif // VIRTUAL_COMMUNICATOR_H 115