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_client.h"
17 #include <unistd.h>
18 #include "common/media_log.h"
19 #include "network/socket/socket_utils.h"
20 #include "network/socket/udp_socket.h"
21 #include "utils/utils.h"
22 namespace OHOS {
23 namespace Sharing {
24
~UdpClient()25 UdpClient::~UdpClient()
26 {
27 SHARING_LOGD("trace.");
28 Disconnect();
29 }
30
UdpClient()31 UdpClient::UdpClient()
32 {
33 SHARING_LOGD("trace.");
34 }
35
Connect(const std::string & peerHost,uint16_t peerPort,const std::string & localIp,uint16_t localPort)36 bool UdpClient::Connect(const std::string &peerHost, uint16_t peerPort, const std::string &localIp, uint16_t localPort)
37 {
38 SHARING_LOGD("peerIp:%{public}s, peerPort:%{public}d, thread_id: %{public}llu.", peerHost.c_str(), peerPort,
39 GetThreadId());
40 std::unique_lock<std::shared_mutex> lk(mutex_);
41 int32_t retCode = 0;
42 socket_ = std::make_unique<UdpSocket>();
43 if (socket_) {
44 if (socket_->Connect(peerHost, peerPort, retCode, false, true, localIp, localPort)) {
45 SHARING_LOGD("connect success.");
46 auto eventRunner = OHOS::AppExecFwk::EventRunner::Create(true);
47 eventHandler_ = std::make_shared<UdpClientEventHandler>();
48 eventHandler_->SetClient(shared_from_this());
49 eventHandler_->SetEventRunner(eventRunner);
50 eventRunner->Run();
51
52 eventListener_ = std::make_shared<UdpClientEventListener>();
53 eventListener_->SetClient(shared_from_this());
54
55 bool ret = eventListener_->AddFdListener(socket_->GetLocalFd(), eventListener_, eventHandler_);
56
57 auto callback = callback_.lock();
58 if (callback) {
59 callback->OnClientConnect(true);
60 }
61
62 return ret;
63 }
64 }
65 auto callback = callback_.lock();
66 if (callback) {
67 callback->OnClientConnect(false);
68 }
69 SHARING_LOGE("connect failed!");
70 return false;
71 }
72
Disconnect()73 void UdpClient::Disconnect()
74 {
75 SHARING_LOGD("trace.");
76 std::unique_lock<std::shared_mutex> lk(mutex_);
77 if (socket_ != nullptr) {
78 eventListener_->RemoveFdListener(socket_->GetLocalFd());
79 SocketUtils::ShutDownSocket(socket_->GetLocalFd());
80 SocketUtils::CloseSocket(socket_->GetLocalFd());
81 socket_.reset();
82 }
83 }
84
Send(const DataBuffer::Ptr & buf,int32_t nSize)85 bool UdpClient::Send(const DataBuffer::Ptr &buf, int32_t nSize)
86 {
87 SHARING_LOGD("trace.");
88 return Send(buf->Peek(), nSize);
89 }
90
Send(const char * buf,int32_t nSize)91 bool UdpClient::Send(const char *buf, int32_t nSize)
92 {
93 SHARING_LOGD("trace.");
94 std::unique_lock<std::shared_mutex> lk(mutex_);
95 if (socket_ != nullptr) {
96 int32_t fd = socket_->GetLocalFd();
97 SHARING_LOGD("send fd: %{public}d.", fd);
98 if (::write(fd, buf, nSize) != -1) {
99 return true;
100 } else {
101 if (socket_) {
102 MEDIA_LOGE("send [%{public}s:%{public}d]Failed, %{public}s.", socket_->GetPeerIp().c_str(),
103 (int32_t)socket_->GetPeerPort(), strerror(errno));
104 }
105
106 return false;
107 }
108 } else {
109 return false;
110 }
111 }
112
Send(const std::string & msg)113 bool UdpClient::Send(const std::string &msg)
114 {
115 SHARING_LOGD("trace.");
116 return Send(msg.c_str(), msg.size());
117 }
118
GetSocketInfo()119 SocketInfo::Ptr UdpClient::GetSocketInfo()
120 {
121 SHARING_LOGD("trace.");
122 return socket_;
123 }
124
OnClientReadable(int32_t fd)125 void UdpClient::OnClientReadable(int32_t fd)
126 {
127 MEDIA_LOGD("fd: %{public}d, thread_id: %{public}llu.", fd, GetThreadId());
128 int32_t retCode = 0;
129 do {
130 DataBuffer::Ptr buf = std::make_shared<DataBuffer>(DEAFULT_READ_BUFFER_SIZE);
131 int32_t retCode = read(fd, buf->Data(), DEAFULT_READ_BUFFER_SIZE);
132 MEDIA_LOGD("recvSocket len: %{public}d.", retCode);
133 if (retCode > 0) {
134 buf->UpdateSize(retCode);
135 auto callback = callback_.lock();
136 if (callback) {
137 callback->OnClientReadData(fd, std::move(buf));
138 }
139 } else if (retCode == 0) {
140 SHARING_LOGE("recvSocket failed, error:%{public}s!", strerror(errno));
141 Disconnect();
142 }
143 } while (retCode > 0);
144 }
145 } // namespace Sharing
146 } // namespace OHOS
147