1 /*
2 * Copyright (c) 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 "sensors_errors.h"
21
22 namespace OHOS {
23 namespace Sensors {
24 #ifndef OHOS_BUILD_ENABLE_RUST
25 namespace {
26 constexpr OHOS::HiviewDFX::HiLogLabel LABEL = { LOG_CORE, SENSOR_LOG_DOMAIN, "StreamSocket" };
27 } // namespace
28 #endif // OHOS_BUILD_ENABLE_RUST
29
StreamSocket()30 StreamSocket::StreamSocket() {}
31
~StreamSocket()32 StreamSocket::~StreamSocket()
33 {
34 #ifdef OHOS_BUILD_ENABLE_RUST
35 StreamSocketClose(streamSocketPtr_.get());
36 #else
37 Close();
38 #endif // OHOS_BUILD_ENABLE_RUST
39 }
40
41 #ifndef OHOS_BUILD_ENABLE_RUST
OnReadPackets(CircleStreamBuffer & circBuf,StreamSocket::PacketCallBackFun callbackFun)42 void StreamSocket::OnReadPackets(CircleStreamBuffer &circBuf, StreamSocket::PacketCallBackFun callbackFun)
43 {
44 constexpr size_t headSize = sizeof(PackHead);
45 for (size_t i = 0; i < ONCE_PROCESS_NETPACKET_LIMIT; ++i) {
46 const size_t unreadSize = circBuf.UnreadSize();
47 if (unreadSize < headSize) {
48 break;
49 }
50 size_t dataSize = unreadSize - headSize;
51 char *buf = const_cast<char *>(circBuf.ReadBuf());
52 CHKPB(buf);
53 PackHead *head = reinterpret_cast<PackHead *>(buf);
54 CHKPB(head);
55 if (head->size < 0 || head->size > MAX_PACKET_BUF_SIZE) {
56 SEN_HILOGE("Packet header parsing error, and this error cannot be recovered. The buffer will be reset."
57 " head->size:%{public}zu, unreadSize:%{public}zu", head->size, unreadSize);
58 circBuf.Reset();
59 break;
60 }
61 if (head->size > dataSize) {
62 break;
63 }
64 NetPacket pkt(head->idMsg);
65 if ((head->size > 0) && (!pkt.Write(&buf[headSize], head->size))) {
66 SEN_HILOGW("Error writing data in the NetPacket. It will be retried next time. messageid:%{public}d,"
67 "size:%{public}zu", head->idMsg, head->size);
68 break;
69 }
70 if (!circBuf.SeekReadPos(pkt.GetPacketLength())) {
71 SEN_HILOGW("Set read position error, and this error cannot be recovered, and the buffer will be reset."
72 " packetSize:%{public}zu, unreadSize:%{public}zu", pkt.GetPacketLength(), unreadSize);
73 circBuf.Reset();
74 break;
75 }
76 callbackFun(pkt);
77 if (circBuf.IsEmpty()) {
78 circBuf.Reset();
79 break;
80 }
81 }
82 }
83 #endif // OHOS_BUILD_ENABLE_RUST
84
Close()85 void StreamSocket::Close()
86 {
87 #ifdef OHOS_BUILD_ENABLE_RUST
88 StreamSocketClose(streamSocketPtr_.get());
89 #else
90 if (fd_ >= 0) {
91 auto rf = close(fd_);
92 if (rf != 0) {
93 SEN_HILOGE("Socket close failed, rf:%{public}d", rf);
94 }
95 }
96 fd_ = -1;
97 #endif // OHOS_BUILD_ENABLE_RUST
98 }
99
GetFd() const100 int32_t StreamSocket::GetFd() const
101 {
102 #ifdef OHOS_BUILD_ENABLE_RUST
103 return StreamSocketGetFd(streamSocketPtr_.get());
104 #else
105 return fd_;
106 #endif // OHOS_BUILD_ENABLE_RUST
107 }
108 } // namespace Sensors
109 } // namespace OHOS