1 /*
2 * Copyright (c) 2023 Shenzhen Kaihong Digital Industry Development 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 "udp_session.h"
17 #include <arpa/inet.h>
18 #include <iostream>
19 #include "common/common_macro.h"
20 #include "common/media_log.h"
21 #include "network/socket/socket_utils.h"
22 #include "utils/utils.h"
23
24 namespace OHOS {
25 namespace Sharing {
UdpSession(SocketInfo::Ptr socket)26 UdpSession::UdpSession(SocketInfo::Ptr socket) : BaseNetworkSession(socket)
27 {
28 SHARING_LOGD("trace.");
29 }
30
~UdpSession()31 UdpSession ::~UdpSession()
32 {
33 SHARING_LOGD("trace.");
34 }
35
Start()36 bool UdpSession::Start()
37 {
38 SHARING_LOGD("trace.");
39 if (socket_) {
40 SHARING_LOGD("udpSession AddFdListener.");
41
42 auto eventRunner = OHOS::AppExecFwk::EventRunner::Create(true);
43 eventHandler_ = std::make_shared<UdpSessionEventHandler>();
44 eventHandler_->SetSession(shared_from_this());
45 eventHandler_->SetEventRunner(eventRunner);
46 eventRunner->Run();
47
48 eventListener_ = std::make_shared<UdpSessionEventListener>();
49 eventListener_->SetSession(shared_from_this());
50
51 return eventListener_->AddFdListener(socket_->GetPeerFd(), eventListener_, eventHandler_);
52 }
53
54 return false;
55 }
56
Shutdown()57 void UdpSession::Shutdown()
58 {
59 SHARING_LOGD("trace.");
60 std::lock_guard<std::mutex> lock(mutex_);
61 if (socket_) {
62 if (eventListener_) {
63 eventListener_->RemoveFdListener(socket_->GetPeerFd());
64 }
65 SocketUtils::ShutDownSocket(socket_->GetPeerFd());
66 SocketUtils::CloseSocket(socket_->GetPeerFd());
67 socket_->resetPeerFd();
68 socket_.reset();
69 }
70 }
71
Send(const DataBuffer::Ptr & buf,int32_t nSize)72 bool UdpSession::Send(const DataBuffer::Ptr &buf, int32_t nSize)
73 {
74 MEDIA_LOGD("trace.");
75 return UdpSession::Send(buf->Peek(), nSize);
76 }
77
Send(const char * buf,int32_t nSize)78 bool UdpSession::Send(const char *buf, int32_t nSize)
79 {
80 MEDIA_LOGD("trace.");
81 if (socket_ == nullptr) {
82 SHARING_LOGE("socket nullptr!");
83 return false;
84 }
85 MEDIA_LOGD("fd: %{public}d, address: %{public}s, port: %{public}d, len: %{public}d.", socket_->GetLocalFd(),
86 GetAnonymousIp(ConvertSinAddrToStr(socket_->udpClientAddr_)).c_str(), socket_->udpClientAddr_.sin_port,
87 socket_->udpClientLen_);
88 RETURN_FALSE_IF_NULL(buf);
89
90 int32_t retCode = ::sendto(socket_->GetLocalFd(), buf, nSize, 0, (struct sockaddr *)&socket_->udpClientAddr_,
91 socket_->udpClientLen_);
92 if ((retCode < 0) && (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)) {
93 SHARING_LOGE("send error!");
94 return false;
95 } else if (retCode > 0) {
96 return true;
97 } else {
98 char errmsg[256] = {0};
99 strerror_r(errno, errmsg, sizeof(errmsg));
100 SHARING_LOGE("sendSocket error: %{public}s!", errmsg);
101 return false;
102 }
103 }
104
OnSessionReadble(int32_t fd)105 void UdpSession::OnSessionReadble(int32_t fd)
106 {
107 std::lock_guard<std::mutex> lock(mutex_);
108 if (socket_ == nullptr) {
109 MEDIA_LOGE("socket nullptr!");
110 return;
111 }
112 SHARING_LOGD("fd: %{public}d, local_fd:%{public}d, thread_id: %{public}llu.", fd, socket_->GetLocalFd(),
113 GetThreadId());
114
115 if (fd == socket_->GetLocalFd()) {
116 int32_t retCode = 0;
117 do {
118 DataBuffer::Ptr buf = std::make_shared<DataBuffer>(DEFAULT_READ_BUFFER_SIZE);
119 struct sockaddr_in clientAddr;
120 socklen_t len = sizeof(struct sockaddr_in);
121 retCode = ::recvfrom(fd, buf->Data(), DEFAULT_READ_BUFFER_SIZE, 0, (struct sockaddr *)&clientAddr, &len);
122 MEDIA_LOGD("recvSocket len: %{public}d, address: %{public}s, port: %{public}d.", retCode,
123 GetAnonymousIp(ConvertSinAddrToStr(clientAddr)).c_str(), clientAddr.sin_port);
124
125 if (retCode > 0) {
126 buf->UpdateSize(retCode);
127 if (callback_) {
128 callback_->OnSessionReadData(fd, std::move(buf));
129 }
130 } else if (retCode == 0) {
131 MEDIA_LOGW("recvSocket RET CODE 0!");
132 } else {
133 MEDIA_LOGE("recvSocket error!");
134 }
135 } while (retCode > 0);
136 } else {
137 MEDIA_LOGD("onReadable receive msg.");
138 }
139 }
140 } // namespace Sharing
141 } // namespace OHOS