diff --git a/services/sink/transport/BUILD.gn b/services/sink/transport/BUILD.gn new file mode 100644 index 0000000..c3f2cf0 --- /dev/null +++ b/services/sink/transport/BUILD.gn @@ -0,0 +1,64 @@ +# 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. + +import("//build/ohos.gni") +import("//foundation/distributedhardware/distributedinput/distributedinput.gni") + +ohos_shared_library("libdinput_sink_trans") { + include_dirs = [ + "include", + "${common_path}/include", + "//utils/native/base/include", + "//utils/system/safwk/native/include", + "${fwk_common_path}/log/include", + "${fwk_common_path}/utils/include", + "${fwk_utils_path}/include/log", + "${fwk_utils_path}/include", + "//third_party/json/include", + "//foundation/communication/dsoftbus/interfaces/kits/transport", + "${service_common}/include", + ] + + sources = [ + "src/distributed_input_sink_transport.cpp", + "src/distributed_input_sink_switch.cpp" + ] + + defines = [ + "HI_LOG_ENABLE", + "DH_LOG_TAG=\"distributedinputsinktransport\"", + "LOG_DOMAIN=0xD004100", + ] + + deps = [ + "//foundation/appexecfwk/standard/interfaces/innerkits/libeventhandler:libeventhandler", + "//foundation/distributedschedule/samgr/interfaces/innerkits/samgr_proxy:samgr_proxy", + "//utils/native/base:utils", + "${fwk_utils_path}:distributedhardwareutils", + "//foundation/communication/dsoftbus/sdk:softbus_client", + ] + + external_deps = [ + "ipc:ipc_core", + "safwk:system_ability_fwk", + "hiviewdfx_hilog_native:libhilog", + "samgr_standard:samgr_proxy", + ] + + cflags_cc = [ "-DHILOG_ENABLE" ] + + subsystem_name = "distributedhardware" + + part_name = "distributed_input" + +} diff --git a/services/sink/transport/include/distributed_input_sink_switch.h b/services/sink/transport/include/distributed_input_sink_switch.h new file mode 100644 index 0000000..509d4df --- /dev/null +++ b/services/sink/transport/include/distributed_input_sink_switch.h @@ -0,0 +1,52 @@ +/* + * Copyright (c) 2022 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. + */ + +#ifndef DISTRIBUTED_INPUT_SINK_SWITCH_H +#define DISTRIBUTED_INPUT_SINK_SWITCH_H + +#include +#include +#include +#include +#include "dinput_softbus_define.h" + +namespace OHOS { +namespace DistributedHardware { +namespace DistributedInput { +class DistributedInputSinkSwitch { +public: + static DistributedInputSinkSwitch &GetInstance(); + DistributedInputSinkSwitch(); + ~DistributedInputSinkSwitch(); + void InitSwitch(); + + int32_t StartSwitch(int32_t sessionId); + void StopSwitch(int32_t sessionId); + void AddSession(int32_t sessionId); + void RemoveSession(int32_t sessionId); + void StopAllSwitch(); + std::vector GetAllSessionId(); + int32_t GetSwitchOpenedSession(); + +private: + + std::mutex operationMutex_; + std::vector switchVector_; +}; +} // namespace DistributedInput +} // namespace DistributedHardware +} // namespace OHOS + +#endif // DISTRIBUTED_INPUT_SINK_SWITCH_H diff --git a/services/sink/transport/include/distributed_input_sink_transport.h b/services/sink/transport/include/distributed_input_sink_transport.h new file mode 100644 index 0000000..fead35b --- /dev/null +++ b/services/sink/transport/include/distributed_input_sink_transport.h @@ -0,0 +1,84 @@ +/* + * 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. + */ + +#ifndef DISTRIBUTED_INPUT_SINK_TRANSPORT_H +#define DISTRIBUTED_INPUT_SINK_TRANSPORT_H + +#include +#include +#include +#include + +#include "nlohmann/json.hpp" +#include "constants.h" +#include "event_handler.h" +#include "dinput_sink_trans_callback.h" +#include "dinput_softbus_define.h" +#include "distributed_input_sink_switch.h" + +namespace OHOS { +namespace DistributedHardware { +namespace DistributedInput { +class DistributedInputSinkTransport { +public: + static DistributedInputSinkTransport &GetInstance(); + DistributedInputSinkTransport(); + ~DistributedInputSinkTransport(); + + int32_t Init(); + + void RegistSinkRespCallback(std::shared_ptr callback); + int32_t RespPrepareRemoteInput(const int32_t sessionId, std::string &smsg); + int32_t RespUnprepareRemoteInput(const int32_t sessionId, std::string &smsg); + int32_t RespStartRemoteInput(const int32_t sessionId, std::string &smsg); + int32_t RespStopRemoteInput(const int32_t sessionId, std::string &smsg); + + int32_t OnSessionOpened(int32_t sessionId, int32_t result); + void OnSessionClosed(int32_t sessionId); + void OnBytesReceived(int32_t sessionId, const void *data, uint32_t dataLen); + + class DInputSinkEventHandler : public AppExecFwk::EventHandler { + public: + DInputSinkEventHandler(const std::shared_ptr &runner); + ~DInputSinkEventHandler() {} + + void ProcessEvent(const AppExecFwk::InnerEvent::Pointer &event) override; + }; + + std::shared_ptr GetEventHandler(); + void CloseAllSession(); + +private: + int32_t SendMessage(int32_t sessionId, std::string &message); + void HandleSessionData(int32_t sessionId, const std::string& messageData); + void NotifyPrepareRemoteInput(int32_t sessionId, const nlohmann::json &recMsg); + void NotifyUnprepareRemoteInput(int32_t sessionId, const nlohmann::json &recMsg); + void NotifyStartRemoteInput(int32_t sessionId, const nlohmann::json &recMsg); + void NotifyStopRemoteInput(int32_t sessionId, const nlohmann::json &recMsg); + +private: + std::string deviceId_; + std::set sessionIdSet_; + std::map sessionDevMap_; + std::string mySessionName_; + + std::shared_ptr eventHandler_; + std::shared_ptr callback_; +}; +} // namespace DistributedInput +} // namespace DistributedHardware +} // namespace OHOS + +#endif // DISTRIBUTED_INPUT_SINK_TRANSPORT_H diff --git a/services/sink/transport/src/distributed_input_sink_switch.cpp b/services/sink/transport/src/distributed_input_sink_switch.cpp new file mode 100644 index 0000000..c55fd6a --- /dev/null +++ b/services/sink/transport/src/distributed_input_sink_switch.cpp @@ -0,0 +1,190 @@ +/* + * Copyright (c) 2022 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 "distributed_input_sink_switch.h" + +#include "constants_dinput.h" +#include "distributed_hardware_log.h" + +namespace OHOS { +namespace DistributedHardware { +namespace DistributedInput { +DistributedInputSinkSwitch &DistributedInputSinkSwitch::GetInstance() +{ + static DistributedInputSinkSwitch instance; + return instance; +} + +DistributedInputSinkSwitch::DistributedInputSinkSwitch() +{ + InitSwitch(); +} + +DistributedInputSinkSwitch::~DistributedInputSinkSwitch() +{ + DHLOGI("~DistributedInputSinkSwitch()"); + InitSwitch(); +} + +void DistributedInputSinkSwitch::InitSwitch() +{ + DHLOGI("InitSwitch."); + std::unique_lock switchLock(operationMutex_); + switchVector_.clear(); +} + +int32_t DistributedInputSinkSwitch::StartSwitch(int32_t sessionId) +{ + std::unique_lock switchLock(operationMutex_); + if (switchVector_.empty()) { + DHLOGE("StartSwitch sessionId:%d fail,switchVector_ is null.", sessionId); + return FAILURE; + } else { + bool findOld = false; + for (std::vector::iterator it = switchVector_.begin(); it < switchVector_.end(); it++) { + if ((*it).sessionId == sessionId) { + (*it).switchState = true; + findOld = true; + } else { + (*it).switchState = false; + } + } + + if (findOld) { + DHLOGI("StartSwitch sessionId:%d is find.", sessionId); + return SUCCESS; + } else { + DHLOGE("StartSwitch sessionId:%d fail, not found.", sessionId); + return FAILURE; + } + } +} + +void DistributedInputSinkSwitch::StopSwitch(int32_t sessionId) +{ + std::unique_lock switchLock(operationMutex_); + if (switchVector_.empty()) { + DHLOGE("StopSwitch sessionId:%d fail,switchVector_ is null.", sessionId); + } else { + bool findOld = false; + for (std::vector::iterator it = switchVector_.begin(); it < switchVector_.end(); it++) { + if ((*it).sessionId == sessionId) { + (*it).switchState = false; + findOld = true; + break; + } + } + + if (findOld) { + DHLOGI("StopSwitch sessionId:%d is success.", sessionId); + } else { + DHLOGE("StopSwitch sessionId:%d fail,not find it.", sessionId); + } + } +} + +void DistributedInputSinkSwitch::StopAllSwitch() +{ + std::unique_lock switchLock(operationMutex_); + if (switchVector_.empty()) { + DHLOGW("StopAllSwitch switchVector_ is null."); + } else { + for (std::vector::iterator it = switchVector_.begin(); it < switchVector_.end(); it++) { + (*it).switchState = false; + } + DHLOGI("StopAllSwitch success."); + } +} + +void DistributedInputSinkSwitch::AddSession(int32_t sessionId) +{ + std::unique_lock switchLock(operationMutex_); + if (switchVector_.empty()) { + SwitchStateData tmp(sessionId, false); + switchVector_.push_back(tmp); + DHLOGI("AddSession sessionId:%d add first.", sessionId); + } else { + bool findOld = false; + for (std::vector::iterator it = switchVector_.begin(); it < switchVector_.end(); it++) { + if ((*it).sessionId == sessionId) { + (*it).switchState = false; + findOld = true; + break; + } + } + + if (findOld) { + DHLOGI("AddSession sessionId:%d is find.", sessionId); + } else { + SwitchStateData tmp(sessionId, false); + switchVector_.push_back(tmp); + DHLOGI("AddSession sessionId:%d add new.", sessionId); + } + } +} + +void DistributedInputSinkSwitch::RemoveSession(int32_t sessionId) +{ + std::unique_lock switchLock(operationMutex_); + if (switchVector_.empty()) { + DHLOGE("RemoveSession sessionId:%d fail,switch_vector is null.", sessionId); + } else { + bool findOld = false; + for (std::vector::iterator it = switchVector_.begin(); it < switchVector_.end();) { + if ((*it).sessionId == sessionId) { + it = switchVector_.erase(it); + findOld = true; + break; + } else { + it++; + } + } + if (findOld) { + DHLOGI("RemoveSession sessionId:%d is success.", sessionId); + } else { + DHLOGE("RemoveSession sessionId:%d fail,not find it.", sessionId); + } + } +} + +std::vector DistributedInputSinkSwitch::GetAllSessionId() +{ + std::unique_lock switchLock(operationMutex_); + std::vector tmpVecSession; + for (std::vector::iterator it = switchVector_.begin(); it < switchVector_.end(); it++) { + tmpVecSession.push_back((*it).sessionId); + } + return tmpVecSession; +} + +// get current session which state is on, if error return -1. +int32_t DistributedInputSinkSwitch::GetSwitchOpenedSession() +{ + std::unique_lock switchLock(operationMutex_); + if (switchVector_.empty()) { + DHLOGE("GetSwitchOpenedSession error, no data."); + return FAILURE; + } + for (std::vector::iterator it = switchVector_.begin(); it < switchVector_.end(); it++) { + if ((*it).switchState == true) { + return (*it).sessionId; + } + } + DHLOGE("GetSwitchOpenedSession no session is open."); + return FAILURE; +} +} // namespace DistributedInput +} // namespace DistributedHardware +} // namespace OHOS diff --git a/services/sink/transport/src/distributed_input_sink_transport.cpp b/services/sink/transport/src/distributed_input_sink_transport.cpp new file mode 100644 index 0000000..c51c599 --- /dev/null +++ b/services/sink/transport/src/distributed_input_sink_transport.cpp @@ -0,0 +1,434 @@ +/* + * 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 "distributed_input_sink_transport.h" +#include "session.h" +#include "constants_dinput.h" +#include "securec.h" +#include "softbus_bus_center.h" +#include "distributed_hardware_log.h" + +namespace OHOS { +namespace DistributedHardware { +namespace DistributedInput { +static int32_t SessionOpened(int32_t sessionId, int32_t result) +{ + return DistributedInput::DistributedInputSinkTransport::GetInstance().OnSessionOpened(sessionId, result); +} + +static void SessionClosed(int32_t sessionId) +{ + DistributedInput::DistributedInputSinkTransport::GetInstance().OnSessionClosed(sessionId); +} + +static void BytesReceived(int32_t sessionId, const void *data, uint32_t dataLen) +{ + DistributedInput::DistributedInputSinkTransport::GetInstance().OnBytesReceived(sessionId, data, dataLen); +} + +static void MessageReceived(int32_t sessionId, const void *data, uint32_t dataLen) +{ + (void)sessionId; + (void)data; + (void)dataLen; + DHLOGI("sessionId:%d, dataLen:%d", sessionId, dataLen); +} + +static void StreamReceived(int32_t sessionId, const StreamData *data, const StreamData *ext, + const StreamFrameInfo *param) +{ + (void)sessionId; + (void)data; + (void)ext; + (void)param; + DHLOGI("sessionId:%d", sessionId); +} +DistributedInputSinkTransport &DistributedInputSinkTransport::GetInstance() +{ + static DistributedInputSinkTransport instance; + return instance; +} + +DistributedInputSinkTransport::DistributedInputSinkTransport() +{ + std::shared_ptr runner = AppExecFwk::EventRunner::Create(true); + eventHandler_ = std::make_shared(runner); + mySessionName_ = ""; + DHLOGI("DistributedInputSinkTransport eventHandler_"); +} + +DistributedInputSinkTransport::~DistributedInputSinkTransport() +{ + DHLOGI("~DistributedInputSinkTransport"); + sessionDevMap_.clear(); + (void)RemoveSessionServer(DH_FWK_PKG_NAME.c_str(), mySessionName_.c_str()); +} + +DistributedInputSinkTransport::DInputSinkEventHandler::DInputSinkEventHandler( + const std::shared_ptr &runner) : AppExecFwk::EventHandler(runner) +{ +} + +void DistributedInputSinkTransport::DInputSinkEventHandler::ProcessEvent(const AppExecFwk::InnerEvent::Pointer &event) +{ + DHLOGI("ProcessEvent"); + EHandlerMsgType eventId = static_cast(event->GetInnerEventId()); + switch (eventId) { + case EHandlerMsgType::DINPUT_SINK_EVENT_HANDLER_MSG: { + std::shared_ptr innerMsg = event->GetSharedObject(); + nlohmann::json sendMsg; + sendMsg[DINPUT_SOFTBUS_KEY_CMD_TYPE] = TRANS_SINK_MSG_BODY_DATA; + sendMsg[DINPUT_SOFTBUS_KEY_INPUT_DATA] = innerMsg->dump(); + std::string smsg = sendMsg.dump(); + + int32_t sessionId = DistributedInputSinkSwitch::GetInstance().GetSwitchOpenedSession(); + if (sessionId > 0) { + DistributedInputSinkTransport::GetInstance().SendMessage(sessionId, smsg); + } else { + DHLOGE("ProcessEvent can't send input data, because no session switch on."); + } + break; + } + default: + DHLOGE("ProcessEvent error, because eventId is unkonwn."); + break; + } +} + +int32_t DistributedInputSinkTransport::Init() +{ + // 1.create session + DHLOGI("Init"); + ISessionListener iSessionListener = { + .OnSessionOpened = SessionOpened, + .OnSessionClosed = SessionClosed, + .OnBytesReceived = BytesReceived, + .OnMessageReceived = MessageReceived, + .OnStreamReceived = StreamReceived + }; + auto localNode = std::make_unique(); + if (localNode == nullptr) { + DHLOGE("Init unique_ptr localNode nullptr."); + return FAILURE; + } + int32_t retCode = GetLocalNodeDeviceInfo(DH_FWK_PKG_NAME.c_str(), localNode.get()); + if (retCode != SUCCESS) { + DHLOGE("Init could not get local device id."); + return FAILURE; + } + std::string networkId = localNode->networkId; + DHLOGI("Init device networkId is %s", networkId.c_str()); + mySessionName_ = SESSION_NAME_SINK + networkId.substr(0, INTERCEPT_STRING_LENGTH); + + int32_t ret = CreateSessionServer(DH_FWK_PKG_NAME.c_str(), mySessionName_.c_str(), &iSessionListener); + if (ret != SUCCESS) { + DHLOGE("Init CreateSessionServer failed, error code %d.", ret); + return FAILURE; + } + return SUCCESS; +} + +std::shared_ptr DistributedInputSinkTransport::GetEventHandler() +{ + DHLOGI("GetEventHandler"); + return eventHandler_; +} + +void DistributedInputSinkTransport::RegistSinkRespCallback(std::shared_ptr callback) +{ + DHLOGI("RegistSinkRespCallback"); + callback_ = callback; +} + +int32_t DistributedInputSinkTransport::RespPrepareRemoteInput( + const int32_t sessionId, std::string &smsg) +{ + if (sessionId > 0) { + DHLOGI("RespPrepareRemoteInput session:%d, smsg:%s.", sessionId, smsg.c_str()); + int32_t ret = SendMessage(sessionId, smsg); + if (ret != SUCCESS) { + DHLOGE("RespPrepareRemoteInput error, SendMessage fail."); + return FAILURE; + } + return SUCCESS; + } else { + DHLOGE("RespPrepareRemoteInput error, sessionId <= 0."); + return FAILURE; + } +} + +int32_t DistributedInputSinkTransport::RespUnprepareRemoteInput( + const int32_t sessionId, std::string &smsg) +{ + if (sessionId > 0) { + DHLOGI("RespUnprepareRemoteInput sessionId:%d, smsg:%s.", sessionId, smsg.c_str()); + int32_t ret = SendMessage(sessionId, smsg); + if (ret != SUCCESS) { + DHLOGE("RespUnprepareRemoteInput error, SendMessage fail."); + return FAILURE; + } + return SUCCESS; + } else { + DHLOGE("RespUnprepareRemoteInput error, sessionId <= 0."); + return FAILURE; + } +} + +int32_t DistributedInputSinkTransport::RespStartRemoteInput( + const int32_t sessionId, std::string &smsg) +{ + if (sessionId > 0) { + DHLOGI("RespStartRemoteInput sessionId:%d, result:%d, smsg:%s.", sessionId, smsg.c_str()); + int32_t ret = SendMessage(sessionId, smsg); + if (ret != SUCCESS) { + DHLOGE("RespStartRemoteInput error, SendMessage fail."); + return FAILURE; + } + return SUCCESS; + } else { + DHLOGE("RespStartRemoteInput error, sessionId <= 0."); + return FAILURE; + } +} + +int32_t DistributedInputSinkTransport::RespStopRemoteInput( + const int32_t sessionId, std::string &smsg) +{ + if (sessionId > 0) { + DHLOGI("RespStopRemoteInput sessionId:%d, smsg:%s.", sessionId, smsg.c_str()); + int32_t ret = SendMessage(sessionId, smsg); + if (ret != SUCCESS) { + DHLOGE("RespStopRemoteInput error, SendMessage fail."); + return FAILURE; + } + return SUCCESS; + } else { + DHLOGE("RespStopRemoteInput error, sessionId <= 0."); + return FAILURE; + } +} + +int32_t DistributedInputSinkTransport::SendMessage(int32_t sessionId, std::string &message) +{ + DHLOGI("start SendMessage"); + if (message.size() > MSG_MAX_SIZE) { + DHLOGE("SendMessage error: message.size() > MSG_MAX_SIZE"); + return FAILURE; + } + uint8_t *buf = (uint8_t *)calloc((MSG_MAX_SIZE), sizeof(uint8_t)); + if (buf == nullptr) { + DHLOGE("SendMessage: malloc memory failed"); + return FAILURE; + } + int32_t outLen = 0; + if (memcpy_s(buf, MSG_MAX_SIZE, (const uint8_t *)message.c_str(), message.size()) != SUCCESS) { + DHLOGE("SendMessage: memcpy memory failed"); + free(buf); + return FAILURE; + } + outLen = message.size(); + int32_t ret = SendBytes(sessionId, buf, outLen); + free(buf); + return ret; +} + +int32_t DistributedInputSinkTransport::OnSessionOpened(int32_t sessionId, int32_t result) +{ + if (result != SUCCESS) { + DHLOGE("session open failed, sessionId %d", sessionId); + if (sessionIdSet_.count(sessionId) > 0) { + sessionIdSet_.erase(sessionId); + } + return SUCCESS; + } + + // return 1 is client + int32_t sessionSide = GetSessionSide(sessionId); + DHLOGI("session open succeed, sessionId %d, sessionSide %d", sessionId, sessionSide); + + char mySessionName[SESSION_NAME_SIZE_MAX] = ""; + char peerSessionName[SESSION_NAME_SIZE_MAX] = ""; + char peerDevId[DEVICE_ID_SIZE_MAX] = ""; + int ret = GetMySessionName(sessionId, mySessionName, sizeof(mySessionName)); + if (ret != SUCCESS) { + DHLOGI("get my session name failed, session id is %d.", sessionId); + } + // get other device session name + ret = GetPeerSessionName(sessionId, peerSessionName, sizeof(peerSessionName)); + if (ret != SUCCESS) { + DHLOGI("get my peer session name failed, session id is %d.", sessionId); + } + + ret = GetPeerDeviceId(sessionId, peerDevId, sizeof(peerDevId)); + if (ret != SUCCESS) { + DHLOGI("get my peer device id failed, session id is %d.", sessionId); + } + DHLOGI("mySessionName:%s, peerSessionName:%s, peerDevId:%s.", + mySessionName, peerSessionName, peerDevId); + + return SUCCESS; +} + +void DistributedInputSinkTransport::OnSessionClosed(int32_t sessionId) +{ + DHLOGI("OnSessionClosed, sessionId:%d", sessionId); + if (sessionIdSet_.count(sessionId) > 0) { + sessionIdSet_.erase(sessionId); + } + + DistributedInputSinkSwitch::GetInstance().RemoveSession(sessionId); +} + +void DistributedInputSinkTransport::OnBytesReceived(int32_t sessionId, const void *data, uint32_t dataLen) +{ + DHLOGI("OnBytesReceived, sessionId:%d, dataLen:%d", sessionId, dataLen); + if (sessionId < 0 || data == nullptr || dataLen <= 0) { + DHLOGE("OnBytesReceived param check failed"); + return; + } + + uint8_t *buf = (uint8_t *)calloc(dataLen + 1, sizeof(uint8_t)); + if (buf == nullptr) { + DHLOGE("OnBytesReceived: malloc memory failed"); + return; + } + + if (memcpy_s(buf, dataLen + 1, (const uint8_t*)data, dataLen) != SUCCESS) { + DHLOGE("OnBytesReceived: memcpy memory failed"); + free(buf); + return; + } + + std::string message = (char *)buf; + DHLOGI("OnBytesReceived message:%s.", message.c_str()); + HandleSessionData(sessionId, message); + + free(buf); + DHLOGI("OnBytesReceived completed"); + return; +} + +void DistributedInputSinkTransport::NotifyPrepareRemoteInput(int32_t sessionId, const nlohmann::json &recMsg) +{ + DHLOGI("OnBytesReceived cmdType is TRANS_SOURCE_MSG_PREPARE."); + if (!recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID].is_string()) { + DHLOGE("OnBytesReceived cmdType TRANS_SOURCE_MSG_PREPARE data type error."); + return; + } + std::string deviceId = recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID]; + DHLOGI("OnBytesReceived cmdType is TRANS_SOURCE_MSG_PREPARE deviceId:%s.", + deviceId.c_str()); + callback_->onPrepareRemoteInput(sessionId, deviceId); +} + +void DistributedInputSinkTransport::NotifyUnprepareRemoteInput(int32_t sessionId, const nlohmann::json &recMsg) +{ + if (!recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID].is_string()) { + DHLOGE("OnBytesReceived cmdType TRANS_SOURCE_MSG_UNPREPARE, data type is error."); + return; + } + std::string deviceId = recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID]; + DHLOGI("OnBytesReceived cmdType TRANS_SOURCE_MSG_UNPREPARE deviceId:%s.", + deviceId.c_str()); + callback_->onUnprepareRemoteInput(sessionId); +} + +void DistributedInputSinkTransport::NotifyStartRemoteInput(int32_t sessionId, const nlohmann::json &recMsg) +{ + if (!recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID].is_string()) { + DHLOGE("OnBytesReceived cmdType TRANS_SOURCE_MSG_START, data type error."); + return; + } + std::string deviceId = recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID]; + uint32_t inputTypes = recMsg[DINPUT_SOFTBUS_KEY_INPUT_TYPE]; + DHLOGI("OnBytesRecei,ved cmdType is TRANS_SOURCE_MSG_START deviceId:%s inputTypes:%d .", + deviceId.c_str(), inputTypes); + callback_->onStartRemoteInput(sessionId, inputTypes); +} + +void DistributedInputSinkTransport::NotifyStopRemoteInput(int32_t sessionId, const nlohmann::json &recMsg) +{ + if (!recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID].is_string()) { + DHLOGE("OnBytesReceived cmdType TRANS_SOURCE_MSG_STOP, data type is error."); + return; + } + std::string deviceId = recMsg[DINPUT_SOFTBUS_KEY_DEVICE_ID]; + uint32_t inputTypes = recMsg[DINPUT_SOFTBUS_KEY_INPUT_TYPE]; + DHLOGI("OnBytesReceived cmdType is TRANS_SOURCE_MSG_STOP deviceId:%s.", deviceId.c_str()); + callback_->onStopRemoteInput(sessionId, inputTypes); +} + +void DistributedInputSinkTransport::HandleSessionData(int32_t sessionId, const std::string& message) +{ + if (callback_ == nullptr) { + DHLOGE("OnBytesReceived the callback_ is null, the message:%s abort.", message.c_str()); + return; + } + + nlohmann::json recMsg = nlohmann::json::parse(message); + if (recMsg.is_discarded()) { + DHLOGE("OnBytesReceived jsonStr error."); + return; + } + + if (recMsg.contains(DINPUT_SOFTBUS_KEY_CMD_TYPE) != true) { + DHLOGE("OnBytesReceived message:%s is error, not contain cmdType.", message.c_str()); + return; + } + + if (recMsg[DINPUT_SOFTBUS_KEY_CMD_TYPE].is_number() != true) { + DHLOGE("OnBytesReceived cmdType is not number type."); + return; + } + + int cmdType = recMsg[DINPUT_SOFTBUS_KEY_CMD_TYPE]; + switch (cmdType) { + case TRANS_SOURCE_MSG_PREPARE: { + NotifyPrepareRemoteInput(sessionId, recMsg); + break; + } + case TRANS_SOURCE_MSG_UNPREPARE: { + NotifyUnprepareRemoteInput(sessionId, recMsg); + break; + } + case TRANS_SOURCE_MSG_START: { + NotifyStartRemoteInput(sessionId, recMsg); + break; + } + case TRANS_SOURCE_MSG_STOP: { + NotifyStopRemoteInput(sessionId, recMsg); + break; + } + default: + DHLOGE("OnBytesReceived cmdType is undefined."); + break; + } +} + +void DistributedInputSinkTransport::CloseAllSession() +{ + std::vector vecSession = DistributedInputSinkSwitch::GetInstance().GetAllSessionId(); + DHLOGI("CloseAllSession session vector size is %d.", vecSession.size()); + for (int32_t kIndex = 0; kIndex < vecSession.size(); ++kIndex) { + CloseSession(vecSession[kIndex]); + DHLOGI("CloseAllSession [%d] sessionid is %d.", kIndex, vecSession[kIndex]); + } + + // clear session data + DistributedInputSinkSwitch::GetInstance().InitSwitch(); +} +} // namespace DistributedInput +} // namespace DistributedHardware +} // namespace OHOS