1 /*
2 * Copyright (c) 2021-2023 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 "stream_socket.h"
17
18 #include <cinttypes>
19
20 #include "devicestatus_define.h"
21 #include "fi_log.h"
22
23 #undef LOG_TAG
24 #define LOG_TAG "StreamSocket"
25
26 namespace OHOS {
27 namespace Msdp {
28 namespace DeviceStatus {
29
StreamSocket()30 StreamSocket::StreamSocket() {}
31
~StreamSocket()32 StreamSocket::~StreamSocket()
33 {
34 Close();
35 EpollClose();
36 }
37
EpollCreate()38 int32_t StreamSocket::EpollCreate()
39 {
40 CALL_INFO_TRACE;
41 epollFd_ = ::epoll_create1(EPOLL_CLOEXEC);
42 if (epollFd_ < 0) {
43 FI_HILOGE("epoll_create1 failed:%{public}s", ::strerror(errno));
44 return RET_ERR;
45 }
46 return RET_OK;
47 }
48
EpollCtl(int32_t fd,int32_t op,struct epoll_event & event)49 int32_t StreamSocket::EpollCtl(int32_t fd, int32_t op, struct epoll_event &event)
50 {
51 if (fd < 0) {
52 FI_HILOGE("Invalid fd:%{public}d", fd);
53 return RET_ERR;
54 }
55 if (epollFd_ < 0) {
56 FI_HILOGE("Invalid epollFd:%{public}d", epollFd_);
57 return RET_ERR;
58 }
59 if (::epoll_ctl(epollFd_, op, fd, &event) != 0) {
60 FI_HILOGE("epoll_ctl(%{public}d,%{public}d,%{public}d) failed:%{public}s", epollFd_, op, fd, ::strerror(errno));
61 return RET_ERR;
62 }
63 return RET_OK;
64 }
65
EpollWait(int32_t maxevents,int32_t timeout,struct epoll_event & events)66 int32_t StreamSocket::EpollWait(int32_t maxevents, int32_t timeout, struct epoll_event &events)
67 {
68 if (epollFd_ < 0) {
69 FI_HILOGE("Invalid epollFd:%{public}d", epollFd_);
70 return RET_ERR;
71 }
72 return epoll_wait(epollFd_, &events, maxevents, timeout);
73 }
74
OnReadPackets(CircleStreamBuffer & circBuf,StreamSocket::PacketCallBackFun callbackFun)75 void StreamSocket::OnReadPackets(CircleStreamBuffer &circBuf, StreamSocket::PacketCallBackFun callbackFun)
76 {
77 constexpr int32_t headSize = static_cast<int32_t>(sizeof(PackHead));
78 for (int32_t i = 0; i < ONCE_PROCESS_NETPACKET_LIMIT; i++) {
79 const int32_t residualSize = circBuf.ResidualSize();
80 if (residualSize < headSize) {
81 break;
82 }
83 int32_t dataSize = residualSize - headSize;
84 char *buf = const_cast<char *>(circBuf.ReadBuf());
85 CHKPB(buf);
86 PackHead *head = reinterpret_cast<PackHead *>(buf);
87 CHKPB(head);
88 if ((static_cast<int32_t>(head->size) < 0) || (static_cast<size_t>(head->size) > MAX_PACKET_BUF_SIZE)) {
89 FI_HILOGE("Packet header parsing error, and this error cannot be recovered, the buffer will be reset, "
90 "head->size:%{public}d, residualSize:%{public}d", head->size, residualSize);
91 circBuf.Reset();
92 break;
93 }
94 if (head->size > dataSize) {
95 break;
96 }
97 NetPacket pkt(head->idMsg);
98 if ((head->size > 0) && (!pkt.Write(&buf[headSize], head->size))) {
99 FI_HILOGW("Error writing data in the NetPacket, it will be retried next time, messageid:%{public}d, "
100 "size:%{public}d", head->idMsg, head->size);
101 break;
102 }
103 if (!circBuf.SeekReadPos(pkt.GetPacketLength())) {
104 FI_HILOGW("Set read position error, and this error cannot be recovered, and the buffer will be reset, "
105 "packetSize:%{public}d, residualSize:%{public}d", pkt.GetPacketLength(), residualSize);
106 circBuf.Reset();
107 break;
108 }
109 callbackFun(pkt);
110 if (circBuf.empty()) {
111 circBuf.Reset();
112 break;
113 }
114 }
115 }
116
EpollClose()117 void StreamSocket::EpollClose()
118 {
119 if (epollFd_ >= 0) {
120 if (close(epollFd_) < 0) {
121 FI_HILOGE("Close epoll fd failed, error:%{public}s, epollFd_:%{public}d", strerror(errno), epollFd_);
122 }
123 epollFd_ = -1;
124 }
125 }
126
Close()127 void StreamSocket::Close()
128 {
129 if (fd_ >= 0) {
130 int32_t rf = close(fd_);
131 if (rf < 0) {
132 FI_HILOGE("Socket close failed rf:%{public}d", rf);
133 }
134 }
135 fd_ = -1;
136 }
137 } // namespace DeviceStatus
138 } // namespace Msdp
139 } // namespace OHOS