1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "le_loop.h"
17
18 #include <errno.h>
19 #include <sys/socket.h>
20
21 #include "le_socket.h"
22 #include "le_task.h"
23
HandleSendMsg_(const LoopHandle loopHandle,const TaskHandle taskHandle,const LE_SendMessageComplete complete)24 static LE_STATUS HandleSendMsg_(const LoopHandle loopHandle,
25 const TaskHandle taskHandle, const LE_SendMessageComplete complete)
26 {
27 EventLoop *loop = (EventLoop *)loopHandle;
28 StreamTask *stream = (StreamTask *)taskHandle;
29 LE_Buffer *buffer = GetFirstBuffer(stream);
30 while (buffer) {
31 int ret = write(GetSocketFd(taskHandle), buffer->data, buffer->dataSize);
32 LE_LOGV("HandleSendMsg_ fd:%d send data size %d %d", GetSocketFd(taskHandle), buffer->dataSize, ret);
33 buffer->result = (ret == (int)buffer->dataSize) ? 0 : errno;
34 if (complete != NULL) {
35 complete(taskHandle, buffer);
36 }
37 FreeBuffer(loopHandle, stream, buffer);
38 buffer = GetFirstBuffer(stream);
39 }
40 if (IsBufferEmpty(stream)) {
41 LE_LOGV("HandleSendMsg_ fd:%d empty wait read", GetSocketFd(taskHandle));
42 loop->modEvent(loop, (const BaseTask *)taskHandle, Event_Read);
43 return LE_SUCCESS;
44 }
45 return LE_SUCCESS;
46 }
47
HandleRecvMsg_(const LoopHandle loopHandle,const TaskHandle taskHandle,const LE_RecvMessage recvMessage)48 static LE_STATUS HandleRecvMsg_(const LoopHandle loopHandle,
49 const TaskHandle taskHandle, const LE_RecvMessage recvMessage)
50 {
51 LE_STATUS status = LE_SUCCESS;
52 LE_Buffer *buffer = CreateBuffer(LOOP_DEFAULT_BUFFER);
53 int readLen = 0;
54 while (1) {
55 readLen = recv(GetSocketFd(taskHandle), buffer->data, LOOP_DEFAULT_BUFFER, 0);
56 LE_LOGV("HandleRecvMsg fd:%d read msg len %d", GetSocketFd(taskHandle), readLen);
57 if (readLen < 0) {
58 if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN) {
59 continue;
60 }
61 status = LE_DIS_CONNECTED;
62 break;
63 } else if (readLen == 0) {
64 // 若另一端已关闭连接则返回0，这种关闭是对方主动且正常的关闭
65 status = LE_DIS_CONNECTED;
66 break;
67 } else {
68 break;
69 }
70 }
71 if (status != LE_SUCCESS) {
72 FreeBuffer(loopHandle, NULL, buffer);
73 return status;
74 }
75 if (recvMessage) {
76 recvMessage(taskHandle, buffer->data, readLen);
77 }
78 FreeBuffer(loopHandle, NULL, buffer);
79 return status;
80 }
81
HandleStreamEvent_(const LoopHandle loopHandle,const TaskHandle handle,uint32_t oper)82 static LE_STATUS HandleStreamEvent_(const LoopHandle loopHandle, const TaskHandle handle, uint32_t oper)
83 {
84 EventLoop *loop = (EventLoop *)loopHandle;
85 StreamConnectTask *stream = (StreamConnectTask *)handle;
86 LE_LOGV("HandleStreamEvent_ fd:%d oper 0x%x", GetSocketFd(handle), oper);
87
88 LE_STATUS status = LE_SUCCESS;
89 if (LE_TEST_FLAGS(oper, Event_Write)) {
90 status = HandleSendMsg_(loopHandle, handle, stream->sendMessageComplete);
91 }
92 if (LE_TEST_FLAGS(oper, Event_Read)) {
93 status = HandleRecvMsg_(loopHandle, handle, stream->recvMessage);
94 }
95 if (status == LE_DIS_CONNECTED) {
96 loop->delEvent(loop, GetSocketFd(handle), Event_Read | Event_Write);
97 if (stream->disConnectComplete) {
98 stream->disConnectComplete(handle);
99 }
100 LE_CloseStreamTask(loopHandle, handle);
101 }
102 return status;
103 }
104
HandleClientEvent_(const LoopHandle loopHandle,const TaskHandle handle,uint32_t oper)105 static LE_STATUS HandleClientEvent_(const LoopHandle loopHandle, const TaskHandle handle, uint32_t oper)
106 {
107 StreamClientTask *client = (StreamClientTask *)handle;
108 LE_LOGV("HandleClientEvent_ fd:%d oper 0x%x", GetSocketFd(handle), oper);
109
110 LE_STATUS status = LE_SUCCESS;
111 if (LE_TEST_FLAGS(oper, Event_Write)) {
112 LE_ONLY_CHECK(!(client->connected == 0 && client->connectComplete), client->connectComplete(handle));
113 client->connected = 1;
114 status = HandleSendMsg_(loopHandle, handle, client->sendMessageComplete);
115 }
116 if (LE_TEST_FLAGS(oper, Event_Read)) {
117 status = HandleRecvMsg_(loopHandle, handle, client->recvMessage);
118 }
119 if (status == LE_DIS_CONNECTED) {
120 if (client->disConnectComplete) {
121 client->disConnectComplete(handle);
122 }
123 client->connected = 0;
124 LE_CloseStreamTask(loopHandle, handle);
125 }
126 return status;
127 }
128
HandleStreamTaskClose_(const LoopHandle loopHandle,const TaskHandle taskHandle)129 static void HandleStreamTaskClose_(const LoopHandle loopHandle, const TaskHandle taskHandle)
130 {
131 BaseTask *task = (BaseTask *)taskHandle;
132 CloseTask(loopHandle, task);
133 DelTask((EventLoop *)loopHandle, task);
134 if (task->taskId.fd > 0) {
135 close(task->taskId.fd);
136 }
137 }
138
HandleServerEvent_(const LoopHandle loopHandle,const TaskHandle serverTask,uint32_t oper)139 static LE_STATUS HandleServerEvent_(const LoopHandle loopHandle, const TaskHandle serverTask, uint32_t oper)
140 {
141 LE_LOGV("HandleServerEvent_ fd %d oper 0x%x", GetSocketFd(serverTask), oper);
142 if (!LE_TEST_FLAGS(oper, Event_Read)) {
143 return LE_FAILURE;
144 }
145 StreamServerTask *server = (StreamServerTask *)serverTask;
146 LE_ONLY_CHECK(server->incommingConnect != NULL, return LE_SUCCESS);
147
148 int ret = server->incommingConnect(loopHandle, serverTask);
149 if (ret != LE_SUCCESS) {
150 LE_LOGE("HandleServerEvent_ fd %d do not accept socket", GetSocketFd(serverTask));
151 }
152 EventLoop *loop = (EventLoop *)loopHandle;
153 loop->modEvent(loop, (const BaseTask *)serverTask, Event_Read);
154 return LE_SUCCESS;
155 }
156
LE_CreateStreamServer(const LoopHandle loopHandle,TaskHandle * taskHandle,const LE_StreamServerInfo * info)157 LE_STATUS LE_CreateStreamServer(const LoopHandle loopHandle,
158 TaskHandle *taskHandle, const LE_StreamServerInfo *info)
159 {
160 LE_CHECK(loopHandle != NULL && taskHandle != NULL && info != NULL, return LE_INVALID_PARAM, "Invalid parameters");
161 LE_CHECK(info->server != NULL, return LE_INVALID_PARAM, "Invalid parameters server");
162 LE_CHECK(info->incommingConnect != NULL, return LE_INVALID_PARAM,
163 "Invalid parameters incommingConnect %s", info->server);
164
165 int fd = info->socketId;
166 int ret = 0;
167 if (info->socketId <= 0) {
168 fd = CreateSocket(info->baseInfo.flags, info->server);
169 LE_CHECK(fd > 0, return LE_FAILURE, "Failed to create socket %s", info->server);
170 } else {
171 ret = listenSocket(fd, info->baseInfo.flags, info->server);
172 LE_CHECK(ret == 0, return LE_FAILURE, "Failed to listen socket %s", info->server);
173 }
174
175 EventLoop *loop = (EventLoop *)loopHandle;
176 StreamServerTask *task = (StreamServerTask *)CreateTask(loopHandle, fd, &info->baseInfo,
177 sizeof(StreamServerTask) + strlen(info->server) + 1);
178 LE_CHECK(task != NULL, close(fd);
179 return LE_NO_MEMORY, "Failed to create task");
180 task->base.handleEvent = HandleServerEvent_;
181 task->base.innerClose = HandleStreamTaskClose_;
182 task->incommingConnect = info->incommingConnect;
183 loop->addEvent(loop, (const BaseTask *)task, Event_Read);
184 ret = memcpy_s(task->server, strlen(info->server) + 1, info->server, strlen(info->server) + 1);
185 LE_CHECK(ret == 0, return LE_FAILURE, "Failed to copy server name %s", info->server);
186 *taskHandle = (TaskHandle)task;
187 return LE_SUCCESS;
188 }
189
LE_CreateStreamClient(const LoopHandle loopHandle,TaskHandle * taskHandle,const LE_StreamInfo * info)190 LE_STATUS LE_CreateStreamClient(const LoopHandle loopHandle,
191 TaskHandle *taskHandle, const LE_StreamInfo *info)
192 {
193 LE_CHECK(loopHandle != NULL && taskHandle != NULL && info != NULL, return LE_INVALID_PARAM, "Invalid parameters");
194 LE_CHECK(info->recvMessage != NULL, return LE_FAILURE, "Invalid parameters recvMessage %s", info->server);
195
196 int fd = CreateSocket(info->baseInfo.flags, info->server);
197 LE_CHECK(fd > 0, return LE_FAILURE, "Failed to create socket %s", info->server);
198
199 StreamClientTask *task = (StreamClientTask *)CreateTask(loopHandle, fd, &info->baseInfo, sizeof(StreamClientTask));
200 LE_CHECK(task != NULL, close(fd);
201 return LE_NO_MEMORY, "Failed to create task");
202 task->stream.base.handleEvent = HandleClientEvent_;
203 task->stream.base.innerClose = HandleStreamTaskClose_;
204 OH_ListInit(&task->stream.buffHead);
205 LoopMutexInit(&task->stream.mutex);
206
207 task->connectComplete = info->connectComplete;
208 task->sendMessageComplete = info->sendMessageComplete;
209 task->recvMessage = info->recvMessage;
210 task->disConnectComplete = info->disConnectComplete;
211 EventLoop *loop = (EventLoop *)loopHandle;
212 loop->addEvent(loop, (const BaseTask *)task, Event_Read);
213 *taskHandle = (TaskHandle)task;
214 return LE_SUCCESS;
215 }
216
LE_AcceptStreamClient(const LoopHandle loopHandle,const TaskHandle server,TaskHandle * taskHandle,const LE_StreamInfo * info)217 LE_STATUS LE_AcceptStreamClient(const LoopHandle loopHandle, const TaskHandle server,
218 TaskHandle *taskHandle, const LE_StreamInfo *info)
219 {
220 LE_CHECK(loopHandle != NULL && info != NULL, return LE_INVALID_PARAM, "Invalid parameters");
221 LE_CHECK(server != NULL && taskHandle != NULL, return LE_INVALID_PARAM, "Invalid parameters");
222 LE_CHECK(info->recvMessage != NULL, return LE_INVALID_PARAM, "Invalid parameters recvMessage");
223 int fd = -1;
224 if ((info->baseInfo.flags & TASK_TEST) != TASK_TEST) {
225 fd = AcceptSocket(GetSocketFd(server), info->baseInfo.flags);
226 LE_CHECK(fd > 0, return LE_FAILURE, "Failed to accept socket %d", GetSocketFd(server));
227 }
228 StreamConnectTask *task = (StreamConnectTask *)CreateTask(
229 loopHandle, fd, &info->baseInfo, sizeof(StreamConnectTask));
230 LE_CHECK(task != NULL, close(fd);
231 return LE_NO_MEMORY, "Failed to create task");
232 task->stream.base.handleEvent = HandleStreamEvent_;
233 task->stream.base.innerClose = HandleStreamTaskClose_;
234 task->disConnectComplete = info->disConnectComplete;
235 task->sendMessageComplete = info->sendMessageComplete;
236 task->recvMessage = info->recvMessage;
237 task->serverTask = (StreamServerTask *)server;
238 OH_ListInit(&task->stream.buffHead);
239 LoopMutexInit(&task->stream.mutex);
240 if ((info->baseInfo.flags & TASK_TEST) != TASK_TEST) {
241 EventLoop *loop = (EventLoop *)loopHandle;
242 loop->addEvent(loop, (const BaseTask *)task, Event_Read);
243 }
244 *taskHandle = (TaskHandle)task;
245 return 0;
246 }
247
LE_CloseStreamTask(const LoopHandle loopHandle,const TaskHandle taskHandle)248 void LE_CloseStreamTask(const LoopHandle loopHandle, const TaskHandle taskHandle)
249 {
250 LE_CHECK(loopHandle != NULL && taskHandle != NULL, return, "Invalid parameters");
251 LE_CloseTask(loopHandle, taskHandle);
252 }
253
LE_GetSocketFd(const TaskHandle taskHandle)254 int LE_GetSocketFd(const TaskHandle taskHandle)
255 {
256 LE_CHECK(taskHandle != NULL, return -1, "Invalid parameters");
257 return GetSocketFd(taskHandle);
258 }