1 /*
2  * Copyright (c) 2021-2024 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 #ifndef VTP_STREAM_SOCKET_H
17 #define VTP_STREAM_SOCKET_H
18 
19 #include <condition_variable>
20 #include <cstdint>
21 #include <map>
22 #include <memory>
23 #include <string>
24 #include <utility>
25 
26 #include "fillpinc.h"
27 
28 #include "common_inner.h"
29 #include "i_stream.h"
30 #include "i_stream_socket.h"
31 #include "stream_common.h"
32 #include "vtp_instance.h"
33 #include "vtp_stream_opt.h"
34 
35 namespace Communication {
36 namespace SoftBus {
37 struct ConnectStatus {
38     enum {
39         UNCONNECTED,
40         CONNECTED,
41         CLOSED,
42     };
43 
44     int status = UNCONNECTED;
45 };
46 
47 class VtpStreamSocket : public std::enable_shared_from_this<VtpStreamSocket>, public IStreamSocket {
48 public:
49     static constexpr int FILLP_VTP_SEND_CACHE_SIZE = 500;
50     static constexpr int FILLP_VTP_RECV_CACHE_SIZE = 500;
51     static constexpr int FILLP_KEEP_ALIVE_TIME = 300000;
52 
53     VtpStreamSocket();
54     ~VtpStreamSocket() override;
55     std::shared_ptr<VtpStreamSocket> GetSelf();
56 
57     bool CreateClient(IpAndPort &local, int32_t streamType, std::pair<uint8_t *, uint32_t> sessionKey) override;
58     bool CreateClient(IpAndPort &local, const IpAndPort &remote, int32_t streamType,
59         std::pair<uint8_t*, uint32_t> sessionKey) override;
60 
61     bool CreateServer(IpAndPort &local, int32_t streamType, std::pair<uint8_t *, uint32_t> sessionKey) override;
62 
63     void DestroyStreamSocket() override;
64 
65     bool Connect(const IpAndPort &remote) override;
66     bool Send(std::unique_ptr<IStream> stream) override;
67 
68     bool SetOption(int32_t type, const StreamAttr &value) override;
69     int32_t SetMultiLayer(const void *para) override;
70     StreamAttr GetOption(int32_t type) const override;
71 
72     bool SetStreamListener(std::shared_ptr<IStreamSocketListener> receiver) override;
73 
74     static bool InitVtpInstance(const std::string &pkgName);
75     static void DestroyVtpInstance(const std::string &pkgName);
76 
77     ssize_t GetEncryptOverhead() const;
78 
79     ssize_t Encrypt(const void *in, ssize_t inLen, void *out, ssize_t outLen) const;
80 
81     ssize_t Decrypt(const void *in, ssize_t inLen, void *out, ssize_t outLen) const;
82 
83 private:
84     using MySetFunc = bool (VtpStreamSocket::*)(int32_t, const StreamAttr &);
85     using MyGetFunc = StreamAttr (VtpStreamSocket::*)(int32_t) const;
86     struct OptionFunc {
87         ValueType valueType;
88         MySetFunc set;
89         MyGetFunc get;
90     };
91 
92     const std::map<int32_t, FillpConfigAppListEnum> FILLP_TYPE_MAP {
93         { SEND_CACHE, FT_CONF_SEND_CACHE },           { RECV_CACHE, FT_CONF_RECV_CACHE },
94         { SEND_BUF_SIZE, FT_CONF_SEND_BUFFER_SIZE },  { RECV_BUF_SIZE, FT_CONF_RECV_BUFFER_SIZE },
95         { PACKET_SIZE, FT_CONF_PACKET_SIZE },         { KEEP_ALIVE_TIMEOUT, FT_CONF_TIMER_KEEP_ALIVE },
96         { MAX_VTP_SOCKET_NUM, FT_CONF_MAX_SOCK_NUM }, { MAX_VTP_CONNECT_NUM, FT_CONF_MAX_CONNECTION_NUM },
97         { REDUNANCY_SWITCH, FT_CONF_USE_FEC },        { REDUNANCY_LEVEL, FT_CONF_FEC_REDUNDANCY_LEVEL },
98     };
99 
100     const std::map<int32_t, FillpConfigAppListEnum> INNER_FILLP_TYPE_MAP {
101         { NACK_DELAY, FT_CONF_ENABLE_NACK_DELAY },
102         { NACK_DELAY_TIMEOUT, FT_CONF_NACK_DELAY_TIMEOUT },
103         { PACK_INTERVAL_ENLARGE, FT_CONF_ENLARGE_PACK_INTERVAL },
104         { PKT_STATISTICS, FT_CONF_APP_FC_STATISTICS },
105         { PKT_LOSS, FT_CONF_APP_FC_RECV_PKT_LOSS },
106     };
107     bool EncryptStreamPacket(std::unique_ptr<IStream> stream, std::unique_ptr<char[]> &data, ssize_t &len);
108     bool ProcessCommonDataStream(std::unique_ptr<char[]> &dataBuffer, int32_t &dataLength,
109         std::unique_ptr<char[]> &extBuffer, int32_t &extLen, StreamFrameInfo &info);
110     void InsertElementToFuncMap(int32_t type, ValueType valueType, MySetFunc set, MyGetFunc get);
111     int32_t CreateAndBindSocket(IpAndPort &local, bool isServer) override;
112     bool Accept() override;
113 
114     int32_t EpollTimeout(int32_t fd, int32_t timeout) override;
115     int32_t SetSocketEpollMode(int32_t fd) override;
116 
117     void InsertBufferLength(int32_t num, int32_t length, uint8_t *output) const;
118     std::unique_ptr<IStream> MakeStreamData(StreamData &data, const StreamFrameInfo &info) const;
119     int32_t RecvStreamLen();
120     void DoStreamRecv();
121     std::unique_ptr<char[]> RecvStream(int32_t dataLength) override;
122 
123     void SetDefaultConfig(int32_t fd);
124     bool SetIpTos(int32_t fd, const StreamAttr &tos);
125     StreamAttr GetIpTos(int32_t type = -1) const;
126     StreamAttr GetStreamSocketFd(int32_t type = -1) const;
127     StreamAttr GetListenSocketFd(int32_t type = -1) const;
128     bool SetSocketBoundInner(int32_t fd, std::string ip = "") const;
129     bool SetSocketBindToDevices(int32_t type, const StreamAttr &ip);
130     bool SetVtpStackConfigDelayed(int32_t type, const StreamAttr &value);
131     bool SetVtpStackConfig(int32_t type, const StreamAttr &value);
132     StreamAttr GetVtpStackConfig(int32_t type) const;
133     bool SetNonBlockMode(int32_t fd, const StreamAttr &value);
134     StreamAttr GetNonBlockMode(int32_t fd) const;
135     StreamAttr GetIp(int32_t type) const;
136     StreamAttr GetPort(int32_t type) const;
137     bool SetStreamType(int32_t type, const StreamAttr &value);
138     StreamAttr GetStreamType(int32_t type) const;
GetIpType(int32_t type)139     StreamAttr GetIpType(int32_t type) const
140     {
141         if (type != static_cast<int32_t>(IP_TYPE)) {
142             return std::move(StreamAttr());
143         }
144         return std::move(StreamAttr(std::string("V4")));
145     }
GetRemoteScopeId(int32_t type)146     StreamAttr GetRemoteScopeId(int32_t type) const
147     {
148         if (type != static_cast<int32_t>(REMOTE_SCOPE_ID)) {
149             return std::move(StreamAttr());
150         }
151         return std::move(StreamAttr(0));
152     }
153 
IsServer(int32_t type)154     StreamAttr IsServer(int32_t type) const
155     {
156         if (type != static_cast<int32_t>(IS_SERVER)) {
157             return std::move(StreamAttr());
158         }
159         return std::move(StreamAttr(listenFd_ != -1));
160     }
161 
162     bool SetStreamScene(int32_t type, const StreamAttr &value);
163     bool SetStreamHeaderSize(int32_t type, const StreamAttr &value);
164 
165     void NotifyStreamListener();
166 
167     bool EnableBwEstimationAlgo(int32_t streamFd, bool isServer) const;
168 
169     bool EnableJitterDetectionAlgo(int32_t streamFd) const;
170 
171     bool EnableDirectlySend(int32_t streamFd) const;
172 
173     bool EnableSemiReliable(int32_t streamFd) const;
174 
175     void RegisterMetricCallback(bool isServer); /* register the metric callback function */
176 
177     static void AddStreamSocketLock(int32_t fd, std::mutex &streamsocketlock);
178 
179     static void AddStreamSocketListener(int32_t fd, std::shared_ptr<VtpStreamSocket> streamreceiver);
180 
181     static void RemoveStreamSocketLock(int32_t fd);
182 
183     static void RemoveStreamSocketListener(int32_t fd);
184 
185     static int32_t HandleFillpFrameStats(int32_t fd, const FtEventCbkInfo *info);
186 
187     static int32_t HandleRipplePolicy(int32_t fd, const FtEventCbkInfo *info);
188 
189     static int32_t HandleFillpFrameEvt(int32_t fd, const FtEventCbkInfo *info);
190 
191     int32_t HandleFillpFrameEvtInner(int32_t fd, const FtEventCbkInfo *info);
192 
193     static int32_t FillpStatistics(int32_t fd, const FtEventCbkInfo *info);
194 
195     void FillpAppStatistics();
196 
197     static void FillSupportDet(int32_t fd, const FtEventCbkInfo *info, QosTv* metricList);
198 
199     void CreateServerProcessThread();
200 
201     void CreateClientProcessThread();
202 
203     static std::map<int32_t, std::mutex &> g_streamSocketLockMap;
204     static std::mutex streamSocketLockMapLock_;
205     static std::map<int32_t, std::shared_ptr<VtpStreamSocket>> g_streamSocketMap;
206     static std::mutex streamSocketMapLock_;
207 
208     std::map<int32_t, OptionFunc> optFuncMap_ {};
209     static std::shared_ptr<VtpInstance> vtpInstance_;
210     std::condition_variable configCv_;
211     std::mutex streamSocketLock_;
212     int32_t scene_ = UNKNOWN_SCENE;
213     int32_t streamHdrSize_ = 0;
214     bool isDestroyed_ = false;
215     OnFrameEvt onStreamEvtCb_;
216 };
217 } // namespace SoftBus
218 } // namespace Communication
219 
220 #endif
221