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 23namespace OHOS { 24namespace Msdp { 25namespace DeviceStatus { 26 27StreamSocket::StreamSocket() {} 28 29StreamSocket::~StreamSocket() 30{ 31 Close(); 32 EpollClose(); 33} 34 35int32_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 46int32_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 63int32_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 72void 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 114void 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 124void 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