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 KVDB_SYNCABLE_TEST_H 17 #define KVDB_SYNCABLE_TEST_H 18 19 #include <map> 20 #include <mutex> 21 #include <vector> 22 23 #include "single_ver_kvdb_sync_interface.h" 24 #include "query_object.h" 25 #include "store_types.h" 26 27 namespace DistributedDB { 28 struct VirtualDataItem { 29 Key key; 30 Value value; 31 Timestamp timestamp = 0; 32 Timestamp writeTimestamp = 0; 33 uint64_t flag = 0; 34 bool isLocal = true; 35 uint32_t deviceId = 0; // 0: means local 36 static const uint64_t DELETE_FLAG = 0x01; 37 static const uint64_t LOCAL_FLAG = 0x02; 38 }; 39 40 struct VirtualContinueToken { 41 Timestamp begin = 0u; 42 Timestamp end = 0u; 43 }; 44 class VirtualSingleVerSyncDBInterface : public SingleVerKvDBSyncInterface { 45 public: 46 VirtualSingleVerSyncDBInterface(); 47 int GetInterfaceType() const override; 48 49 void IncRefCount() override; 50 51 void DecRefCount() override; 52 53 std::vector<uint8_t> GetIdentifier() const override; 54 55 int GetMetaData(const Key& key, Value& value) const override; 56 57 int GetMetaDataByPrefixKey(const Key &keyPrefix, std::map<Key, Value> &data) const override; 58 59 int PutMetaData(const Key& key, const Value& value, bool isInTransaction) override; 60 61 int DeleteMetaData(const std::vector<Key> &keys) override; 62 // Delete multiple meta data records with key prefix in a transaction. 63 int DeleteMetaDataByPrefixKey(const Key &keyPrefix) const override; 64 65 int GetAllMetaKeys(std::vector<Key>& keys) const override; 66 67 int GetSyncData(Timestamp begin, Timestamp end, std::vector<DataItem> &dataItems, 68 ContinueToken &continueStmtToken, const DataSizeSpecInfo &dataSizeInfo) const override; 69 70 int GetSyncDataNext(std::vector<DataItem> &dataItems, ContinueToken &continueStmtToken, 71 const DataSizeSpecInfo &dataSizeInfo) const override; 72 73 void ReleaseContinueToken(ContinueToken& continueStmtToken) const override; 74 75 void GetMaxTimestamp(Timestamp& stamp) const override; 76 77 int RemoveDeviceData(const std::string &deviceName, bool isNeedNotify) override; 78 79 int GetSyncData(const Key& key, VirtualDataItem& dataItem); 80 81 int PutSyncData(const DataItem& item); 82 83 int PutData(const Key &key, const Value &value, const Timestamp &time, int flag); 84 85 int GetSyncData(Timestamp begin, Timestamp end, std::vector<SingleVerKvEntry *> &entries, 86 ContinueToken &continueStmtToken, const DataSizeSpecInfo &dataSizeInfo) const override; 87 88 int GetSyncData(QueryObject &query, const SyncTimeRange &timeRange, const DataSizeSpecInfo &dataSizeInfo, 89 ContinueToken &continueStmtToken, std::vector<SingleVerKvEntry *> &entries) const override; 90 91 int GetUnSyncTotal(Timestamp begin, Timestamp end, uint32_t &total) const override; 92 93 int GetUnSyncTotal(QueryObject &query, const SyncTimeRange &timeRange, uint32_t &total) const override; 94 95 int GetSyncDataNext(std::vector<SingleVerKvEntry *> &entries, ContinueToken &continueStmtToken, 96 const DataSizeSpecInfo &dataSizeInfo) const override; 97 98 int PutSyncDataWithQuery(const QueryObject &query, const std::vector<SingleVerKvEntry *> &entries, 99 const std::string &deviceName) override; 100 101 SchemaObject GetSchemaInfo() const override; 102 103 bool CheckCompatible(const std::string& schema, uint8_t type) const override; 104 105 void SetSchemaInfo(const std::string& schema); 106 107 const KvDBProperties &GetDbProperties() const override; 108 109 void SetSaveDataDelayTime(uint64_t milliDelayTime); 110 111 int GetSecurityOption(SecurityOption &option) const override; 112 113 bool IsReadable() const override; 114 115 void SetSecurityOption(SecurityOption &option); 116 117 void NotifyRemotePushFinished(const std::string &targetId) const override; 118 119 int GetDatabaseCreateTimestamp(Timestamp &outTime) const override; 120 121 int GetCompressionOption(bool &needCompressOnSync, uint8_t &compressionRate) const override; 122 int GetCompressionAlgo(std::set<CompressAlgorithm> &algorithmSet) const override; 123 124 // return E_OK if subscribe is legal, ERROR on exception. 125 int CheckAndInitQueryCondition(QueryObject &query) const override; 126 127 int InterceptData(std::vector<SingleVerKvEntry *> &entries, const std::string &sourceID, 128 const std::string &targetID, bool isPush) const override; 129 130 int AddSubscribe(const std::string &subscribeId, const QueryObject &query, bool needCacheSubscribe) override; 131 132 int RemoveSubscribe(const std::string &subscribeId) override; 133 134 int RemoveSubscribe(const std::vector<std::string> &subscribeIds) override; 135 136 void SetBusy(bool busy, bool readBusy = false); 137 138 void PutDeviceData(const std::string &deviceName, const Key &key, const Value &value); 139 140 void GetDeviceData(const std::string &deviceName, const Key &key, Value &value); 141 142 void SetIdentifier(std::vector<uint8_t> &identifier); 143 144 void SetDbProperties(KvDBProperties &kvDBProperties); 145 146 void DelayGetSyncData(uint32_t milliDelayTime); 147 148 void SetGetDataErrCode(int whichTime, int errCode, bool isGetDataControl); 149 void ResetDataControl(); 150 151 void SetSaveDataCallback(const std::function<void()> &callback); 152 153 void ForkGetSecurityOption(std::function<int(SecurityOption &)> callBack); 154 155 void SetPushNotifier(const std::function<void(const std::string &)> &pushNotifier); 156 157 void SetCompressSync(bool compressSync); 158 private: 159 int GetSyncData(Timestamp begin, Timestamp end, const DataSizeSpecInfo &dataSizeInfo, 160 std::vector<VirtualDataItem> &dataItems, ContinueToken &continueStmtToken) const; 161 162 int GetSyncDataNext(std::vector<VirtualDataItem>& dataItems, 163 uint32_t blockSize, ContinueToken& continueStmtToken) const; 164 165 int PutSyncData(std::vector<VirtualDataItem>& dataItems, const std::string &deviceName); 166 167 int DataControl() const; 168 169 bool GetDataInner(Timestamp begin, Timestamp end, Timestamp ¤tWaterMark, 170 const DataSizeSpecInfo &dataSizeInfo, std::vector<VirtualDataItem> &dataItems) const; 171 172 mutable std::map<std::vector<uint8_t>, std::vector<uint8_t>> metadata_; 173 std::vector<VirtualDataItem> dbData_; 174 std::map<std::string, uint32_t> deviceMapping_; // key: deviceName, value: deviceId 175 uint32_t availableDeviceId_ = 0; 176 std::string schema_; 177 SchemaObject schemaObj_; 178 KvDBProperties properties_; 179 uint64_t saveDataDelayTime_ = 0; 180 SecurityOption secOption_; 181 bool busy_ = false; 182 bool readBusy_ = false; 183 184 std::mutex deviceDataLock_; 185 std::map<std::string, std::map<Key, Value>> deviceData_; 186 std::vector<uint8_t> identifier_; 187 uint64_t getDataDelayTime_ = 0; 188 uint64_t dbCreateTime_; 189 190 int countDown_ = -1; 191 int expectedErrCode_ = E_OK; 192 bool isGetDataControl_ = true; // control get data: true, control save data : false 193 194 std::mutex saveDataMutex_; 195 std::function<void()> saveDataCallback_; 196 197 std::function<int(SecurityOption &)> getSecurityOptionCallBack_; 198 199 mutable std::mutex pushNotifierMutex_; 200 std::function<void(const std::string &)> pushNotifier_; 201 202 bool compressSync_ = false; 203 }; 204 } // namespace DistributedDB 205 206 #endif // KVDB_SYNCABLE_TEST_H