/* * Copyright (c) 2021 Huawei Device Co., Ltd. * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #include "db_adapter.h" #include "anonymous_string.h" #include "capability_info.h" #include "capability_info_manager.h" #include "capability_utils.h" #include "constants.h" #include "dh_context.h" #include "distributed_hardware_errno.h" #include "distributed_hardware_log.h" #include "event_bus.h" namespace OHOS { namespace DistributedHardware { #undef DH_LOG_TAG #define DH_LOG_TAG "DBAdapter" DBAdapter::DBAdapter(const std::string &appId, const std::string &storeId, const std::shared_ptr &changeListener) { this->appId_.appId = appId; this->storeId_.storeId = storeId; this->dataChangeListener_ = changeListener; DHLOGI("DBAdapter Constructor Success, appId: %s, storeId: %s", appId.c_str(), storeId.c_str()); } DBAdapter::~DBAdapter() { DHLOGI("DBAdapter Destruction"); } DistributedKv::Status DBAdapter::GetKvStorePtr() { DistributedKv::Options options = { .createIfMissing = true, .encrypt = false, .autoSync = true, .securityLevel = DistributedKv::SecurityLevel::S1, .kvStoreType = DistributedKv::KvStoreType::SINGLE_VERSION }; return kvDataMgr_.GetSingleKvStore(options, appId_, storeId_, kvStoragePtr_); } int32_t DBAdapter::Init() { DHLOGI("Init DB, storeId: %s", storeId_.storeId.c_str()); std::lock_guard lock(dbAdapterMutex_); int32_t tryTimes = MAX_INIT_RETRY_TIMES; while (tryTimes > 0) { DistributedKv::Status status = GetKvStorePtr(); if (status == DistributedKv::Status::SUCCESS && kvStoragePtr_) { DHLOGI("Init KvStorePtr Success"); RegisterManualSyncListener(); RegisterChangeListener(); RegisterKvStoreDeathListener(); return DH_FWK_SUCCESS; } DHLOGD("CheckKvStore, left times: %d", tryTimes); usleep(INIT_RETRY_SLEEP_INTERVAL); tryTimes--; } if (kvStoragePtr_ == nullptr) { DHLOGE("Init KvStorePtr failed"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } return DH_FWK_SUCCESS; } void DBAdapter::UnInit() { DHLOGI("DBAdapter UnInit"); std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return; } UnRegisterKvStoreDeathListener(); UnRegisterChangeListener(); UnRegisterManualSyncListener(); kvStoragePtr_.reset(); } int32_t DBAdapter::ReInit() { DHLOGI("ReInit DB, storeId: %s", storeId_.storeId.c_str()); std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } UnRegisterManualSyncListener(); kvStoragePtr_.reset(); DistributedKv::Status status = GetKvStorePtr(); if (status != DistributedKv::Status::SUCCESS || !kvStoragePtr_) { DHLOGW("Get kvStoragePtr_ failed, status: %d", status); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } RegisterManualSyncListener(); RegisterKvStoreDeathListener(); return DH_FWK_SUCCESS; } void DBAdapter::SyncCompleted(const std::map &results) { DHLOGI("DBAdapter SyncCompleted start"); std::lock_guard lock(dbAdapterMutex_); for (const auto &result : results) { std::string deviceId = result.first; DHLOGI("SyncCompleted, deviceId: %s", GetAnonyString(deviceId).c_str()); if (manualSyncCountMap_.count(deviceId) == 0) { DHLOGE("SyncCompleted, error, ManualSyncCount is removed"); return; } DHLOGI("ManualSyncCallback::SyncCompleted, retryCount: %d", manualSyncCountMap_[deviceId]); if (result.second == DistributedKv::Status::SUCCESS) { manualSyncCountMap_[deviceId] = 0; } else { manualSyncCountMap_[deviceId]++; if (manualSyncCountMap_[deviceId] >= MANUAL_SYNC_TIMES) { manualSyncCountMap_[deviceId] = 0; } else { auto retryTask = [this, deviceId] { this->ManualSync(deviceId); usleep(MANUAL_SYNC_INTERVAL); }; DHContext::GetInstance().GetEventBus()->PostTask(retryTask, "retryTask", 0); } } } } int32_t DBAdapter::GetDataByKey(const std::string &key, std::string &data) { DHLOGI("Get data by key: %s", GetAnonyString(key).c_str()); std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } DistributedKv::Key kvKey(key); DistributedKv::Value kvValue; DistributedKv::Status status = kvStoragePtr_->Get(kvKey, kvValue); if (status != DistributedKv::Status::SUCCESS) { DHLOGE("Query from db failed, key: %s", GetAnonyString(key).c_str()); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } data = kvValue.ToString(); return DH_FWK_SUCCESS; } int32_t DBAdapter::GetDataByKeyPrefix(const std::string &keyPrefix, std::vector &values) { DHLOGI("Get data by key prefix: %s", GetAnonyString(keyPrefix).c_str()); std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } // if prefix is empty, get all entries. DistributedKv::Key allEntryKeyPrefix(keyPrefix); std::vector allEntries; DistributedKv::Status status = kvStoragePtr_->GetEntries(allEntryKeyPrefix, allEntries); if (status != DistributedKv::Status::SUCCESS) { DHLOGE("Query data by keyPrefix failed, prefix: %s", GetAnonyString(keyPrefix).c_str()); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } for (const auto& item : allEntries) { values.push_back(item.value.ToString()); } return DH_FWK_SUCCESS; } int32_t DBAdapter::PutData(const std::string &key, std::string &value) { std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } DistributedKv::Key kvKey(key); DistributedKv::Value kvValue(value); DistributedKv::Status status = kvStoragePtr_->Put(kvKey, kvValue); if (status == DistributedKv::Status::IPC_ERROR) { DHLOGE("Put kv to db failed, ret: %d", status); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } return DH_FWK_SUCCESS; } int32_t DBAdapter::PutDataBatch(const std::vector &keys, const std::vector &values) { std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } if (keys.size() != values.size()) { DHLOGE("Param invalid"); return ERR_DH_FWK_PARA_INVALID; } std::vector entries; for (unsigned long i = 0; i < keys.size(); i++) { DistributedKv::Entry entry; entry.key = keys[i]; entry.value = values[i]; entries.push_back(entry); } DistributedKv::Status status = kvStoragePtr_->PutBatch(entries); if (status != DistributedKv::Status::SUCCESS) { DHLOGE("Put kv batch to db failed, ret: %d", status); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } DHLOGI("Put kv batch to db success"); return DH_FWK_SUCCESS; } void DBAdapter::CreateManualSyncCount(const std::string &deviceId) { std::lock_guard lock(dbAdapterMutex_); manualSyncCountMap_[deviceId] = 0; } void DBAdapter::RemoveManualSyncCount(const std::string &deviceId) { std::lock_guard lock(dbAdapterMutex_); manualSyncCountMap_.erase(deviceId); } int32_t DBAdapter::ManualSync(const std::string &networkId) { DHLOGI("Manual sync between networkId: %s", GetAnonyString(networkId).c_str()); std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } std::vector devList = { networkId }; DistributedKv::Status status = kvStoragePtr_->Sync(devList, DistributedKv::SyncMode::PULL); if (status != DistributedKv::Status::SUCCESS) { DHLOGE("ManualSync Data failed, networkId: %s", GetAnonyString(networkId).c_str()); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } return DH_FWK_SUCCESS; } void DBAdapter::SyncDBForRecover() { DHLOGI("Sync store id: %s after db recover", storeId_.storeId.c_str()); CapabilityInfoEvent recoverEvent(*this, CapabilityInfoEvent::EventType::RECOVER); DHContext::GetInstance().GetEventBus()->PostEvent(recoverEvent); } int32_t DBAdapter::RegisterChangeListener() { DHLOGI("Register db data change listener"); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } DistributedKv::Status status = kvStoragePtr_->SubscribeKvStore(DistributedKv::SubscribeType::SUBSCRIBE_TYPE_REMOTE, dataChangeListener_); if (status == DistributedKv::Status::IPC_ERROR) { DHLOGE("Register db data change listener failed, ret: %d", status); return ERR_DH_FWK_RESOURCE_REGISTER_DB_FAILED; } return DH_FWK_SUCCESS; } int32_t DBAdapter::UnRegisterChangeListener() { DHLOGI("UnRegister db data change listener"); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } DistributedKv::Status status = kvStoragePtr_->UnSubscribeKvStore( DistributedKv::SubscribeType::SUBSCRIBE_TYPE_REMOTE, dataChangeListener_); if (status == DistributedKv::Status::IPC_ERROR) { DHLOGE("UnRegister db data change listener failed, ret: %d", status); return ERR_DH_FWK_RESOURCE_UNREGISTER_DB_FAILED; } return DH_FWK_SUCCESS; } void DBAdapter::RegisterKvStoreDeathListener() { DHLOGI("Register kvStore death listener"); kvDataMgr_.RegisterKvStoreServiceDeathRecipient(shared_from_this()); } void DBAdapter::UnRegisterKvStoreDeathListener() { DHLOGI("UnRegister kvStore death listener"); kvDataMgr_.UnRegisterKvStoreServiceDeathRecipient(shared_from_this()); } void DBAdapter::RegisterManualSyncListener() { DHLOGI("Register manualSyncCallback"); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return; } kvStoragePtr_->RegisterSyncCallback(shared_from_this()); } void DBAdapter::UnRegisterManualSyncListener() { DHLOGI("UnRegister manualSyncCallback"); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return; } kvStoragePtr_->UnRegisterSyncCallback(); } void DBAdapter::OnRemoteDied() { DHLOGI("OnRemoteDied, recover db begin"); auto ReInitTask = [this] { int32_t times = 0; while (times < DIED_CHECK_MAX_TIMES) { // init kvStore. if (this->ReInit() == DH_FWK_SUCCESS) { // register data change listener again. this->RegisterChangeListener(); this->SyncDBForRecover(); DHLOGE("Current times is %d", times); break; } times++; usleep(DIED_CHECK_INTERVAL); } }; DHContext::GetInstance().GetEventBus()->PostTask(ReInitTask, "ReInitTask", 0); DHLOGI("OnRemoteDied, recover db end"); } void DBAdapter::DeleteKvStore() { std::lock_guard lock(dbAdapterMutex_); DistributedKv::Status status = kvDataMgr_.DeleteKvStore(appId_, storeId_); if (status != DistributedKv::Status::SUCCESS) { DHLOGE("DeleteKvStore error, appId: %s, storeId: %s, status: %d", appId_.appId.c_str(), storeId_.storeId.c_str(), status); return; } DHLOGI("DeleteKvStore success appId: %s", appId_.appId.c_str()); } int32_t DBAdapter::RemoveDeviceData(const std::string &deviceId) { std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } DistributedKv::Status status = kvStoragePtr_->RemoveDeviceData(deviceId); if (status != DistributedKv::Status::SUCCESS) { DHLOGE("Remove device data failed, deviceId: %s", GetAnonyString(deviceId).c_str()); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } DHLOGD("Remove device data success, deviceId: %s", GetAnonyString(deviceId).c_str()); return DH_FWK_SUCCESS; } int32_t DBAdapter::RemoveDataByKey(const std::string &key) { std::lock_guard lock(dbAdapterMutex_); if (kvStoragePtr_ == nullptr) { DHLOGE("kvStoragePtr_ is null"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_POINTER_NULL; } DistributedKv::Key kvKey(key); DistributedKv::Status status = kvStoragePtr_->Delete(kvKey); if (status != DistributedKv::Status::SUCCESS) { DHLOGE("Remove data by key failed"); return ERR_DH_FWK_RESOURCE_KV_STORAGE_OPERATION_FAIL; } DHLOGD("Remove data by key success"); return DH_FWK_SUCCESS; } } }