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 if (info->socketId <= 0) {
167 fd = CreateSocket(info->baseInfo.flags, info->server);
168 LE_CHECK(fd > 0, return LE_FAILURE, "Failed to create socket %s", info->server);
169 }
170
171 EventLoop *loop = (EventLoop *)loopHandle;
172 StreamServerTask *task = (StreamServerTask *)CreateTask(loopHandle, fd, &info->baseInfo,
173 sizeof(StreamServerTask) + strlen(info->server) + 1);
174 LE_CHECK(task != NULL, close(fd);
175 return LE_NO_MEMORY, "Failed to create task");
176 task->base.handleEvent = HandleServerEvent_;
177 task->base.innerClose = HandleStreamTaskClose_;
178 task->incommingConnect = info->incommingConnect;
179 loop->addEvent(loop, (const BaseTask *)task, Event_Read);
180 int ret = memcpy_s(task->server, strlen(info->server) + 1, info->server, strlen(info->server) + 1);
181 LE_CHECK(ret == 0, return LE_FAILURE, "Failed to copy server name %s", info->server);
182 *taskHandle = (TaskHandle)task;
183 return LE_SUCCESS;
184 }
185
LE_CreateStreamClient(const LoopHandle loopHandle,TaskHandle * taskHandle,const LE_StreamInfo * info)186 LE_STATUS LE_CreateStreamClient(const LoopHandle loopHandle,
187 TaskHandle *taskHandle, const LE_StreamInfo *info)
188 {
189 LE_CHECK(loopHandle != NULL && taskHandle != NULL && info != NULL, return LE_INVALID_PARAM, "Invalid parameters");
190 LE_CHECK(info->recvMessage != NULL, return LE_FAILURE, "Invalid parameters recvMessage %s", info->server);
191
192 int fd = CreateSocket(info->baseInfo.flags, info->server);
193 LE_CHECK(fd > 0, return LE_FAILURE, "Failed to create socket %s", info->server);
194
195 StreamClientTask *task = (StreamClientTask *)CreateTask(loopHandle, fd, &info->baseInfo, sizeof(StreamClientTask));
196 LE_CHECK(task != NULL, close(fd);
197 return LE_NO_MEMORY, "Failed to create task");
198 task->stream.base.handleEvent = HandleClientEvent_;
199 task->stream.base.innerClose = HandleStreamTaskClose_;
200 OH_ListInit(&task->stream.buffHead);
201 LoopMutexInit(&task->stream.mutex);
202
203 task->connectComplete = info->connectComplete;
204 task->sendMessageComplete = info->sendMessageComplete;
205 task->recvMessage = info->recvMessage;
206 task->disConnectComplete = info->disConnectComplete;
207 EventLoop *loop = (EventLoop *)loopHandle;
208 loop->addEvent(loop, (const BaseTask *)task, Event_Read);
209 *taskHandle = (TaskHandle)task;
210 return LE_SUCCESS;
211 }
212
LE_AcceptStreamClient(const LoopHandle loopHandle,const TaskHandle server,TaskHandle * taskHandle,const LE_StreamInfo * info)213 LE_STATUS LE_AcceptStreamClient(const LoopHandle loopHandle, const TaskHandle server,
214 TaskHandle *taskHandle, const LE_StreamInfo *info)
215 {
216 LE_CHECK(loopHandle != NULL && info != NULL, return LE_INVALID_PARAM, "Invalid parameters");
217 LE_CHECK(server != NULL && taskHandle != NULL, return LE_INVALID_PARAM, "Invalid parameters");
218 LE_CHECK(info->recvMessage != NULL, return LE_INVALID_PARAM, "Invalid parameters recvMessage");
219 int fd = -1;
220 if ((info->baseInfo.flags & TASK_TEST) != TASK_TEST) {
221 fd = AcceptSocket(GetSocketFd(server), info->baseInfo.flags);
222 LE_CHECK(fd > 0, return LE_FAILURE, "Failed to accept socket %d", GetSocketFd(server));
223 }
224 StreamConnectTask *task = (StreamConnectTask *)CreateTask(
225 loopHandle, fd, &info->baseInfo, sizeof(StreamConnectTask));
226 LE_CHECK(task != NULL, close(fd);
227 return LE_NO_MEMORY, "Failed to create task");
228 task->stream.base.handleEvent = HandleStreamEvent_;
229 task->stream.base.innerClose = HandleStreamTaskClose_;
230 task->disConnectComplete = info->disConnectComplete;
231 task->sendMessageComplete = info->sendMessageComplete;
232 task->recvMessage = info->recvMessage;
233 task->serverTask = (StreamServerTask *)server;
234 OH_ListInit(&task->stream.buffHead);
235 LoopMutexInit(&task->stream.mutex);
236 if ((info->baseInfo.flags & TASK_TEST) != TASK_TEST) {
237 EventLoop *loop = (EventLoop *)loopHandle;
238 loop->addEvent(loop, (const BaseTask *)task, Event_Read);
239 }
240 *taskHandle = (TaskHandle)task;
241 return 0;
242 }
243
LE_CloseStreamTask(const LoopHandle loopHandle,const TaskHandle taskHandle)244 void LE_CloseStreamTask(const LoopHandle loopHandle, const TaskHandle taskHandle)
245 {
246 LE_CHECK(loopHandle != NULL && taskHandle != NULL, return, "Invalid parameters");
247 LE_CloseTask(loopHandle, taskHandle);
248 }
249
LE_GetSocketFd(const TaskHandle taskHandle)250 int LE_GetSocketFd(const TaskHandle taskHandle)
251 {
252 LE_CHECK(taskHandle != NULL, return -1, "Invalid parameters");
253 return GetSocketFd(taskHandle);
254 }