1 /* 2 * Copyright (c) 2022 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 #ifndef DISTRIBUTEDDATAMGR_OBJECT_MANAGER_H 16 #define DISTRIBUTEDDATAMGR_OBJECT_MANAGER_H 17 18 #include <atomic> 19 20 #include "concurrent_map.h" 21 #include "device_manager_adapter.h" 22 #include "kv_store_delegate_manager.h" 23 #include "kvstore_sync_callback.h" 24 #include "object_asset_loader.h" 25 #include "object_callback.h" 26 #include "object_callback_proxy.h" 27 #include "object_common.h" 28 #include "object_data_listener.h" 29 #include "object_snapshot.h" 30 #include "object_types.h" 31 #include "serializable/serializable.h" 32 #include "types.h" 33 #include "value_proxy.h" 34 namespace OHOS { 35 namespace DistributedObject { 36 using SyncCallBack = std::function<void(const std::map<std::string, int32_t> &results)>; 37 using ObjectRecord = std::map<std::string, std::vector<uint8_t>>; 38 class SequenceSyncManager { 39 public: 40 enum Result { 41 SUCCESS_USER_IN_USE, 42 SUCCESS_USER_HAS_FINISHED, 43 ERR_SID_NOT_EXIST 44 }; GetInstance()45 static SequenceSyncManager *GetInstance() 46 { 47 static SequenceSyncManager sequenceSyncManager; 48 return &sequenceSyncManager; 49 } 50 51 uint64_t AddNotifier(const std::string &userId, SyncCallBack &callback); 52 Result DeleteNotifier(uint64_t sequenceId, std::string &userId); 53 Result Process( 54 uint64_t sequenceId, const std::map<std::string, DistributedDB::DBStatus> &results, std::string &userId); 55 56 private: 57 Result DeleteNotifierNoLock(uint64_t sequenceId, std::string &userId); 58 std::mutex notifierLock_; 59 std::map<std::string, std::vector<uint64_t>> userIdSeqIdRelations_; 60 std::map<uint64_t, SyncCallBack> seqIdCallbackRelations_; 61 }; 62 63 class ObjectStoreManager { 64 public: 65 using DmAdaper = OHOS::DistributedData::DeviceManagerAdapter; 66 using UriToSnapshot = std::shared_ptr<std::map<std::string, std::shared_ptr<Snapshot>>>; 67 68 enum RestoreStatus : int32_t { 69 NONE = 0, 70 DATA_READY, 71 DATA_NOTIFIED, 72 ASSETS_READY, 73 ALL_READY, 74 STATUS_BUTT 75 }; 76 77 ObjectStoreManager(); 78 ~ObjectStoreManager(); GetInstance()79 static ObjectStoreManager *GetInstance() 80 { 81 static ObjectStoreManager *manager = new ObjectStoreManager(); 82 return manager; 83 } 84 int32_t Save(const std::string &appId, const std::string &sessionId, const ObjectRecord &data, 85 const std::string &deviceId, sptr<IRemoteObject> callback); 86 int32_t RevokeSave( 87 const std::string &appId, const std::string &sessionId, sptr<IRemoteObject> callback); 88 int32_t Retrieve(const std::string &bundleName, const std::string &sessionId, 89 sptr<IRemoteObject> callback, uint32_t tokenId); 90 void SetData(const std::string &dataDir, const std::string &userId); 91 int32_t Clear(); 92 int32_t DeleteByAppId(const std::string &appId, int32_t user); 93 void RegisterRemoteCallback(const std::string &bundleName, const std::string &sessionId, 94 pid_t pid, uint32_t tokenId, 95 sptr<IRemoteObject> callback); 96 void UnregisterRemoteCallback(const std::string &bundleName, pid_t pid, uint32_t tokenId, 97 const std::string &sessionId = ""); 98 void NotifyChange(ObjectRecord &changedData); 99 void NotifyAssetsReady(const std::string& objectKey, const std::string& bundleName, 100 const std::string& srcNetworkId = ""); 101 void NotifyAssetsStart(const std::string& objectKey, const std::string& srcNetworkId = ""); 102 void CloseAfterMinute(); 103 int32_t Open(); 104 void SetThreadPool(std::shared_ptr<ExecutorPool> executors); 105 UriToSnapshot GetSnapShots(const std::string &bundleName, const std::string &storeName); 106 int32_t BindAsset(const uint32_t tokenId, const std::string& appId, const std::string& sessionId, 107 ObjectStore::Asset& asset, ObjectStore::AssetBindInfo& bindInfo); 108 int32_t OnAssetChanged(const uint32_t tokenId, const std::string& appId, const std::string& sessionId, 109 const std::string& deviceId, const ObjectStore::Asset& asset); 110 void DeleteSnapshot(const std::string &bundleName, const std::string &sessionId); 111 private: 112 constexpr static const char *SEPERATOR = "_"; 113 constexpr static const char *TIME_REGEX = "_\\d{10}_p_"; 114 constexpr static int BUNDLE_NAME_INDEX = 0; 115 constexpr static int SESSION_ID_INDEX = 1; 116 constexpr static int SOURCE_DEVICE_UDID_INDEX = 2; 117 constexpr static int TIME_INDEX = 4; 118 constexpr static int PROPERTY_NAME_INDEX = 5; 119 constexpr static const char *LOCAL_DEVICE = "local"; 120 constexpr static const char *USERID = "USERID"; 121 constexpr static int8_t MAX_OBJECT_SIZE_PER_APP = 16; 122 constexpr static int8_t DECIMAL_BASE = 10; 123 constexpr static int WAIT_TIME = 60; 124 static constexpr size_t TIME_TASK_NUM = 1; 125 static constexpr int64_t INTERVAL = 1; 126 struct CallbackInfo { 127 pid_t pid; 128 std::map<std::string, sptr<ObjectChangeCallbackProxy>> observers_; 129 bool operator<(const CallbackInfo &it_) const 130 { 131 if (pid < it_.pid) { 132 return true; 133 } 134 return false; 135 } 136 }; 137 struct SaveInfo : DistributedData::Serializable { 138 std::string bundleName; 139 std::string sessionId; 140 std::string sourceDeviceId; 141 std::string targetDeviceId; 142 std::string timestamp; 143 SaveInfo() = default; 144 SaveInfo(const std::string &bundleName, const std::string &sessionId, const std::string &sourceDeviceId, 145 const std::string &targetDeviceId, const std::string ×tamp); 146 bool Marshal(json &node) const override; 147 bool Unmarshal(const json &node) override; 148 std::string ToPropertyPrefix(); 149 }; 150 DistributedDB::KvStoreNbDelegate *OpenObjectKvStore(); 151 void FlushClosedStore(); 152 void Close(); 153 int32_t SetSyncStatus(bool status); 154 int32_t SaveToStore(const std::string &appId, const std::string &sessionId, const std::string &toDeviceId, 155 const ObjectRecord &data); 156 int32_t SyncOnStore(const std::string &prefix, const std::vector<std::string> &deviceList, SyncCallBack &callback); 157 int32_t RevokeSaveToStore(const std::string &prefix); 158 int32_t RetrieveFromStore(const std::string &appId, const std::string &sessionId, ObjectRecord &results); 159 void SyncCompleted(const std::map<std::string, DistributedDB::DBStatus> &results, uint64_t sequenceId); 160 std::vector<std::string> SplitEntryKey(const std::string &key); 161 void ProcessOldEntry(const std::string &appId); 162 void ProcessSyncCallback(const std::map<std::string, int32_t> &results, const std::string &appId, 163 const std::string &sessionId, const std::string &deviceId); 164 void SaveUserToMeta(); 165 std::string GetCurrentUser(); 166 void DoNotify(uint32_t tokenId, const CallbackInfo& value, const std::map<std::string, ObjectRecord>& data, 167 bool allReady); 168 void DoNotifyAssetsReady(uint32_t tokenId, const CallbackInfo& value, const std::string& objectKey, bool allReady); 169 void DoNotifyWaitAssetTimeout(const std::string &objectKey); 170 std::map<std::string, std::map<std::string, Assets>> GetAssetsFromStore(const ObjectRecord& changedData); 171 static bool IsAssetKey(const std::string& key); 172 static bool IsAssetComplete(const ObjectRecord& result, const std::string& assetPrefix); 173 Assets GetAssetsFromDBRecords(const ObjectRecord& result); 174 bool RegisterAssetsLister(); 175 void ComputeStatus(const std::string& objectKey, const SaveInfo& saveInfo, 176 const std::map<std::string, ObjectRecord>& data); 177 void NotifyDataChanged(const std::map<std::string, ObjectRecord>& data, const SaveInfo& saveInfo); 178 int32_t PushAssets(const std::string &srcBundleName, const std::string &dstBundleName, const std::string &sessionId, 179 const ObjectRecord &data, const std::string &deviceId); 180 int32_t WaitAssets(const std::string& objectKey, const SaveInfo& saveInfo, 181 const std::map<std::string, ObjectRecord>& data); 182 void PullAssets(const std::map<std::string, ObjectRecord>& data, const SaveInfo& saveInfo); 183 std::map<std::string, ObjectRecord> GetObjectData(const ObjectRecord& changedData, SaveInfo& saveInfo, 184 bool& hasAsset); GetPropertyPrefix(const std::string & appId,const std::string & sessionId)185 inline std::string GetPropertyPrefix(const std::string &appId, const std::string &sessionId) 186 { 187 return appId + SEPERATOR + sessionId + SEPERATOR + DmAdaper::GetInstance().GetLocalDevice().udid + SEPERATOR; 188 }; GetPropertyPrefix(const std::string & appId,const std::string & sessionId,const std::string & toDeviceId)189 inline std::string GetPropertyPrefix( 190 const std::string &appId, const std::string &sessionId, const std::string &toDeviceId) 191 { 192 return appId + SEPERATOR + sessionId + SEPERATOR + DmAdaper::GetInstance().GetLocalDevice().udid 193 + SEPERATOR + toDeviceId + SEPERATOR; 194 }; GetPrefixWithoutDeviceId(const std::string & appId,const std::string & sessionId)195 inline std::string GetPrefixWithoutDeviceId(const std::string &appId, const std::string &sessionId) 196 { 197 return appId + SEPERATOR + sessionId + SEPERATOR; 198 }; GetMetaUserIdKey(const std::string & userId,const std::string & appId)199 inline std::string GetMetaUserIdKey(const std::string &userId, const std::string &appId) 200 { 201 return std::string(USERID) + SEPERATOR + userId + SEPERATOR + appId + SEPERATOR 202 + DmAdaper::GetInstance().GetLocalDevice().udid; 203 }; 204 std::recursive_mutex kvStoreMutex_; 205 std::mutex mutex_; 206 DistributedDB::KvStoreDelegateManager *kvStoreDelegateManager_ = nullptr; 207 DistributedDB::KvStoreNbDelegate *delegate_ = nullptr; 208 ObjectDataListener *objectDataListener_ = nullptr; 209 sptr<ObjectAssetsRecvListener> objectAssetsRecvListener_ = nullptr; 210 sptr<ObjectAssetsSendListener> objectAssetsSendListener_ = nullptr; 211 uint32_t syncCount_ = 0; 212 std::string userId_; 213 std::atomic<bool> isSyncing_ = false; 214 ConcurrentMap<uint32_t /* tokenId */, CallbackInfo > callbacks_; 215 std::shared_ptr<ExecutorPool> executors_; 216 DistributedData::AssetBindInfo ConvertBindInfo(ObjectStore::AssetBindInfo& bindInfo); 217 VBucket ConvertVBucket(ObjectStore::ValuesBucket &vBucket); 218 ConcurrentMap<std::string, std::shared_ptr<Snapshot>> snapshots_; // key:bundleName_sessionId 219 ConcurrentMap<std::string, UriToSnapshot> bindSnapshots_; // key:bundleName_storeName 220 ConcurrentMap<std::string, RestoreStatus> restoreStatus_; // key:bundleName+sessionId 221 ConcurrentMap<std::string, ExecutorPool::TaskId> objectTimer_; // key:bundleName+sessionId 222 }; 223 } // namespace DistributedObject 224 } // namespace OHOS 225 #endif // DISTRIBUTEDDATAMGR_OBJECT_MANAGER_H 226