use dsoftbus new trans api

Signed-off-by: hwzhangchuang <zhangchuang.zhang@huawei.com>

common/test/mock/session_mock.h

@@ -1,204 +0,0 @@
-/*
- * 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.
- */
-
-/**
- * @addtogroup Softbus
- * @{
- *
- * @brief Provides high-speed, secure communication between devices.
- *
- * This module implements unified distributed communication capability management between
- * nearby devices, and provides link-independent device discovery and transmission interfaces
- * to support service publishing and data transmission.
- *
- * @since 1.0
- * @version 1.0
- */
-
-/**
- * @file session.h
- *
- * @brief Declares unified data transmission interfaces.
- *
- * This file provides data transmission capabilities, including creating and removing a session server,
- * opening and closing sessions, receiving data, and querying basic session information. \n
- * After multiple nearby devices are discovered and networked, these interfaces can be used to
- * transmit data across devices. \n
- *
- * @since 1.0
- * @version 1.0
- */
-#ifndef SESSION_MOCK_H
-#define SESSION_MOCK_H
-
-#include <cstdint>
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-/**
- * @brief business type of session
- *
- * @since 1.0
- * @version 1.0
- */
-typedef enum {
-    TYPE_MESSAGE = 1,
-    TYPE_BYTES,
-    TYPE_FILE,
-    TYPE_STREAM,
-    TYPE_BUTT,
-} SessionType;
-
-typedef enum {
-    INVALID = -1,
-    /*
-     * Send any segment of a frame each time.
-     */
-    RAW_STREAM,
-    /*
-     * Send a whole video frame each time.
-     */
-    COMMON_VIDEO_STREAM,
-    /*
-     * Send a whole audio frame each time.
-     */
-    COMMON_AUDIO_STREAM,
-    /*
-     * Slice frame mode.
-     */
-    VIDEO_SLICE_STREAM,
-} StreamType;
-
-typedef enum {
-    LINK_TYPE_WIFI_WLAN_5G = 1,
-    LINK_TYPE_WIFI_WLAN_2G = 2,
-    LINK_TYPE_WIFI_P2P = 3,
-    LINK_TYPE_BR = 4,
-    LINK_TYPE_MAX = 4,
-} LinkType;
-
-typedef struct {
-    /* @brief dataType{@link SessionType} */
-    int dataType;
-    int linkTypeNum;
-    LinkType linkType[LINK_TYPE_MAX];
-    union {
-        struct StreamAttr {
-            int streamType;
-        } streamAttr;
-    } attr;
-} SessionAttribute;
-
-typedef struct {
-    char *buf;
-    int bufLen;
-} StreamData;
-
-typedef struct {
-    int type;
-    int64_t value;
-} TV;
-
-typedef struct {
-    int frameType;
-    int64_t timeStamp;
-    int seqNum;
-    int seqSubNum;
-    int level;
-    int bitMap;
-    int tvCount;
-    TV *tvList;
-} FrameInfo;
-
-enum FrameType {
-    NONE,
-    VIDEO_I,
-    VIDEO_P,
-    VIDEO_MAX = 50,
-    RADIO = VIDEO_MAX + 1,
-    RADIO_MAX = 100,
-};
-
-// APP should update StreamFrameInfo each time.
-struct StreamFrameInfo {
-    uint32_t streamId = 0;
-    uint32_t seqNum = 0;
-    uint32_t level = 0;
-    FrameType frameType = NONE;
-    uint32_t timestamp = 0;
-    uint32_t bitrate = 0;
-};
-
-constexpr uint32_t DEVICE_ID_SIZE_MAX = 65;
-constexpr uint32_t CHAR_ARRAY_SIZE = 100;
-
-typedef struct {
-    int (*OnSessionOpened)(int sessionId, int result);
-    void (*OnSessionClosed)(int sessionId);
-    void (*OnBytesReceived)(int sessionId, const void *data, unsigned int dataLen);
-    void (*OnMessageReceived)(int sessionId, const void *data, unsigned int dataLen);
-    void (*OnStreamReceived)(int sessionId, const StreamData *data, const StreamData *ext,
-        const StreamFrameInfo *param);
-} ISessionListener;
-
-typedef struct {
-    int (*OnReceiveFileStarted)(int sessionId, const char *files, int fileCnt);
-    int (*OnReceiveFileProcess)(int sessionId, const char *firstFile, uint64_t bytesUpload, uint64_t bytesTotal);
-    void (*OnReceiveFileFinished)(int sessionId, const char *files, int fileCnt);
-    void (*OnFileTransError)(int sessionId);
-} IFileReceiveListener;
-
-typedef struct {
-    int (*OnSendFileProcess)(int sessionId, uint64_t bytesUpload, uint64_t bytesTotal);
-    int (*OnSendFileFinished)(int sessionId, const char *firstFile);
-    void (*OnFileTransError)(int sessionId);
-} IFileSendListener;
-
-int CreateSessionServer(const char *pkgName, const char *sessionName, const ISessionListener *listener);
-
-int RemoveSessionServer(const char *pkgName, const char *sessionName);
-
-int OpenSession(const char *mySessionName, const char *peerSessionName, const char *peerDeviceId, const char *groupId,
-    const SessionAttribute *attr);
-void OpenSessionResult(void);
-
-void CloseSession(int sessionId);
-
-int SendBytes(int sessionId, const void *data, unsigned int len);
-
-int SendMessage(int sessionId, const void *data, unsigned int len);
-
-int SendStream(int sessionId, const StreamData *data, const StreamData *ext, const FrameInfo *param);
-
-int GetMySessionName(int sessionId, char *sessionName, unsigned int len);
-
-int GetPeerSessionName(int sessionId, char *sessionName, unsigned int len);
-
-int GetPeerDeviceId(int sessionId, char *devId, unsigned int len);
-
-int GetSessionSide(int sessionId);
-
-int SetFileReceiveListener(const char *pkgName, const char *sessionName, const IFileReceiveListener *recvListener,
-    const char *rootDir);
-
-int SetFileSendListener(const char *pkgName, const char *sessionName, const IFileSendListener *sendListener);
-
-int SendFile(int sessionId, const char *sFileList[], const char *dFileList[], uint32_t fileCnt);
-
-#ifdef __cplusplus
-}
-#endif
-#endif // SESSION_MOCK_H

common/test/mock/socket_mock.cpp

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2022-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 "socket_mock.h"
+
+int32_t Socket(SocketInfo info)
+{
+    (void)info;
+    return 0;
+}
+
+int32_t Listen(int32_t socket, const QosTV qos[], uint32_t qosCount, const ISocketListener *listener)
+{
+    (void)socket;
+    (void)qos;
+    (void)qosCount;
+    (void)listener;
+    return 0;
+}
+
+int32_t Bind(int32_t socket, const QosTV qos[], uint32_t qosCount, const ISocketListener *listener)
+{
+    (void)socket;
+    (void)qos;
+    (void)qosCount;
+    (void)listener;
+    return 0;
+}
+
+int32_t SendBytes(int32_t socket, const void *data, uint32_t len)
+{
+    (void)socket;
+    (void)data;
+    (void)len;
+    return 0;
+}
+
+int32_t SendMessage(int32_t socket, const void *data, uint32_t len)
+{
+    (void)socket;
+    (void)data;
+    (void)len;
+    return 0;
+}
+
+int32_t SendStream(int32_t socket, const StreamData *data, const StreamData *ext, const StreamFrameInfo *param)
+{
+    (void)socket;
+    (void)data;
+    (void)ext;
+    (void)param;
+    return 0;
+}
+
+int32_t SendFile(int32_t socket, const char *sFileList[], const char *dFileList[], uint32_t fileCnt)
+{
+    (void)socket;
+    (void)sFileList;
+    (void)dFileList;
+    (void)fileCnt;
+    return 0;
+}
+
+void Shutdown(int32_t socket)
+{
+    (void)socket;
+}
+
+int32_t EvaluateQos(const char *peerNetworkId, TransDataType dataType, const QosTV *qos, uint32_t qosCount)
+{
+    (void)peerNetworkId;
+    (void)dataType;
+    (void)qos;
+    (void)qosCount;
+    return 0;
+}

common/test/mock/softbus_bus_center_mock.h

@@ -1,57 +0,0 @@
-/*
- * 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 SOFTBUS_BUS_CENTER_MOCK_H
-#define SOFTBUS_BUS_CENTER_MOCK_H
-
-#include <cstdint>
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-constexpr uint32_t MOCK_NETWORK_ID_BUF_LEN = 65;
-constexpr uint32_t MOCK_DEVICE_NAME_BUF_LEN = 65;
-constexpr uint32_t UUID_BUF_LEN = 65;
-/**
- * @brief Defines the basic information about a device.
- * @since 1.0
- * @version 1.0
- */
-typedef struct {
-    char networkId[MOCK_NETWORK_ID_BUF_LEN];
-    char deviceName[MOCK_DEVICE_NAME_BUF_LEN];
-    uint16_t deviceTypeId;
-} NodeBasicInfo;
-
-typedef enum {
-    NODE_KEY_UDID = 0,    /**< UDID in string format*/
-    NODE_KEY_UUID,        /**< UUID in string format */
-    NODE_KEY_MASTER_UDID, /**< UDID of master node in string format */
-    NODE_KEY_BR_MAC,      /**< BR MAC in string format */
-    NODE_KEY_IP_ADDRESS,  /**< IP address in string format */
-    NODE_KEY_DEV_NAME,    /**< Device name in string format */
-    NODE_KEY_NETWORK_CAPABILITY, /**< Network capability in number format */
-    NODE_KEY_NETWORK_TYPE,       /**< Network type in number format */
-    NODE_KEY_BLE_OFFLINE_CODE,   /**< Ble offlinecode in string format */
-    NODE_KEY_DATA_CHANGE_FLAG,
-} NodeDeviceInfoKey;
-
-int32_t GetLocalNodeDeviceInfo(const char *pkgName, NodeBasicInfo *info);
-int32_t GetNodeKeyInfo(const char *pkgName, const char *networkId,
-    NodeDeviceInfoKey key, uint8_t *info, int32_t infoLen);
-#ifdef __cplusplus
-}
-#endif
-#endif // SOFTBUS_BUS_CENTER_MOCK_H

services/sink/transport/src/distributed_input_sink_transport.cpp

@@ -29,7 +29,6 @@
 #include "dinput_softbus_define.h"
 #include "dinput_utils_tool.h"
 #include "hidumper.h"
-#include "session.h"
 #include "softbus_bus_center.h"
 #include "xcollie/watchdog.h"
 
@@ -53,7 +52,6 @@ DistributedInputSinkTransport::DistributedInputSinkTransport() : mySessionName_(
 DistributedInputSinkTransport::~DistributedInputSinkTransport()
 {
     DHLOGI("DistributedInputSinkTransport dtor.");
-    (void)RemoveSessionServer(DINPUT_PKG_NAME.c_str(), mySessionName_.c_str());
 }
 
 DistributedInputSinkTransport::DInputSinkEventHandler::DInputSinkEventHandler(
@@ -550,13 +548,7 @@ void DistributedInputSinkTransport::HandleData(int32_t sessionId, const std::str
 
 void DistributedInputSinkTransport::CloseAllSession()
 {
-    std::vector<int32_t> vecSession = DistributedInputSinkSwitch::GetInstance().GetAllSessionId();
-    DHLOGI("CloseAllSession session vector size is %d", vecSession.size());
-    for (size_t kIndex = 0; kIndex < vecSession.size(); ++kIndex) {
-        CloseSession(vecSession[kIndex]);
-        DHLOGI("CloseAllSession [%d] sessionid is %s", kIndex, GetAnonyInt32(vecSession[kIndex]).c_str());
-    }
-
+    DistributedInputTransportBase::GetInstance().StopAllSession();
     // clear session data
     DistributedInputSinkSwitch::GetInstance().InitSwitch();
 }

services/sink/transport/test/sinktransunittest/BUILD.gn

@@ -75,7 +75,6 @@ ohos_unittest("distributed_input_sinktrans_test") {
     "HI_LOG_ENABLE",
     "DH_LOG_TAG=\"distributedinpututtest\"",
     "LOG_DOMAIN=0xD004100",
-    "COMPILE_TEST_MODE",
   ]
 
   deps = [

services/source/sourcemanager/test/sourcemanagerunittest/BUILD.gn

@@ -107,7 +107,6 @@ ohos_unittest("distributed_input_sourcemanager_test") {
     "HI_LOG_ENABLE",
     "DH_LOG_TAG=\"distributedinpututtest\"",
     "LOG_DOMAIN=0xD004100",
-    "COMPILE_TEST_MODE",
   ]
 
   deps = [

services/source/transport/test/sourcetransunittest/BUILD.gn

@@ -71,7 +71,6 @@ ohos_unittest("distributed_input_sourcetrans_test") {
     "HI_LOG_ENABLE",
     "DH_LOG_TAG=\"distributedinpututtest\"",
     "LOG_DOMAIN=0xD004100",
-    "COMPILE_TEST_MODE",
   ]
 
   deps = [

services/transportbase/include/distributed_input_transport_base.h

@@ -31,6 +31,10 @@
 #include "securec.h"
 #include "single_instance.h"
 
+#include "socket.h"
+
+#include "softbus_bus_center.h"
+
 #include "dinput_sink_manager_callback.h"
 #include "dinput_source_manager_callback.h"
 #include "dinput_transbase_source_callback.h"
@@ -46,6 +50,7 @@ public:
     int32_t Init();
     int32_t StartSession(const std::string &remoteDevId);
     void StopSession(const std::string &remoteDevId);
+    void StopAllSession();
 
     void RegisterSrcHandleSessionCallback(std::shared_ptr<DInputTransbaseSourceCallback> callback);
     void RegisterSinkHandleSessionCallback(std::shared_ptr<DInputTransbaseSinkCallback> callback);
@@ -53,8 +58,8 @@ public:
     void RegisterSinkManagerCallback(std::shared_ptr<DInputSinkManagerCallback> callback);
     void RegisterSessionStateCb(sptr<ISessionStateCallback> callback);
     void UnregisterSessionStateCb();
-    int32_t OnSessionOpened(int32_t sessionId, int32_t result);
-    void OnSessionClosed(int32_t sessionId);
+    int32_t OnSessionOpened(int32_t sessionId, PeerSocketInfo info);
+    void OnSessionClosed(int32_t sessionId, ShutdownReason reason);
     void OnBytesReceived(int32_t sessionId, const void *data, uint32_t dataLen);
 
     int32_t GetCurrentSessionId();
@@ -67,21 +72,23 @@ public:
 private:
     DistributedInputTransportBase() = default;
     ~DistributedInputTransportBase();
-    void OnSessionOpenedError(int32_t sessionId, int32_t result);
     int32_t CheckDeviceSessionState(const std::string &remoteDevId);
     bool CheckRecivedData(const std::string &message);
     void HandleSession(int32_t sessionId, const std::string &message);
     void Release();
     void RunSessionStateCallback(const std::string &remoteDevId, const uint32_t sessionState);
 
+    int32_t CreateServerSocket();
+    int32_t CreateClientSocket(const std::string &remoteDevId);
+
 private:
     std::atomic<bool> isSessSerCreateFlag_ = false;
+    std::atomic<int32_t> localServerSocket_;
     std::mutex sessSerOperMutex_;
     std::mutex operationMutex_;
     std::string remoteDeviceId_;
     std::map<std::string, int32_t> remoteDevSessionMap_;
     std::map<std::string, bool> channelStatusMap_;
-    std::condition_variable openSessionWaitCond_;
     std::string localSessionName_ = "";
     int32_t sessionId_ = 0;
 

services/transportbase/src/distributed_input_transport_base.cpp

@@ -32,36 +32,21 @@
 #include "dinput_utils_tool.h"
 #include "hidumper.h"
 
-#ifndef COMPILE_TEST_MODE
-#include "session.h"
-#else
-#include "session_mock.h"
-#endif
-
-#ifndef COMPILE_TEST_MODE
-#include "softbus_bus_center.h"
-#else
-#include "softbus_bus_center_mock.h"
-#endif
-
 #include "softbus_common.h"
 
 namespace OHOS {
 namespace DistributedHardware {
 namespace DistributedInput {
-const int32_t DINPUT_LINK_TYPE_MAX = 4;
+namespace {
 const int32_t SESSION_STATUS_OPENED = 0;
 const int32_t SESSION_STATUS_CLOSED = 1;
-static SessionAttribute g_sessionAttr = {
-    .dataType = SessionType::TYPE_BYTES,
-    .linkTypeNum = DINPUT_LINK_TYPE_MAX,
-    .linkType = {
-        LINK_TYPE_WIFI_P2P,
-        LINK_TYPE_WIFI_WLAN_2G,
-        LINK_TYPE_WIFI_WLAN_5G,
-        LINK_TYPE_BR
-    }
+static QosTV g_qosInfo[] = {
+    { .qos = QOS_TYPE_MIN_BW, .value = 10 * 1024 * 1024},
+    { .qos = QOS_TYPE_MAX_LATENCY, .value = 2000 },
+    { .qos = QOS_TYPE_TRANS_RELIABILITY, .value = QOS_RELIABILITY_FULL }
 };
+static uint32_t g_QosTV_Param_Index = static_cast<uint32_t>(sizeof(g_qosInfo) / sizeof(g_qosInfo[0]));
+}
 IMPLEMENT_SINGLE_INSTANCE(DistributedInputTransportBase);
 DistributedInputTransportBase::~DistributedInputTransportBase()
 {
@@ -69,50 +54,91 @@ DistributedInputTransportBase::~DistributedInputTransportBase()
     Release();
 }
 
-static int32_t SessionOpened(int32_t sessionId, int32_t result)
+void OnBind(int32_t socket, PeerSocketInfo info)
 {
-    return DistributedInput::DistributedInputTransportBase::GetInstance().OnSessionOpened(sessionId, result);
+    DistributedInput::DistributedInputTransportBase::GetInstance().OnSessionOpened(socket, info);
 }
 
-static void SessionClosed(int32_t sessionId)
+void OnShutdown(int32_t socket, ShutdownReason reason)
 {
-    DistributedInput::DistributedInputTransportBase::GetInstance().OnSessionClosed(sessionId);
+    DistributedInput::DistributedInputTransportBase::GetInstance().OnSessionClosed(socket, reason);
 }
 
-static void BytesReceived(int32_t sessionId, const void *data, uint32_t dataLen)
+void OnBytes(int32_t socket, const void *data, uint32_t dataLen)
 {
-    DistributedInput::DistributedInputTransportBase::GetInstance().OnBytesReceived(sessionId, data, dataLen);
+    DistributedInput::DistributedInputTransportBase::GetInstance().OnBytesReceived(socket, data, dataLen);
 }
 
-static void MessageReceived(int32_t sessionId, const void *data, uint32_t dataLen)
+void OnMessage(int32_t socket, const void *data, uint32_t dataLen)
 {
-    (void)sessionId;
+    (void)socket;
     (void)data;
     (void)dataLen;
-    DHLOGI("sessionId: %d, dataLen:%d", sessionId, dataLen);
+    DHLOGI("socket: %d, dataLen:%d", socket, dataLen);
 }
 
-static void StreamReceived(int32_t sessionId, const StreamData *data, const StreamData *ext,
+void OnStream(int32_t socket, const StreamData *data, const StreamData *ext,
     const StreamFrameInfo *param)
 {
-    (void)sessionId;
+    (void)socket;
     (void)data;
     (void)ext;
     (void)param;
-    DHLOGI("sessionId: %d", sessionId);
+    DHLOGI("socket: %d", socket);
 }
 
+void OnFile(int32_t socket, FileEvent *event)
+{
+    (void)event;
+    DHLOGI("socket: %d", socket);
+}
+
+void OnQos(int32_t socket, QoSEvent eventId, const QosTV *qos, uint32_t qosCount)
+{
+    DHLOGI("OnQos, socket: %d, QoSEvent: %d, qosCount: %ld", socket, (int32_t)eventId, qosCount);
+    for (uint32_t idx = 0; idx < qosCount; idx++) {
+        DHLOGI("QosTV: type: %d, value: %d", (int32_t)qos[idx].qos, qos[idx].value);
+    }
+}
+
+ISocketListener iSocketListener = {
+    .OnBind = OnBind,
+    .OnShutdown = OnShutdown,
+    .OnBytes = OnBytes,
+    .OnMessage = OnMessage,
+    .OnStream = OnStream,
+    .OnFile = OnFile,
+    .OnQos = OnQos
+};
+
 int32_t DistributedInputTransportBase::Init()
 {
     DHLOGI("Init Transport Base Session");
-    ISessionListener iSessionListener = {
-        .OnSessionOpened = SessionOpened,
-        .OnSessionClosed = SessionClosed,
-        .OnBytesReceived = BytesReceived,
-        .OnMessageReceived = MessageReceived,
-        .OnStreamReceived = StreamReceived
-    };
+    std::unique_lock<std::mutex> sessionServerLock(sessSerOperMutex_);
+    if (isSessSerCreateFlag_.load()) {
+        DHLOGI("SessionServer already create success.");
+        return DH_SUCCESS;
+    }
+    int32_t socket = CreateServerSocket();
+    if (socket < DH_SUCCESS) {
+        DHLOGE("CreateServerSocket failed, ret: %d", socket);
+        return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_INIT_FAIL;
+    }
 
+    int32_t ret = Listen(socket, g_qosInfo, g_QosTV_Param_Index, &iSocketListener);
+    if (ret != DH_SUCCESS) {
+        DHLOGE("Socket Listen failed, error code %d.", ret);
+        return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_INIT_FAIL;
+    }
+    isSessSerCreateFlag_.store(true);
+    localServerSocket_ = socket;
+    DHLOGI("Finish Init DSoftBus Server Socket, socket: %d", socket);
+    return DH_SUCCESS;
+}
+
+int32_t DistributedInputTransportBase::CreateServerSocket()
+{
+    DHLOGI("CreateServerSocket start");
     auto localNode = std::make_unique<NodeBasicInfo>();
     int32_t retCode = GetLocalNodeDeviceInfo(DINPUT_PKG_NAME.c_str(), localNode.get());
     if (retCode != DH_SUCCESS) {
@@ -120,21 +146,17 @@ int32_t DistributedInputTransportBase::Init()
         return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_INIT_FAIL;
     }
     std::string networkId = localNode->networkId;
-    DHLOGI("Init device local networkId is %s", GetAnonyString(networkId).c_str());
-
-    std::unique_lock<std::mutex> sessionServerLock(sessSerOperMutex_);
-    if (isSessSerCreateFlag_.load()) {
-        DHLOGI("SessionServer already create success.");
-        return DH_SUCCESS;
-    }
     localSessionName_ = SESSION_NAME + networkId.substr(0, INTERCEPT_STRING_LENGTH);
-    int32_t ret = CreateSessionServer(DINPUT_PKG_NAME.c_str(), localSessionName_.c_str(), &iSessionListener);
-    if (ret != DH_SUCCESS) {
-        DHLOGE("Init CreateSessionServer failed, error code %d.", ret);
-        return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_INIT_FAIL;
-    }
-    isSessSerCreateFlag_.store(true);
-    return DH_SUCCESS;
+    DHLOGI("CreateServerSocket local networkId is %s, local socketName: %s",
+        networkId.c_str(), localSessionName_.c_str());
+    SocketInfo info = {
+        .name = const_cast<char*>(localSessionName_.c_str()),
+        .pkgName = const_cast<char*>(DINPUT_PKG_NAME.c_str()),
+        .dataType = DATA_TYPE_BYTES
+    };
+    int32_t socket = Socket(info);
+    DHLOGI("CreateServerSocket Finish, socket: %d", socket);
+    return socket;
 }
 
 void DistributedInputTransportBase::Release()
@@ -142,15 +164,18 @@ void DistributedInputTransportBase::Release()
     std::unique_lock<std::mutex> sessionLock(operationMutex_);
     auto iter = remoteDevSessionMap_.begin();
     for (; iter != remoteDevSessionMap_.end(); ++iter) {
-        CloseSession(iter->second);
+        DHLOGI("Shutdown client socket: %d to remote dev: %s", iter->second, iter->first.c_str());
+        Shutdown(iter->second);
     }
 
     {
         std::unique_lock<std::mutex> sessionServerLock(sessSerOperMutex_);
         if (!isSessSerCreateFlag_.load()) {
-            DHLOGI("SessionServer already remove success.");
+            DHLOGI("DSoftBus Server Socket already remove success.");
         } else {
-            (void)RemoveSessionServer(DINPUT_PKG_NAME.c_str(), localSessionName_.c_str());
+            DHLOGI("Shutdown DSoftBus Server Socket, socket: %d", localServerSocket_.load());
+            Shutdown(localServerSocket_.load());
+            localServerSocket_ = -1;
             isSessSerCreateFlag_.store(false);
         }
     }
@@ -164,7 +189,7 @@ int32_t DistributedInputTransportBase::CheckDeviceSessionState(const std::string
     if (remoteDevSessionMap_.find(remoteDevId) == remoteDevSessionMap_.end()) {
         return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_DEVICE_SESSION_STATE;
     }
-    DHLOGI("CheckDeviceSessionState has opened %s", GetAnonyString(remoteDevId).c_str());
+    DHLOGI("CheckDeviceSessionState has opened %s", remoteDevId.c_str());
     return DH_SUCCESS;
 }
 
@@ -179,44 +204,50 @@ std::string DistributedInputTransportBase::GetDevIdBySessionId(int32_t sessionId
     return "";
 }
 
+int32_t DistributedInputTransportBase::CreateClientSocket(const std::string &remoteDevId)
+{
+    DHLOGI("CreateClientSocket start, peerNetworkId: %s", remoteDevId.c_str());
+    std::string peerSessionName = SESSION_NAME + remoteDevId.substr(0, INTERCEPT_STRING_LENGTH);
+    SocketInfo info = {
+        .name = const_cast<char*>(localSessionName_.c_str()),
+        .peerName = const_cast<char*>(peerSessionName.c_str()),
+        .peerNetworkId = const_cast<char*>(remoteDevId.c_str()),
+        .pkgName = const_cast<char*>(DINPUT_PKG_NAME.c_str()),
+        .dataType = DATA_TYPE_BYTES
+    };
+    int32_t socket = Socket(info);
+    DHLOGI("Bind Socket server, socket: %d, localSessionName: %s, peerSessionName: %s",
+        socket, localSessionName_.c_str(), peerSessionName.c_str());
+    return socket;
+}
+
 int32_t DistributedInputTransportBase::StartSession(const std::string &remoteDevId)
 {
     int32_t ret = CheckDeviceSessionState(remoteDevId);
     if (ret == DH_SUCCESS) {
-        DHLOGE("Softbus session has already opened, deviceId: %s", GetAnonyString(remoteDevId).c_str());
+        DHLOGE("Softbus session has already opened, deviceId: %s", remoteDevId.c_str());
         return DH_SUCCESS;
     }
 
-    std::string peerSessionName = SESSION_NAME + remoteDevId.substr(0, INTERCEPT_STRING_LENGTH);
-    DHLOGI("OpenInputSoftbus localSessionName: %s, peerSessionName:%s, remoteDevId: %s",
-        localSessionName_.c_str(), peerSessionName.c_str(), remoteDevId.c_str());
-
+    int socket = CreateClientSocket(remoteDevId);
+    if (socket < DH_SUCCESS) {
+        DHLOGE("StartSession failed, ret: %d", socket);
+        return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_OPEN_SESSION_FAIL;
+    }
     StartAsyncTrace(DINPUT_HITRACE_LABEL, DINPUT_OPEN_SESSION_START, DINPUT_OPEN_SESSION_TASK);
-    int32_t sessionId = OpenSession(localSessionName_.c_str(), peerSessionName.c_str(), remoteDevId.c_str(),
-        GROUP_ID.c_str(), &g_sessionAttr);
-    if (sessionId < 0) {
-        DHLOGE("OpenSession fail, remoteDevId: %s, sessionId: %d", GetAnonyString(remoteDevId).c_str(), sessionId);
+    ret = Bind(socket, g_qosInfo, g_QosTV_Param_Index, &iSocketListener);
+    if (ret < DH_SUCCESS) {
+        DHLOGE("OpenSession fail, remoteDevId: %s, socket: %d", remoteDevId.c_str(), socket);
         FinishAsyncTrace(DINPUT_HITRACE_LABEL, DINPUT_OPEN_SESSION_START, DINPUT_OPEN_SESSION_TASK);
+        Shutdown(socket);
         return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_OPEN_SESSION_FAIL;
     }
 
-    HiDumper::GetInstance().CreateSessionInfo(remoteDevId, sessionId, localSessionName_, peerSessionName,
-        SessionStatus::OPENING);
-
-    DHLOGI("Wait for channel session opened.");
-    {
-        std::unique_lock<std::mutex> waitLock(operationMutex_);
-        auto status = openSessionWaitCond_.wait_for(waitLock, std::chrono::seconds(SESSION_WAIT_TIMEOUT_SECOND),
-            [this, remoteDevId] () { return channelStatusMap_[remoteDevId]; });
-        if (!status) {
-            DHLOGE("OpenSession timeout, remoteDevId: %s, sessionId: %d",
-                GetAnonyString(remoteDevId).c_str(), sessionId);
-            return ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_OPEN_SESSION_TIMEOUT;
-        }
-    }
-
-    DHLOGI("OpenSession success, remoteDevId:%s, sessionId: %d", GetAnonyString(remoteDevId).c_str(), sessionId);
-    sessionId_ = sessionId;
+    std::string peerSessionName = SESSION_NAME + remoteDevId.substr(0, INTERCEPT_STRING_LENGTH);
+    HiDumper::GetInstance().CreateSessionInfo(remoteDevId, socket, localSessionName_, peerSessionName,
+        SessionStatus::OPENED);
+    DHLOGI("OpenSession success, remoteDevId:%s, sessionId: %d", remoteDevId.c_str(), socket);
+    sessionId_ = socket;
 
     std::shared_ptr<DistributedHardwareFwkKit> dhFwkKit = DInputContext::GetInstance().GetDHFwkKit();
     if (dhFwkKit != nullptr) {
@@ -224,7 +255,13 @@ int32_t DistributedInputTransportBase::StartSession(const std::string &remoteDev
         dhFwkKit->PublishMessage(DHTopic::TOPIC_LOW_LATENCY, ENABLE_LOW_LATENCY.dump());
     }
 
-    HiDumper::GetInstance().SetSessionStatus(remoteDevId, SessionStatus::OPENED);
+    PeerSocketInfo peerSocketInfo = {
+        .name = const_cast<char*>(peerSessionName.c_str()),
+        .deviceId = const_cast<char*>(remoteDevId.c_str()),
+        .pkgName = const_cast<char*>(DINPUT_PKG_NAME.c_str()),
+        .dataType = DATA_TYPE_BYTES
+    };
+    OnSessionOpened(socket, peerSocketInfo);
     return DH_SUCCESS;
 }
 
@@ -233,19 +270,35 @@ int32_t DistributedInputTransportBase::GetCurrentSessionId()
     return sessionId_;
 }
 
+void DistributedInputTransportBase::StopAllSession()
+{
+    std::map<std::string, int32_t> remoteDevSessions;
+    {
+        std::unique_lock<std::mutex> sessionLock(operationMutex_);
+        std::for_each(remoteDevSessionMap_.begin(), remoteDevSessionMap_.end(),
+            [&remoteDevSessions] (const std::pair<std::string, int32_t> &pair) {
+            remoteDevSessions[pair.first] = pair.second;
+        });
+    }
+
+    std::for_each(remoteDevSessions.begin(), remoteDevSessions.end(),
+        [this](const std::pair<std::string, int32_t> &pair) {
+        StopSession(pair.first);
+    });
+}
+
 void DistributedInputTransportBase::StopSession(const std::string &remoteDevId)
 {
     std::unique_lock<std::mutex> sessionLock(operationMutex_);
-
     if (remoteDevSessionMap_.count(remoteDevId) == 0) {
-        DHLOGE("remoteDevSessionMap not find remoteDevId: %s", GetAnonyString(remoteDevId).c_str());
+        DHLOGE("remoteDevSessionMap not find remoteDevId: %s", remoteDevId.c_str());
         return;
     }
     int32_t sessionId = remoteDevSessionMap_[remoteDevId];
 
-    DHLOGI("RemoteDevId: %s, sessionId: %d", GetAnonyString(remoteDevId).c_str(), sessionId);
+    DHLOGI("RemoteDevId: %s, sessionId: %d", remoteDevId.c_str(), sessionId);
     HiDumper::GetInstance().SetSessionStatus(remoteDevId, SessionStatus::CLOSING);
-    CloseSession(sessionId);
+    Shutdown(sessionId);
     remoteDevSessionMap_.erase(remoteDevId);
     channelStatusMap_.erase(remoteDevId);
 
@@ -262,14 +315,14 @@ void DistributedInputTransportBase::StopSession(const std::string &remoteDevId)
 void DistributedInputTransportBase::RegisterSrcHandleSessionCallback(
     std::shared_ptr<DInputTransbaseSourceCallback> callback)
 {
-    DHLOGI("RegisterTransbaseSourceRespCallback");
+    DHLOGI("RegisterSrcHandleSessionCallback");
     srcCallback_ = callback;
 }
 
 void DistributedInputTransportBase::RegisterSinkHandleSessionCallback(
     std::shared_ptr<DInputTransbaseSinkCallback> callback)
 {
-    DHLOGI("RegisterTransbaseSinkRespCallback");
+    DHLOGI("RegisterSinkHandleSessionCallback");
     sinkCallback_ = callback;
 }
 
@@ -283,7 +336,7 @@ void DistributedInputTransportBase::RegisterSourceManagerCallback(
 void DistributedInputTransportBase::RegisterSinkManagerCallback(
     std::shared_ptr<DInputSinkManagerCallback> callback)
 {
-    DHLOGI("RegisterSourceManagerCallback");
+    DHLOGI("RegisterSinkManagerCallback");
     sinkMgrCallback_ = callback;
 }
 
@@ -320,46 +373,18 @@ void DistributedInputTransportBase::EraseSessionId(const std::string &remoteDevI
     remoteDevSessionMap_.erase(remoteDevId);
 }
 
-int32_t DistributedInputTransportBase::OnSessionOpened(int32_t sessionId, int32_t result)
+int32_t DistributedInputTransportBase::OnSessionOpened(int32_t sessionId, PeerSocketInfo info)
 {
-    DHLOGI("OnSessionOpened, sessionId: %d, result: %d", sessionId, result);
+    DHLOGI("OnSessionOpened, socket: %d, peerSocketName: %s, peerNetworkId: %s, peerPkgName: %s",
+        sessionId, info.name, info.deviceId, info.pkgName);
     FinishAsyncTrace(DINPUT_HITRACE_LABEL, DINPUT_OPEN_SESSION_START, DINPUT_OPEN_SESSION_TASK);
-    if (result != DH_SUCCESS) {
-        OnSessionOpenedError(sessionId, result);
-        return DH_SUCCESS;
-    }
-
-    char mySessionName[SESSION_NAME_SIZE_MAX] = {0};
-    char peerSessionName[SESSION_NAME_SIZE_MAX] = {0};
-    char peerDevId[DEVICE_ID_SIZE_MAX] = {0};
-    int32_t ret = GetMySessionName(sessionId, mySessionName, sizeof(mySessionName));
-    if (ret != DH_SUCCESS) {
-        DHLOGE("get my session name failed, session id is %d", sessionId);
-    }
-    ret = GetPeerSessionName(sessionId, peerSessionName, sizeof(peerSessionName));
-    if (ret != DH_SUCCESS) {
-        DHLOGE("get peer session name failed, session id is %d", sessionId);
-    }
-    ret = GetPeerDeviceId(sessionId, peerDevId, sizeof(peerDevId));
-    if (ret != DH_SUCCESS) {
-        DHLOGE("get peer device id failed, session id is %d", sessionId);
-    }
 
+    std::string peerDevId;
+    peerDevId.assign(info.deviceId);
     {
         std::unique_lock<std::mutex> sessionLock(operationMutex_);
         remoteDevSessionMap_[peerDevId] = sessionId;
-    }
-
-    int32_t sessionSide = GetSessionSide(sessionId);
-    DHLOGI("session open succeed, sessionId: %d, sessionSide:%d(1 is "
-        "client side), deviceId:%s", sessionId, sessionSide, GetAnonyString(peerDevId).c_str());
-
-    DHLOGI("mySessionName:%s, peerSessionName:%s, peerDevId:%s",
-        mySessionName, peerSessionName, GetAnonyString(peerDevId).c_str());
-    {
-        std::lock_guard<std::mutex> notifyLock(operationMutex_);
         channelStatusMap_[peerDevId] = true;
-        openSessionWaitCond_.notify_all();
     }
     RunSessionStateCallback(peerDevId, SESSION_STATUS_OPENED);
     std::shared_ptr<DistributedHardwareFwkKit> dhFwkKit = DInputContext::GetInstance().GetDHFwkKit();
@@ -371,23 +396,12 @@ int32_t DistributedInputTransportBase::OnSessionOpened(int32_t sessionId, int32_
     return DH_SUCCESS;
 }
 
-void DistributedInputTransportBase::OnSessionOpenedError(int32_t sessionId, int32_t result)
+void DistributedInputTransportBase::OnSessionClosed(int32_t sessionId, ShutdownReason reason)
 {
-    std::string deviceId = GetDevIdBySessionId(sessionId);
-    DHLOGE("session open failed, sessionId: %d, result:%d, deviceId:%s", sessionId, result,
-        GetAnonyString(deviceId).c_str());
-    std::unique_lock<std::mutex> sessionLock(operationMutex_);
-    if (CountSession(deviceId) > 0) {
-        EraseSessionId(deviceId);
-    }
-}
-
-void DistributedInputTransportBase::OnSessionClosed(int32_t sessionId)
-{
-    DHLOGI("OnSessionClosed, sessionId: %d", sessionId);
+    DHLOGI("OnSessionClosed, socket: %d, reason: %d", sessionId, (int32_t)reason);
     std::string deviceId = GetDevIdBySessionId(sessionId);
     DHLOGI("OnSessionClosed notify session closed, sessionId: %d, peer deviceId:%s",
-        sessionId, GetAnonyString(deviceId).c_str());
+        sessionId, deviceId.c_str());
     RunSessionStateCallback(deviceId, SESSION_STATUS_CLOSED);
 
     {
@@ -533,7 +547,7 @@ int32_t DistributedInputTransportBase::GetSessionIdByDevId(const std::string &sr
     if (it != remoteDevSessionMap_.end()) {
         return it->second;
     }
-    DHLOGE("get session id failed, srcId = %s", GetAnonyString(srcId).c_str());
+    DHLOGE("get session id failed, srcId = %s", srcId.c_str());
     return ERR_DH_INPUT_SERVER_SINK_TRANSPORT_GET_SESSIONID_FAIL;
 }
 } // namespace DistributedInput

services/transportbase/test/transbaseunittest/BUILD.gn

@@ -48,7 +48,7 @@ ohos_unittest("distributed_input_transbase_test") {
   ]
 
   sources = [
-    "${common_path}/test/mock/session_mock.cpp",
+    "${common_path}/test/mock/socket_mock.cpp",
     "${common_path}/test/mock/softbus_bus_center_mock.cpp",
     "${distributedinput_path}/services/transportbase/src/distributed_input_transport_base.cpp",
     "${services_sink_path}/transport/src/distributed_input_sink_switch.cpp",
@@ -69,7 +69,6 @@ ohos_unittest("distributed_input_transbase_test") {
     "HI_LOG_ENABLE",
     "DH_LOG_TAG=\"distributedinpututtest\"",
     "LOG_DOMAIN=0xD004100",
-    "COMPILE_TEST_MODE",
   ]
 
   deps = [

services/transportbase/test/transbaseunittest/distributed_input_transbase_test.cpp

@@ -26,6 +26,11 @@ using namespace std;
 namespace OHOS {
 namespace DistributedHardware {
 namespace DistributedInput {
+namespace {
+    const std::string PEER_SESSION_NAME = "ohos.dhardware.dinput.session8647073e02e7a78f09473aa122";
+    const std::string REMOTE_DEV_ID = "f6d4c0864707aefte7a78f09473aa122ff57fc81c00981fcf5be989e7d112591";
+    const std::string DINPUT_PKG_NAME_TEST = "ohos.dhardware.dinput";
+}
 void DistributedInputTransbaseTest::SetUp()
 {
     sourceTransport_ = new DistributedInputSourceTransport();
@@ -86,16 +91,21 @@ HWTEST_F(DistributedInputTransbaseTest, GetDevIdBySessionId01, testing::ext::Tes
 
 HWTEST_F(DistributedInputTransbaseTest, OnSessionOpened01, testing::ext::TestSize.Level0)
 {
+
     int32_t sessionId = 0;
-    int32_t result = ERR_DH_INPUT_SERVER_SOURCE_TRANSPORT_STOP_FAIL;
+    PeerSocketInfo peerSocketInfo = {
+        .name = const_cast<char*>(PEER_SESSION_NAME.c_str()),
+        .deviceId = const_cast<char*>(REMOTE_DEV_ID.c_str()),
+        .pkgName = const_cast<char*>(DINPUT_PKG_NAME_TEST.c_str()),
+        .dataType = DATA_TYPE_BYTES
+    };
     std::string srcId = "f6d4c08647073e02e7a78f09473aa122ff57fc81c00981fcf5be989e7d112591";
     DistributedInputTransportBase::GetInstance().remoteDevSessionMap_[srcId] = sessionId;
-    int32_t ret = DistributedInputTransportBase::GetInstance().OnSessionOpened(sessionId, result);
+    int32_t ret = DistributedInputTransportBase::GetInstance().OnSessionOpened(sessionId, peerSocketInfo);
     DistributedInputTransportBase::GetInstance().remoteDevSessionMap_.clear();
     EXPECT_EQ(DH_SUCCESS, ret);
 
-    result = DH_SUCCESS;
-    ret = DistributedInputTransportBase::GetInstance().OnSessionOpened(sessionId, result);
+    ret = DistributedInputTransportBase::GetInstance().OnSessionOpened(sessionId, peerSocketInfo);
     EXPECT_EQ(DH_SUCCESS, ret);
 }
 

utils/src/dinput_utils_tool.cpp

@@ -31,11 +31,7 @@
 #include "nlohmann/json.hpp"
 #include "securec.h"
 
-#ifndef COMPILE_TEST_MODE
 #include "softbus_bus_center.h"
-#else
-#include "softbus_bus_center_mock.h"
-#endif
 
 #include "constants_dinput.h"
 #include "dinput_errcode.h"

utils/test/unittest/BUILD.gn

@@ -62,7 +62,6 @@ ohos_unittest("distributed_input_utils_test") {
     "HI_LOG_ENABLE",
     "DH_LOG_TAG=\"distributedinpututtest\"",
     "LOG_DOMAIN=0xD004100",
-    "COMPILE_TEST_MODE",
   ]
 
   deps = [