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