鸿蒙子系统解读-分布式软总线子系统初步研究（下）

鸿蒙子系统解读-分布式软总线子系统初步研究（下）

本文作者：江苏润和软件股份有限公司 郎建中

3.2. 数据传输

上一节我们看了PublishService()函数的代码实现过程。被发现端发布服务后，软总线回调发布成功的函数中，被发现端需要使用CreateSessionServer()来创建会话的服务器等待发现端的连接创建会话。下面我们就来分析CreateSessionServer()函数的实现过程。

数据传输部分的代码位于foundation/communication/services/softbus_lite/trans_service目录中。其目录结构如下：

CreateSessionServer()函数的实现就在tcp_session_manager.c中。我们看看这个函数的代码实现：

SoftBusCheckPermission()前面有介绍过，就是检查权限。

GetTcpMgrLock()和ReleaseTcpMgrLock()就是通过g_sessionManagerLock这个Mutex进行互斥操作，防止同一时间多个模块同时在创建Session Server。

真正的业务逻辑在CreateSessionServerInner()函数中，我们来看看代码：

上面SessionListenerMap结构中，最重要的是listener成员，这个成员的结构如下：

其中：

onSessionOpened：是在会话创建时被回调的函数

onSessionClosed：是在会话结束时被回调的函数

onBytesReceived：是会话的数据到达的回调函数，注册的模块可以通过这个函数接收会话的报文，按照自己的格式进行解析，并执行会话要求的动作。例如：在分布式调度模块中，接收的数据解析后，可能是START_FA的命令。

上一节我们介绍过，在StartBus()函数会调用StartSession()函数创建基于TCP的socket的会话管理服务。这个服务的任务线程是SelectSessionLoop()。下面我们来分析下这个线程：

这个线程的主体结构就是一个while循环，这个循环只有在maxFd为负值的时候才会退出循环。maxFd是InitSelectList()函数的返回值，大家看下代码可以知道，只有出错的情况才会返回负值。InitSelectList()函数就是根据g_sessionMgr->sessionMap_[]数组以及g_sessionMgr->listenFd构造一个fd_set，这个fd_set将在下面的select中使用。

g_sessionMgr->listenFd：就是前面介绍过在StartBus()流程里面创建的本地会话服务器的基于TCP的socket。这个socket负责会话的连接请求，所有其他智慧屏设备都是向这个轻量设备侧的socket发起连接请求的。

g_sessionMgr->sessionMap_[]：这个里面放的是TcpSession类型的数组。当连接请求被接受后，本地会生成一个新的socket用于数据的传输，这个socket就被放在TcpSession类型里面。所以g_sessionMgr->sessionMap_[]里面放的是所有的已经建立连接的会话的socket。

因此，上面的InitSelectList()函数得到的是 ListenFd + 所有已经建立连接的socket这样一个fd_set。那么select()函数的功能就是等待这些所有socket上是否有数据发生，如果有数据发生就会往下走去处理数据。

处理数据是在ProcessData()函数里面，代码如下：

这个函数传入的参数 tsm就是 g_sessionMgr , rfds就是前面 InitSelectList()函数得到的 ListenFd + 所有已经建立连接的socket这样一个fd_set。

FD_ISSET()语句就是判断rfds 中的listenFd是否被置位了，也就是监听连接的socket是否有数据了。如果有数据就用 ProcessConnection()函数去处理连接请求。这个函数里面会在accept后新生成一个socket，然后构造一个TcpSession把新生成用于数据传输的socket放在TcpSession里面，然后再把TCPSession放在g_sessionMgr->sessionMap_[]数组里面。当下一次执行InitSelectList()的时候，fd_set中就会包含这个新建立连接的socket了。关于ProcessConnection()函数就是处理连接请求并进行认证，这里不再展开，大家有兴趣可以自己看下代码。

最后就是ProcessSessionData()函数的调用。这个函数就是处理已经建立连接的那些会话的数据。我们看下代码：

从代码看到，for循环就是针对g_sessionMgr->sessionMap_[]的每一个TcpSession对象，这里每一个对象就代表一个会话连接。通过FD_ISSET()来判断这个会话连接是否有数据需要处理，如果有则调用OnProcessDataAvailable()来处理数据。大家可以看到所有的会话的数据是串行处理的，只有当一个会话处理完才会处理另一个会话，这在并行同时有几个会话的情况下，可能会影响性能。

我们来看看OnProcessDataAvailable()函数的实现：

这个函数的代码首先是判断会话的sessionName是否未知的。正常情况应该不会走到这里。所以我们看else语句部分。

GetSessionListenerByName()函数就是通过sessionName从g_sessionMgr->serverListenerMap[]数组中找到匹配的SessionListenerMap对象。g_sessionMgr->serverListenerMap[]就是我们前面介绍过的CreateSessionServerInner()函数中添加进去的。

下面调用TcpSessionRecv()函数进行数据的接收和解码。这里代码不再展开，有兴趣同学自己看一下。最后数据被保存在buf中。

当数据接收完成后，就通过SessionListenerMap对象中的onBytesReceived()回调函数回调给软总线的使用者。

流程图如下：

总结：

1、软总线的使用者调用CreateSessionServer()创建Session会话。

2、SelectSessionLoop()线程处理所有的会话数据接收

3、通过调用使用者注册的onBytesReceived()函数回调给使用者（这里可以看下《分布式调度子系统》中如何拉起FA的过程）