OpenHarmony-v6.0-Release/s

/*
 * Copyright (c) 2023 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 "dsoftbus_adapter_impl.h"

#ifdef ENABLE_PERFORMANCE_CHECK
#include <chrono>
#endif // ENABLE_PERFORMANCE_CHECK

#include <netinet/in.h>
#include <netinet/tcp.h>
#include "device_manager.h"
#include "dfs_session.h"
#include "securec.h"
#include "softbus_bus_center.h"
#include "softbus_error_code.h"

#include "devicestatus_define.h"
#include "i_ddm_adapter.h"
#include "utility.h"
#include "inner_socket.h"

#undef LOG_TAG
#define LOG_TAG "DSoftbusAdapterImpl"

namespace OHOS {
namespace Msdp {
namespace DeviceStatus {
namespace {
#define SERVER_SESSION_NAME "ohos.msdp.device_status.intention.serversession"
#define D_DEV_MGR DistributedHardware::DeviceManager::GetInstance()
const std::string CLIENT_SESSION_NAME { "ohos.msdp.device_status.intention.clientsession." };
constexpr size_t BIND_STRING_LENGTH { 15 };
constexpr size_t DEVICE_NAME_SIZE_MAX { 256 };
constexpr size_t PKG_NAME_SIZE_MAX { 65 };
constexpr int32_t MIN_BW { 80 * 1024 * 1024 };
constexpr int32_t LATENCY { 3000 };
constexpr int32_t SOCKET_SERVER { 0 };
constexpr int32_t SOCKET_CLIENT { 1 };
constexpr int32_t INVALID_SOCKET { -1 };
constexpr int32_t HEART_BEAT_INTERVAL_MS { 64 };
constexpr int32_t HEART_BEAT_SIZE_BYTE { 28 }; // Ensure size of heartBeat packet is 64Bytes.
const std::string HEART_BEAT_THREAD_NAME { "OS_Cooperate_Heart_Beat" };
const char* PARAM_KEY_OS_TYPE = "OS_TYPE";
constexpr int32_t OS_TYPE_OH { 10 };
constexpr int32_t OPT_TYPE_FLOW_INFO { 10005 };
}

std::mutex DSoftbusAdapterImpl::mutex_;
std::shared_ptr<DSoftbusAdapterImpl> DSoftbusAdapterImpl::instance_;

std::shared_ptr<DSoftbusAdapterImpl> DSoftbusAdapterImpl::GetInstance()
{
    if (instance_ == nullptr) {
        std::lock_guard<std::mutex> lock(mutex_);
        if (instance_ == nullptr) {
            instance_ = std::make_shared<DSoftbusAdapterImpl>();
        }
    }
    return instance_;
}

void DSoftbusAdapterImpl::DestroyInstance()
{
    std::lock_guard<std::mutex> lock(mutex_);
    instance_.reset();
}

DSoftbusAdapterImpl::~DSoftbusAdapterImpl()
{
    Disable();
}

int32_t DSoftbusAdapterImpl::Enable()
{
    CALL_DEBUG_ENTER;
    InitHeartBeat();
    std::unique_lock<std::shared_mutex> lock(lock_);
    return SetupServer();
}

void DSoftbusAdapterImpl::Disable()
{
    CALL_DEBUG_ENTER;
    std::unique_lock<std::shared_mutex> lock(lock_);
    ShutdownServer();
}

void DSoftbusAdapterImpl::AddObserver(std::shared_ptr<IDSoftbusObserver> observer)
{
    CALL_DEBUG_ENTER;
    std::unique_lock<std::shared_mutex> lock(lock_);
    CHKPV(observer);
    observers_.erase(Observer());
    observers_.emplace(observer);
}

void DSoftbusAdapterImpl::RemoveObserver(std::shared_ptr<IDSoftbusObserver> observer)
{
    CALL_DEBUG_ENTER;
    std::unique_lock<std::shared_mutex> lock(lock_);
    if (auto iter = observers_.find(Observer(observer)); iter != observers_.end()) {
        observers_.erase(iter);
    }
    observers_.erase(Observer());
}

int32_t DSoftbusAdapterImpl::CheckDeviceOnline(const std::string &networkId)
{
    CALL_DEBUG_ENTER;
    std::vector<DistributedHardware::DmDeviceInfo> deviceList;
    if (D_DEV_MGR.GetTrustedDeviceList(FI_PKG_NAME, "", deviceList) != RET_OK) {
        FI_HILOGE("GetTrustedDeviceList failed");
        return RET_ERR;
    }
    if (deviceList.empty()) {
        FI_HILOGE("Trust device list size is invalid");
        return RET_ERR;
    }
    for (const auto &deviceInfo : deviceList) {
        if (std::string(deviceInfo.networkId) == networkId) {
            return RET_OK;
        }
    }
    return RET_ERR;
}

int32_t DSoftbusAdapterImpl::OpenSession(const std::string &networkId)
{
    CALL_DEBUG_ENTER;
    std::unique_lock<std::shared_mutex> lock(lock_);
#ifdef ENABLE_PERFORMANCE_CHECK
    auto startStamp = std::chrono::steady_clock::now();
#endif // ENABLE_PERFORMANCE_CHECK
    int32_t ret = OpenSessionLocked(networkId);
#ifdef ENABLE_PERFORMANCE_CHECK
    auto openSessionDuration = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - startStamp).count();
    FI_HILOGI("[PERF] OpenSessionLocked ret:%{public}d, elapsed: %{public}lld ms", ret, openSessionDuration);
#endif // ENABLE_PERFORMANCE_CHECK
    return ret;
}

void DSoftbusAdapterImpl::CloseSession(const std::string &networkId)
{
    CALL_INFO_TRACE;
    std::unique_lock<std::shared_mutex> lock(lock_);
    if (auto iter = sessions_.find(networkId); iter != sessions_.end()) {
        ::Shutdown(iter->second.socket_);
        sessions_.erase(iter);
        FI_HILOGI("Shutdown session(%{public}d, %{public}s)", iter->second.socket_,
            Utility::Anonymize(networkId).c_str());
    }
}

void DSoftbusAdapterImpl::CloseAllSessions()
{
    CALL_INFO_TRACE;
    std::unique_lock<std::shared_mutex> lock(lock_);
    CloseAllSessionsLocked();
}

int32_t DSoftbusAdapterImpl::FindConnection(const std::string &networkId)
{
    CALL_DEBUG_ENTER;
    auto iter = sessions_.find(networkId);
    return (iter != sessions_.end() ? iter->second.socket_ : -1);
}

int32_t DSoftbusAdapterImpl::SendPacket(const std::string &networkId, NetPacket &packet)
{
    CALL_DEBUG_ENTER;
    std::shared_lock<std::shared_mutex> lock(lock_);
    int32_t socket = FindConnection(networkId);
    if (socket < 0) {
        FI_HILOGE("Node \'%{public}s\' is not connected", Utility::Anonymize(networkId).c_str());
        return RET_ERR;
    }
    StreamBuffer buffer;
    if (!packet.MakeData(buffer)) {
        FI_HILOGE("Failed to buffer packet");
        return RET_ERR;
    }
    if (buffer.Size() > MAX_PACKET_BUF_SIZE) {
        FI_HILOGE("Packet is too large");
        return RET_ERR;
    }
    int32_t ret = ::SendBytes(socket, buffer.Data(), buffer.Size());
    if (ret != SOFTBUS_OK) {
        FI_HILOGE("DSOFTBUS::SendBytes fail (%{public}d)", ret);
        return RET_ERR;
    }
    return RET_OK;
}

int32_t DSoftbusAdapterImpl::SendParcel(const std::string &networkId, Parcel &parcel)
{
    CALL_DEBUG_ENTER;
    std::shared_lock<std::shared_mutex> lock(lock_);
    int32_t socket = FindConnection(networkId);
    if (socket < 0) {
        FI_HILOGE("Node \'%{public}s\' is not connected", Utility::Anonymize(networkId).c_str());
        return RET_ERR;
    }
    int32_t ret = ::SendBytes(socket, reinterpret_cast<const void*>(parcel.GetData()), parcel.GetDataSize());
    if (ret != SOFTBUS_OK) {
        FI_HILOGE("DSOFTBUS::SendBytes fail, error:%{public}d", ret);
        return RET_ERR;
    }
    return RET_OK;
}

int32_t DSoftbusAdapterImpl::BroadcastPacket(NetPacket &packet)
{
    CALL_INFO_TRACE;
    std::shared_lock<std::shared_mutex> lock(lock_);
    if (sessions_.empty()) {
        FI_HILOGE("No session connected");
        return RET_ERR;
    }
    StreamBuffer buffer;
    if (!packet.MakeData(buffer)) {
        FI_HILOGE("Failed to buffer packet");
        return RET_ERR;
    }
    if (buffer.Size() > MAX_PACKET_BUF_SIZE) {
        FI_HILOGE("Packet is too large");
        return RET_ERR;
    }
    for (const auto &elem : sessions_) {
        int32_t socket = elem.second.socket_;
        if (socket < 0) {
            FI_HILOGE("Node \'%{public}s\' is not connected", Utility::Anonymize(elem.first).c_str());
            continue;
        }
        if (int32_t ret = ::SendBytes(socket, buffer.Data(), buffer.Size()); ret != SOFTBUS_OK) {
            FI_HILOGE("DSOFTBUS::SendBytes fail (%{public}d)", ret);
            continue;
        }
        FI_HILOGI("BroadcastPacket to networkId:%{public}s success", Utility::Anonymize(elem.first).c_str());
    }
    return RET_OK;
}

bool DSoftbusAdapterImpl::HasSessionExisted(const std::string &networkId)
{
    CALL_DEBUG_ENTER;
    auto iter = sessions_.find(networkId);
    return (iter != sessions_.end() && iter->second.socket_ != INVALID_SOCKET);
}

static void OnBindLink(int32_t socket, PeerSocketInfo info)
{
    DSoftbusAdapterImpl::GetInstance()->OnBind(socket, info);
}

static void OnShutdownLink(int32_t socket, ShutdownReason reason)
{
    DSoftbusAdapterImpl::GetInstance()->OnShutdown(socket, reason);
}

static void OnBytesAvailable(int32_t socket, const void *data, uint32_t dataLen)
{
    DSoftbusAdapterImpl::GetInstance()->OnBytes(socket, data, dataLen);
}

void DSoftbusAdapterImpl::OnBind(int32_t socket, PeerSocketInfo info)
{
    CALL_INFO_TRACE;
    std::unique_lock<std::shared_mutex> lock(lock_);
    std::string networkId = info.networkId;
    FI_HILOGI("Bind session(%{public}d, %{public}s)", socket, Utility::Anonymize(networkId).c_str());
    if (!CheckDeviceOsType(networkId)) {
        FI_HILOGE("Refuse bind");
        ::Shutdown(socket);
    }
    if (auto iter = sessions_.find(networkId); iter != sessions_.cend()) {
        if (iter->second.socket_ == socket) {
            FI_HILOGI("(%{public}d, %{public}s) has bound", iter->second.socket_,
                Utility::Anonymize(networkId).c_str());
            return;
        }
        FI_HILOGI("(%{public}d, %{public}s) need erase", iter->second.socket_, Utility::Anonymize(networkId).c_str());
        sessions_.erase(iter);
    }
    ConfigTcpAlive(socket);
    sessions_.emplace(networkId, Session(socket));

    for (const auto &item : observers_) {
        std::shared_ptr<IDSoftbusObserver> observer = item.Lock();
        if (observer != nullptr) {
            FI_HILOGD("Notify binding (%{public}d, %{public}s)", socket, Utility::Anonymize(networkId).c_str());
            observer->OnBind(networkId);
        }
    }
}

void DSoftbusAdapterImpl::OnShutdown(int32_t socket, ShutdownReason reason)
{
    CALL_INFO_TRACE;
    std::unique_lock<std::shared_mutex> lock(lock_);
    auto iter = std::find_if(sessions_.cbegin(), sessions_.cend(),
        [socket](const auto &item) {
            return (item.second.socket_ == socket);
        });
    if (iter == sessions_.cend()) {
        FI_HILOGD("Session(%{public}d) is not bound", socket);
        return;
    }
    std::string networkId = iter->first;
    sessions_.erase(iter);
    FI_HILOGI("Shutdown session(%{public}d, %{public}s)", socket, Utility::Anonymize(networkId).c_str());

    for (const auto &item : observers_) {
        std::shared_ptr<IDSoftbusObserver> observer = item.Lock();
        if (observer != nullptr) {
            FI_HILOGD("Notify shutdown of session(%{public}d, %{public}s)",
                socket, Utility::Anonymize(networkId).c_str());
            observer->OnShutdown(networkId);
        }
    }
}

void DSoftbusAdapterImpl::OnBytes(int32_t socket, const void *data, uint32_t dataLen)
{
    CALL_DEBUG_ENTER;
    std::shared_lock<std::shared_mutex> lock(lock_);
    auto iter = std::find_if(sessions_.begin(), sessions_.end(),
        [socket](const auto &item) {
            return (item.second.socket_ == socket);
        });
    if (iter == sessions_.end()) {
        FI_HILOGE("Invalid socket: %{public}d", socket);
        return;
    }
    const std::string networkId = iter->first;

    if (*reinterpret_cast<const uint32_t*>(data) < static_cast<uint32_t>(MessageId::MAX_MESSAGE_ID)) {
        CircleStreamBuffer &circleBuffer = iter->second.buffer_;

        if (!circleBuffer.Write(reinterpret_cast<const char*>(data), dataLen)) {
            FI_HILOGE("Failed to write buffer");
        }
        HandleSessionData(networkId, circleBuffer);
    } else {
        HandleRawData(networkId, data, dataLen);
    }
}

int32_t DSoftbusAdapterImpl::InitSocket(SocketInfo info, int32_t socketType, int32_t &socket)
{
    CALL_INFO_TRACE;
    socket = ::Socket(info);
    if (socket < 0) {
        FI_HILOGE("DSOFTBUS::Socket failed");
        return RET_ERR;
    }
    QosTV socketQos[] {
        { .qos = QOS_TYPE_MIN_BW, .value = MIN_BW },
        { .qos = QOS_TYPE_MAX_LATENCY, .value = LATENCY },
        { .qos = QOS_TYPE_MIN_LATENCY, .value = LATENCY },
    };
    ISocketListener listener {
        .OnBind = OnBindLink,
        .OnShutdown = OnShutdownLink,
        .OnBytes = OnBytesAvailable,
    };
    int32_t ret { -1 };

    if (socketType == SOCKET_SERVER) {
        ret = ::Listen(socket, socketQos, sizeof(socketQos) / sizeof(socketQos[0]), &listener);
        if (ret != 0) {
            FI_HILOGE("DSOFTBUS::Listen failed");
        }
    } else if (socketType == SOCKET_CLIENT) {
        SetSocketOpt(socket);
        ret = ::Bind(socket, socketQos, sizeof(socketQos) / sizeof(socketQos[0]), &listener);
        if (ret != 0) {
            FI_HILOGE("DSOFTBUS::Bind failed");
        }
    }
    if (ret != 0) {
        ::Shutdown(socket);
        socket = -1;
        return ret;
    }
    return RET_OK;
}

void DSoftbusAdapterImpl::SetSocketOpt(int32_t socket)
{
    CALL_INFO_TRACE;
    if (socket < 0) {
        FI_HILOGE("DSOFTBUS::Socket failed");
        return;
    }
    TransFlowInfo transInfo = {
        .flowSize = 0,
        .sessionType = SHORT_FOREGROUND_SESSION,
        .flowQosType = LOW_LATENCY_10MS,
    };
    if (int32_t ret = ::SetSocketOpt(socket, OPT_LEVEL_SOFTBUS, static_cast<OptType>(OPT_TYPE_FLOW_INFO),
        static_cast<void *>(&transInfo), sizeof(TransFlowInfo)); ret != RET_OK) {
        FI_HILOGE("DSOFTBUS::SetSocketOpt failed, ret:%{public}d", ret);
    }
}

int32_t DSoftbusAdapterImpl::SetupServer()
{
    CALL_INFO_TRACE;
    if (socketFd_ > 0) {
        return RET_OK;
    }
    char name[DEVICE_NAME_SIZE_MAX] { SERVER_SESSION_NAME };
    char pkgName[PKG_NAME_SIZE_MAX] { FI_PKG_NAME };
    FI_HILOGI("Server session name: \'%{public}s\'", name);
    FI_HILOGI("Package name: \'%{public}s\'", pkgName);
    SocketInfo info {
        .name = name,
        .pkgName = pkgName,
        .dataType = DATA_TYPE_BYTES
    };
    int32_t ret = InitSocket(info, SOCKET_SERVER, socketFd_);
    if (ret != RET_OK) {
        FI_HILOGE("Failed to setup server");
        return ret;
    }
    return RET_OK;
}

void DSoftbusAdapterImpl::ShutdownServer()
{
    CALL_INFO_TRACE;
    CloseAllSessionsLocked();
    if (socketFd_ > 0) {
        ::Shutdown(socketFd_);
        socketFd_ = -1;
    }
}

int32_t DSoftbusAdapterImpl::OpenSessionLocked(const std::string &networkId)
{
    CALL_DEBUG_ENTER;
    if (sessions_.find(networkId) != sessions_.end()) {
        FI_HILOGD("InputSoftbus session has already opened");
        return RET_OK;
    }
    std::string sessionName = CLIENT_SESSION_NAME + networkId.substr(0, BIND_STRING_LENGTH);
    char name[DEVICE_NAME_SIZE_MAX] {};
    if (strcpy_s(name, sizeof(name), sessionName.c_str()) != EOK) {
        FI_HILOGE("Invalid name:%{public}s", sessionName.c_str());
        return RET_ERR;
    }
    char peerName[DEVICE_NAME_SIZE_MAX] { SERVER_SESSION_NAME };
    char peerNetworkId[PKG_NAME_SIZE_MAX] {};
    if (strcpy_s(peerNetworkId, sizeof(peerNetworkId), networkId.c_str()) != EOK) {
        FI_HILOGE("Invalid peerNetworkId:%{public}s", Utility::Anonymize(networkId).c_str());
        return RET_ERR;
    }
    char pkgName[PKG_NAME_SIZE_MAX] { FI_PKG_NAME };
    FI_HILOGI("Client session name: \'%{public}s\'", name);
    FI_HILOGI("Peer name: \'%{public}s\'", peerName);
    FI_HILOGI("Peer network id: \'%{public}s\'", Utility::Anonymize(peerNetworkId).c_str());
    FI_HILOGI("Package name: \'%{public}s\'", pkgName);
    SocketInfo info {
        .name = name,
        .peerName = peerName,
        .peerNetworkId = peerNetworkId,
        .pkgName = pkgName,
        .dataType = DATA_TYPE_BYTES
    };
    int32_t socket { -1 };

    int32_t ret = InitSocket(info, SOCKET_CLIENT, socket);
    if (ret != RET_OK) {
        FI_HILOGE("Failed to bind %{public}s", Utility::Anonymize(networkId).c_str());
        return ret;
    }
    ConfigTcpAlive(socket);
    FI_HILOGI("Connected to (%{public}s,%{public}d)", Utility::Anonymize(networkId).c_str(), socket);
    sessions_.emplace(networkId, Session(socket));
    OnConnectedLocked(networkId);
    return RET_OK;
}

void DSoftbusAdapterImpl::OnConnectedLocked(const std::string &networkId)
{
    CALL_INFO_TRACE;
    for (const auto &item : observers_) {
        std::shared_ptr<IDSoftbusObserver> observer = item.Lock();
        CHKPC(observer);
        FI_HILOGI("Notify connected to networkId:%{public}s", Utility::Anonymize(networkId).c_str());
        observer->OnConnected(networkId);
    }
}

void DSoftbusAdapterImpl::CloseAllSessionsLocked()
{
    std::for_each(sessions_.begin(), sessions_.end(), [](const auto &item) {
        ::Shutdown(item.second.socket_);
        FI_HILOGI("Shutdown connection with (%{public}s,%{public}d)",
            Utility::Anonymize(item.first).c_str(), item.second.socket_);
    });
    sessions_.clear();
}

void DSoftbusAdapterImpl::ConfigTcpAlive(int32_t socket)
{
    CALL_DEBUG_ENTER;
    if (socket < 0) {
        FI_HILOGW("Config tcp alive, invalid sessionId");
        return;
    }
    int32_t handle { -1 };
    int32_t result = GetSessionHandle(socket, &handle);
    if (result != RET_OK) {
        FI_HILOGE("Failed to get the session handle, socketId:%{public}d, handle:%{public}d", socket, handle);
        return;
    }
    int32_t keepAliveTimeout { 10 };
    result = setsockopt(handle, IPPROTO_TCP, TCP_KEEPIDLE, &keepAliveTimeout, sizeof(keepAliveTimeout));
    if (result != RET_OK) {
        FI_HILOGE("Config tcp alive, setsockopt set idle failed, result:%{public}d", result);
        return;
    }
    int32_t keepAliveCount { 5 };
    result = setsockopt(handle, IPPROTO_TCP, TCP_KEEPCNT, &keepAliveCount, sizeof(keepAliveCount));
    if (result != RET_OK) {
        FI_HILOGE("Config tcp alive, setsockopt set cnt failed");
        return;
    }
    int32_t interval { 1 };
    result = setsockopt(handle, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(interval));
    if (result != RET_OK) {
        FI_HILOGE("Config tcp alive, setsockopt set intvl failed");
        return;
    }
    int32_t enable { 1 };
    result = setsockopt(handle, SOL_SOCKET, SO_KEEPALIVE, &enable, sizeof(enable));
    if (result != RET_OK) {
        FI_HILOGE("Config tcp alive, setsockopt enable alive failed");
        return;
    }
    int32_t TimeoutMs { 15000 };
    result = setsockopt(handle, IPPROTO_TCP, TCP_USER_TIMEOUT, &TimeoutMs, sizeof(TimeoutMs));
    if (result != RET_OK) {
        FI_HILOGE("Failed to enable setsockopt for timeout, %{public}d", result);
        return;
    }
}

void DSoftbusAdapterImpl::HandleSessionData(const std::string &networkId, CircleStreamBuffer &circleBuffer)
{
    CALL_DEBUG_ENTER;
    while (circleBuffer.ResidualSize() >= static_cast<int32_t>(sizeof(PackHead))) {
        const char *buf = circleBuffer.ReadBuf();
        const PackHead *head = reinterpret_cast<const PackHead *>(buf);

        if ((head->size < 0) || (static_cast<size_t>(head->size) > MAX_PACKET_BUF_SIZE)) {
            FI_HILOGE("Corrupted net packet");
            break;
        }
        if ((head->size + static_cast<int32_t>(sizeof(PackHead))) > circleBuffer.ResidualSize()) {
            FI_HILOGI("Incomplete package, package size:%{public}d, residual size:%{public}d",
                (head->size + static_cast<int32_t>(sizeof(PackHead))), circleBuffer.ResidualSize());
            break;
        }
        NetPacket packet(head->idMsg);

        if ((head->size > 0) && !packet.Write(&buf[sizeof(PackHead)], head->size)) {
            FI_HILOGE("Failed to fill packet, PacketSize:%{public}d", head->size);
            break;
        }
        circleBuffer.SeekReadPos(packet.GetPacketLength());
        HandlePacket(networkId, packet);
    }
}

void DSoftbusAdapterImpl::HandlePacket(const std::string &networkId, NetPacket &packet)
{
    CALL_DEBUG_ENTER;
    for (const auto &item : observers_) {
        std::shared_ptr<IDSoftbusObserver> observer = item.Lock();
        if ((observer != nullptr) &&
            observer->OnPacket(networkId, packet)) {
            return;
        }
    }
}

void DSoftbusAdapterImpl::HandleRawData(const std::string &networkId, const void *data, uint32_t dataLen)
{
    CALL_DEBUG_ENTER;
    for (const auto &item : observers_) {
        std::shared_ptr<IDSoftbusObserver> observer = item.Lock();
        if ((observer != nullptr) &&
            observer->OnRawData(networkId, data, dataLen)) {
            return;
        }
    }
}

void DSoftbusAdapterImpl::InitHeartBeat()
{
    auto runner = AppExecFwk::EventRunner::Create(HEART_BEAT_THREAD_NAME, AppExecFwk::ThreadMode::FFRT);
    CHKPV(runner);
    eventHandler_ = std::make_shared<AppExecFwk::EventHandler>(runner);
    char heartBeatContent[HEART_BEAT_SIZE_BYTE] { 'a' };
    heartBeatPacket_.Write(heartBeatContent, HEART_BEAT_SIZE_BYTE);
}

void DSoftbusAdapterImpl::StartHeartBeat(const std::string &networkId)
{
    CALL_INFO_TRACE;
    if (GetHeartBeatState(networkId)) {
        FI_HILOGI("HeartBeat to %{public}s running ready", Utility::Anonymize(networkId).c_str());
        return;
    }
    if (KeepHeartBeating(networkId) != RET_OK) {
        UpdateHeartBeatState(networkId, false);
        FI_HILOGE("StartHeartBeat to %{public}s failed", Utility::Anonymize(networkId).c_str());
        return;
    }
    UpdateHeartBeatState(networkId, true);
    FI_HILOGI("StartHeartBeat to %{public}s successfully", Utility::Anonymize(networkId).c_str());
}

void DSoftbusAdapterImpl::StopHeartBeat(const std::string &networkId)
{
    UpdateHeartBeatState(networkId, false);
    FI_HILOGI("StopHeartBeat to %{public}s successfully", Utility::Anonymize(networkId).c_str());
}

int32_t DSoftbusAdapterImpl::KeepHeartBeating(const std::string &networkId)
{
    if (SendPacket(networkId, heartBeatPacket_) != RET_OK) {
        FI_HILOGE("HeartBeat to %{public}s failed, stop it", Utility::Anonymize(networkId).c_str());
        UpdateHeartBeatState(networkId, false);
        return RET_ERR;
    }
    CHKPR(eventHandler_, RET_ERR);
    if (!eventHandler_->PostTask(
        [this, networkId]() {
            if (GetHeartBeatState(networkId)) {
                this->KeepHeartBeating(networkId);
            } else {
                UpdateHeartBeatState(networkId, false);
                FI_HILOGE("Switch off, Stop heartBeat to %{public}s", Utility::Anonymize(networkId).c_str());
            }
        }, HEART_BEAT_INTERVAL_MS)) {
        FI_HILOGE("PostTask heartBeat to %{public}s failed", Utility::Anonymize(networkId).c_str());
        UpdateHeartBeatState(networkId, false);
        return RET_ERR;
    }
    return RET_OK;
}

void DSoftbusAdapterImpl::UpdateHeartBeatState(const std::string &networkId, bool state)
{
    std::unique_lock<std::shared_mutex> lock(heartBeatLock_);
    heartBeatStates_[networkId] = state;
    FI_HILOGI("Update %{public}s state:%{public}s", Utility::Anonymize(networkId).c_str(), state ? "true" : "false");
}

bool DSoftbusAdapterImpl::GetHeartBeatState(const std::string &networkId)
{
    std::shared_lock<std::shared_mutex> lock(heartBeatLock_);
    if (heartBeatStates_.find(networkId) != heartBeatStates_.end()) {
        return heartBeatStates_[networkId];
    }
    return false;
}

bool DSoftbusAdapterImpl::CheckDeviceOsType(const std::string &networkId)
{
    CALL_INFO_TRACE;
    DistributedHardware::DmDeviceInfo deviceInfo;
    int32_t res = D_DEV_MGR.GetDeviceInfo(FI_PKG_NAME, networkId, deviceInfo);
    if (res != ERR_OK) {
        FI_HILOGE("Get device failed, res:%{public}d", res);
        return false;
    }
    if (deviceInfo.extraData.empty()) {
        FI_HILOGE("Deviceinfo extradata is empty");
        return false;
    }
    JsonParser extraData(deviceInfo.extraData.c_str());
    if (!cJSON_IsObject(extraData.Get())) {
        FI_HILOGE("extraData is not json object");
        return false;
    }
    cJSON *osType = cJSON_GetObjectItemCaseSensitive(extraData.Get(), PARAM_KEY_OS_TYPE);
    if (cJSON_IsNumber(osType)) {
        if (osType->valueint != OS_TYPE_OH) {
            FI_HILOGE("Ostype:%{public}d", osType->valueint);
            return false;
        }
    } else {
        FI_HILOGE("get ostype error, extraData:%{public}s", deviceInfo.extraData.c_str());
        return false;
    }
    return true;
}
} // namespace DeviceStatus
} // namespace Msdp
} // namespace OHOS