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