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_session.h"
17
18 #include <sstream>
19
20 #include "stream_socket.h"
21
22 #undef LOG_TAG
23 #define LOG_TAG "StreamSession"
24
25 namespace OHOS {
26 namespace Sensors {
27
StreamSession(const std::string & programName,const int32_t fd,const int32_t uid,const int32_t pid)28 StreamSession::StreamSession(const std::string &programName, const int32_t fd, const int32_t uid, const int32_t pid)
29 : programName_(programName)
30 #ifdef OHOS_BUILD_ENABLE_RUST
31 {
32 StreamSessionSetFd(streamSessionPtr_.get(), fd);
33 StreamSessionSetUid(streamSessionPtr_.get(), uid);
34 StreamSessionSetPid(streamSessionPtr_.get(), pid);
35 UpdateDescript();
36 }
37 #else
38 ,
39 fd_(fd),
40 uid_(uid),
41 pid_(pid)
42 {
43 UpdateDescript();
44 }
45 #endif // OHOS_BUILD_ENABLE_RUST
46
47
SendMsg(const char * buf,size_t size) const48 bool StreamSession::SendMsg(const char *buf, size_t size) const
49 {
50 #ifdef OHOS_BUILD_ENABLE_RUST
51 return StreamSessionSendMsg(streamSessionPtr_.get(), buf, size);
52 #else
53 CHKPF(buf);
54 if ((size == 0) || (size > MAX_PACKET_BUF_SIZE)) {
55 SEN_HILOGE("buf size:%{public}zu", size);
56 return false;
57 }
58 if (fd_ < 0) {
59 SEN_HILOGE("The fd_ is less than 0");
60 return false;
61 }
62 size_t idx = 0;
63 size_t retryCount = 0;
64 size_t remSize = size;
65 while (remSize > 0 && retryCount < SEND_RETRY_LIMIT) {
66 ++retryCount;
67 auto count = send(fd_, &buf[idx], remSize, MSG_DONTWAIT | MSG_NOSIGNAL);
68 if (count < 0) {
69 if (errno == EAGAIN || errno == EINTR || errno == EWOULDBLOCK) {
70 #ifdef OHOS_BUILD_ENABLE_RUST
71 sleep(Duration::from_micros(SEND_RETRY_SLEEP_TIME));
72 #else
73 usleep(SEND_RETRY_SLEEP_TIME);
74 #endif
75 SEN_HILOGW("Continue for errno EAGAIN|EINTR|EWOULDBLOCK, errno:%{public}d", errno);
76 continue;
77 }
78 SEN_HILOGE("Send return failed, error:%{public}d, fd:%{public}d", errno, fd_);
79 return false;
80 }
81 idx += static_cast<size_t>(count);
82 remSize -= static_cast<size_t>(count);
83 if (remSize > 0) {
84 #ifdef OHOS_BUILD_ENABLE_RUST
85 sleep(Duration::from_micros(SEND_RETRY_SLEEP_TIME));
86 #else
87 usleep(SEND_RETRY_SLEEP_TIME);
88 #endif
89 }
90 }
91 if (retryCount >= SEND_RETRY_LIMIT || remSize != 0) {
92 SEN_HILOGE("Send too many times:%{public}zu/%{public}zu, size:%{public}zu/%{public}zu, fd:%{public}d",
93 retryCount, SEND_RETRY_LIMIT, idx, size, fd_);
94 return false;
95 }
96 return true;
97 #endif // OHOS_BUILD_ENABLE_RUST
98 }
99
Close()100 void StreamSession::Close()
101 {
102 #ifdef OHOS_BUILD_ENABLE_RUST
103 StreamSessionClose(streamSessionPtr_.get());
104 UpdateDescript();
105 #else
106 if (fd_ >= 0) {
107 close(fd_);
108 fd_ = -1;
109 UpdateDescript();
110 }
111 #endif // OHOS_BUILD_ENABLE_RUST
112 }
113
UpdateDescript()114 void StreamSession::UpdateDescript()
115 {
116 #ifdef OHOS_BUILD_ENABLE_RUST
117 std::ostringstream oss;
118 oss << "fd = " << StreamSessionGetFd(streamSessionPtr_.get())
119 << ", programName = " << programName_
120 << ", moduleType = " << StreamSessionGetModuleType(streamSessionPtr_.get())
121 << ((StreamSessionGetFd(streamSessionPtr_.get()) < 0) ? ", closed" : ", opened")
122 << ", uid = " << StreamSessionGetUid(streamSessionPtr_.get())
123 << ", pid = " << StreamSessionGetPid(streamSessionPtr_.get())
124 << ", tokenType = " << StreamSessionGetTokenType(streamSessionPtr_.get())
125 << std::endl;
126 descript_ = oss.str().c_str();
127 #else
128 std::ostringstream oss;
129 oss << "fd = " << fd_
130 << ", programName = " << programName_
131 << ((fd_ < 0) ? ", closed" : ", opened")
132 << ", uid = " << uid_
133 << ", pid = " << pid_
134 << ", tokenType = " << tokenType_
135 << std::endl;
136 descript_ = oss.str().c_str();
137 #endif // OHOS_BUILD_ENABLE_RUST
138 }
139
SendMsg(const NetPacket & pkt) const140 bool StreamSession::SendMsg(const NetPacket &pkt) const
141 {
142 #ifdef OHOS_BUILD_ENABLE_RUST
143 if (StreamBufferChkRWError(pkt.streamBufferPtr_.get())) {
144 SEN_HILOGE("Read and write status is error");
145 return false;
146 }
147 StreamBuffer buf;
148 pkt.MakeData(buf);
149 return SendMsg(StreamBufferData(buf.streamBufferPtr_.get()), StreamBufferSize(buf.streamBufferPtr_.get()));
150 #else
151 if (pkt.ChkRWError()) {
152 SEN_HILOGE("Read and write status failed");
153 return false;
154 }
155 StreamBuffer buf;
156 pkt.MakeData(buf);
157 return SendMsg(buf.Data(), buf.Size());
158 #endif // OHOS_BUILD_ENABLE_RUST
159 }
160
GetUid() const161 int32_t StreamSession::GetUid() const
162 {
163 #ifdef OHOS_BUILD_ENABLE_RUST
164 return StreamSessionGetUid(streamSessionPtr_.get());
165 #else
166 return uid_;
167 #endif // OHOS_BUILD_ENABLE_RUST
168 }
169
GetPid() const170 int32_t StreamSession::GetPid() const
171 {
172 #ifdef OHOS_BUILD_ENABLE_RUST
173 return StreamSessionGetPid(streamSessionPtr_.get());
174 #else
175 return pid_;
176 #endif // OHOS_BUILD_ENABLE_RUST
177 }
178
GetSharedPtr()179 SessionPtr StreamSession::GetSharedPtr()
180 {
181 return shared_from_this();
182 }
183
GetFd() const184 int32_t StreamSession::GetFd() const
185 {
186 #ifdef OHOS_BUILD_ENABLE_RUST
187 return StreamSessionGetFd(streamSessionPtr_.get());
188 #else
189 return fd_;
190 #endif // OHOS_BUILD_ENABLE_RUST
191 }
192
GetDescript() const193 const std::string &StreamSession::GetDescript() const
194 {
195 return descript_;
196 }
197
GetProgramName() const198 const std::string StreamSession::GetProgramName() const
199 {
200 return programName_;
201 }
202
SetTokenType(int32_t type)203 void StreamSession::SetTokenType(int32_t type)
204 {
205 #ifdef OHOS_BUILD_ENABLE_RUST
206 StreamSessionSetTokenType(streamSessionPtr_.get(), type);
207 #else
208 tokenType_ = type;
209 #endif // OHOS_BUILD_ENABLE_RUST
210 }
211
GetTokenType() const212 int32_t StreamSession::GetTokenType() const
213 {
214 #ifdef OHOS_BUILD_ENABLE_RUST
215 return StreamSessionGetTokenType(streamSessionPtr_.get());
216 #else
217 return tokenType_;
218 #endif // OHOS_BUILD_ENABLE_RUST
219 }
220 } // namespace Sensors
221 } // namespace OHOS