mii/libdatachannel
1
0
Fork 0
mirror of https://github.com/mii443/libdatachannel.git synced 2025-08-22 15:15:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Moved implementation away with pimpl aka cheshire cat pattern

Browse Source
This commit is contained in:
Paul-Louis Ageneau
2021-02-28 11:16:51 +01:00
parent 65dba2c299
commit dde79d78d4
53 changed files with 2870 additions and 2287 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
CMakeLists.txt
View File

@ -46,26 +46,17 @@ endif()
set(LIBDATACHANNEL_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/src/candidate.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/certificate.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/channel.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/configuration.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/datachannel.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/description.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/dtlssrtptransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/dtlstransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/icetransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/init.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/log.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/message.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/peerconnection.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/logcounter.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/rtcpreceivingsession.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/sctptransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/threadpool.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/tls.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/track.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/processor.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/capi.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/websocket.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/rtppacketizationconfig.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/rtcpsrreporter.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/rtppacketizer.cpp
@ -78,36 +69,7 @@ set(LIBDATACHANNEL_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/src/mediahandlerelement.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/mediahandlerrootelement.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/rtcpnackresponder.cpp
)
set(LIBDATACHANNEL_PRIVATE_HEADERS
${CMAKE_CURRENT_SOURCE_DIR}/src/certificate.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/dtlssrtptransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/dtlstransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/icetransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/logcounter.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/sctptransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/threadpool.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/tls.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/processor.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/transport.hpp
)
set(LIBDATACHANNEL_WEBSOCKET_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/src/base64.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/tcptransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/tlstransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/verifiedtlstransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/websocket.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/wstransport.cpp
)
set(LIBDATACHANNEL_WEBSOCKET_PRIVATE_HEADERS
${CMAKE_CURRENT_SOURCE_DIR}/src/base64.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/tcptransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/tlstransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/verifiedtlstransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/wstransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/capi.cpp
)
set(LIBDATACHANNEL_HEADERS
@ -145,6 +107,50 @@ set(LIBDATACHANNEL_HEADERS
${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtcpnackresponder.hpp
)
set(LIBDATACHANNEL_IMPL_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/certificate.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/channel.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/datachannel.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/dtlssrtptransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/dtlstransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/icetransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/peerconnection.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/logcounter.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/sctptransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/threadpool.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/tls.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/track.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/processor.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/base64.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/tcptransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/tlstransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/verifiedtlstransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/websocket.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/wstransport.cpp
)
set(LIBDATACHANNEL_IMPL_HEADERS
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/certificate.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/channel.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/datachannel.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/dtlssrtptransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/dtlstransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/icetransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/peerconnection.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/logcounter.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/sctptransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/threadpool.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/tls.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/track.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/processor.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/base64.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/tcptransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/tlstransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/verifiedtlstransport.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/websocket.hpp
${CMAKE_CURRENT_SOURCE_DIR}/src/impl/wstransport.hpp
)
set(TESTS_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/test/main.cpp
${CMAKE_CURRENT_SOURCE_DIR}/test/connectivity.cpp
@ -163,7 +169,8 @@ set(TESTS_UWP_RESOURCES
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/tests/SmallLogo44x44.png
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/tests/SplashScreen.png
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/tests/StoreLogo.png
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/tests/Windows_TemporaryKey.pfx)
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/tests/Windows_TemporaryKey.pfx
)
set(BENCHMARK_UWP_RESOURCES
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/benchmark/Logo.png
@ -172,7 +179,8 @@ set(BENCHMARK_UWP_RESOURCES
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/benchmark/SmallLogo44x44.png
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/benchmark/SplashScreen.png
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/benchmark/StoreLogo.png
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/benchmark/Windows_TemporaryKey.pfx)
${CMAKE_CURRENT_SOURCE_DIR}/test/uwp/benchmark/Windows_TemporaryKey.pfx
)
set(CMAKE_THREAD_PREFER_PTHREAD TRUE)
set(THREADS_PREFER_PTHREAD_FLAG TRUE)
@ -192,33 +200,16 @@ if (CMAKE_CXX_COMPILER_ID MATCHES "GNU")
endif()
add_library(Usrsctp::Usrsctp ALIAS usrsctp)
if (NO_WEBSOCKET)
add_library(datachannel SHARED
${LIBDATACHANNEL_SOURCES}
${LIBDATACHANNEL_PRIVATE_HEADERS}
${LIBDATACHANNEL_HEADERS})
add_library(datachannel-static STATIC EXCLUDE_FROM_ALL
${LIBDATACHANNEL_SOURCES}
${LIBDATACHANNEL_PRIVATE_HEADERS}
${LIBDATACHANNEL_HEADERS})
target_compile_definitions(datachannel PUBLIC RTC_ENABLE_WEBSOCKET=0)
target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_WEBSOCKET=0)
else()
add_library(datachannel SHARED
${LIBDATACHANNEL_SOURCES}
${LIBDATACHANNEL_PRIVATE_HEADERS}
${LIBDATACHANNEL_WEBSOCKET_SOURCES}
${LIBDATACHANNEL_WEBSOCKET_PRIVATE_HEADERS}
${LIBDATACHANNEL_HEADERS})
add_library(datachannel-static STATIC EXCLUDE_FROM_ALL
${LIBDATACHANNEL_SOURCES}
${LIBDATACHANNEL_PRIVATE_HEADERS}
${LIBDATACHANNEL_WEBSOCKET_SOURCES}
${LIBDATACHANNEL_WEBSOCKET_PRIVATE_HEADERS}
${LIBDATACHANNEL_HEADERS})
target_compile_definitions(datachannel PUBLIC RTC_ENABLE_WEBSOCKET=1)
target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_WEBSOCKET=1)
endif()
add_library(datachannel SHARED
${LIBDATACHANNEL_SOURCES}
${LIBDATACHANNEL_HEADERS}
${LIBDATACHANNEL_IMPL_SOURCES}
${LIBDATACHANNEL_IMPL_HEADERS})
add_library(datachannel-static STATIC EXCLUDE_FROM_ALL
${LIBDATACHANNEL_SOURCES}
${LIBDATACHANNEL_HEADERS}
${LIBDATACHANNEL_IMPL_SOURCES}
${LIBDATACHANNEL_IMPL_HEADERS})
set_target_properties(datachannel PROPERTIES
VERSION ${PROJECT_VERSION}
@ -244,6 +235,14 @@ if(WIN32)
target_link_libraries(datachannel-static PUBLIC ws2_32) # winsock2
endif()
if (NO_WEBSOCKET)
target_compile_definitions(datachannel PUBLIC RTC_ENABLE_WEBSOCKET=0)
target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_WEBSOCKET=0)
else()
target_compile_definitions(datachannel PUBLIC RTC_ENABLE_WEBSOCKET=1)
target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_WEBSOCKET=1)
endif()
if(NO_MEDIA)
target_compile_definitions(datachannel PUBLIC RTC_ENABLE_MEDIA=0)
target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_MEDIA=0)

36
include/rtc/channel.hpp
View File

@ -1,5 +1,5 @@
/**
* Copyright (c) 2019 Paul-Louis Ageneau
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@ -28,10 +28,13 @@
namespace rtc {
class RTC_CPP_EXPORT Channel {
namespace impl {
struct Channel;
}
class RTC_CPP_EXPORT Channel : private CheshireCat<impl::Channel> {
public:
Channel() = default;
virtual ~Channel() = default;
virtual ~Channel();
virtual void close() = 0;
virtual bool send(message_variant data) = 0; // returns false if buffered
@ -54,30 +57,13 @@ public:
void setBufferedAmountLowThreshold(size_t amount);
// Extended API
virtual std::optional<message_variant> receive() = 0; // only if onMessage unset
virtual std::optional<message_variant> peek() = 0; // only if onMessage unset
virtual size_t availableAmount() const; // total size available to receive
std::optional<message_variant> receive(); // only if onMessage unset
std::optional<message_variant> peek(); // only if onMessage unset
size_t availableAmount() const; // total size available to receive
void onAvailable(std::function<void()> callback);
protected:
virtual void triggerOpen();
virtual void triggerClosed();
virtual void triggerError(string error);
virtual void triggerAvailable(size_t count);
virtual void triggerBufferedAmount(size_t amount);
void resetCallbacks();
private:
synchronized_callback<> mOpenCallback;
synchronized_callback<> mClosedCallback;
synchronized_callback<string> mErrorCallback;
synchronized_callback<message_variant> mMessageCallback;
synchronized_callback<> mAvailableCallback;
synchronized_callback<> mBufferedAmountLowCallback;
std::atomic<size_t> mBufferedAmount = 0;
std::atomic<size_t> mBufferedAmountLowThreshold = 0;
Channel(impl_ptr<impl::Channel> impl);
};
} // namespace rtc

77
include/rtc/datachannel.hpp
View File

@ -28,20 +28,23 @@
#include <atomic>
#include <chrono>
#include <functional>
#include <shared_mutex>
#include <type_traits>
#include <variant>
#include <shared_mutex>
namespace rtc {
class SctpTransport;
class PeerConnection;
namespace impl {
class RTC_CPP_EXPORT DataChannel : public std::enable_shared_from_this<DataChannel>,
public Channel {
struct DataChannel;
struct PeerConnection;
} // namespace impl
class RTC_CPP_EXPORT DataChannel final : private CheshireCat<impl::DataChannel>, public Channel {
public:
DataChannel(std::weak_ptr<PeerConnection> pc, uint16_t stream, string label, string protocol,
Reliability reliability);
DataChannel(impl_ptr<impl::DataChannel> impl);
virtual ~DataChannel();
uint16_t stream() const;
@ -50,60 +53,18 @@ public:
string protocol() const;
Reliability reliability() const;
bool isOpen(void) const override;
bool isClosed(void) const override;
size_t maxMessageSize() const override;
void close(void) override;
bool send(message_variant data) override;
bool send(const byte *data, size_t size) override;
template <typename Buffer> bool sendBuffer(const Buffer &buf);
template <typename Iterator> bool sendBuffer(Iterator first, Iterator last);
bool isOpen(void) const override;
bool isClosed(void) const override;
size_t maxMessageSize() const override;
// Extended API
size_t availableAmount() const override;
std::optional<message_variant> receive() override;
std::optional<message_variant> peek() override;
protected:
virtual void open(std::shared_ptr<SctpTransport> transport);
virtual void processOpenMessage(message_ptr message);
void remoteClose();
bool outgoing(message_ptr message);
void incoming(message_ptr message);
const std::weak_ptr<PeerConnection> mPeerConnection;
std::weak_ptr<SctpTransport> mSctpTransport;
uint16_t mStream;
string mLabel;
string mProtocol;
std::shared_ptr<Reliability> mReliability;
mutable std::shared_mutex mMutex;
std::atomic<bool> mIsOpen = false;
std::atomic<bool> mIsClosed = false;
private:
Queue<message_ptr> mRecvQueue;
friend class PeerConnection;
};
class RTC_CPP_EXPORT NegotiatedDataChannel final : public DataChannel {
public:
NegotiatedDataChannel(std::weak_ptr<PeerConnection> pc, uint16_t stream, string label,
string protocol, Reliability reliability);
NegotiatedDataChannel(std::weak_ptr<PeerConnection> pc, std::weak_ptr<SctpTransport> transport,
uint16_t stream);
~NegotiatedDataChannel();
private:
void open(std::shared_ptr<SctpTransport> transport) override;
void processOpenMessage(message_ptr message) override;
friend class PeerConnection;
using CheshireCat<impl::DataChannel>::impl;
};
template <typename Buffer> std::pair<const byte *, size_t> to_bytes(const Buffer &buf) {
@ -115,9 +76,7 @@ template <typename Buffer> std::pair<const byte *, size_t> to_bytes(const Buffer
template <typename Buffer> bool DataChannel::sendBuffer(const Buffer &buf) {
auto [bytes, size] = to_bytes(buf);
auto message = std::make_shared<Message>(size);
std::copy(bytes, bytes + size, message->data());
return outgoing(message);
return send(bytes, size);
}
template <typename Iterator> bool DataChannel::sendBuffer(Iterator first, Iterator last) {
@ -125,13 +84,13 @@ template <typename Iterator> bool DataChannel::sendBuffer(Iterator first, Iterat
for (Iterator it = first; it != last; ++it)
size += it->size();
auto message = std::make_shared<Message>(size);
auto pos = message->begin();
binary buffer(size);
byte *pos = buffer.data();
for (Iterator it = first; it != last; ++it) {
auto [bytes, len] = to_bytes(*it);
pos = std::copy(bytes, bytes + len, pos);
}
return outgoing(message);
return send(std::move(buffer));
}
} // namespace rtc

27
include/rtc/include.hpp
View File

@ -52,14 +52,17 @@
namespace rtc {
using std::nullopt;
using std::byte;
using std::string;
using std::string_view;
using std::shared_ptr;
using std::weak_ptr;
using std::unique_ptr;
using binary = std::vector<byte>;
using binary_ptr = std::shared_ptr<binary>;
using std::nullopt;
using std::size_t;
using std::uint16_t;
using std::uint32_t;
@ -111,6 +114,7 @@ private:
std::function<void()> function;
};
// callback with built-in synchronization
template <typename... Args> class synchronized_callback {
public:
synchronized_callback() = default;
@ -157,6 +161,25 @@ private:
std::function<void(Args...)> callback;
mutable std::recursive_mutex mutex;
};
// pimpl base class
template<typename T> using impl_ptr = std::shared_ptr<T>;
template <typename T> class CheshireCat {
public:
CheshireCat(impl_ptr<T> impl) : mImpl(std::move(impl)) {}
template <typename... Args>
CheshireCat(Args... args) : mImpl(std::make_shared<T>(std::move(args)...)) {}
virtual ~CheshireCat() = default;
protected:
impl_ptr<T> impl() { return mImpl; }
impl_ptr<const T> impl() const { return mImpl; }
private:
impl_ptr<T> mImpl;
};
} // namespace rtc
#endif

96
include/rtc/peerconnection.hpp
View File

@ -30,26 +30,16 @@
#include "rtc.hpp"
#include "track.hpp"
#include <atomic>
#include <chrono>
#include <functional>
#include <future>
#include <list>
#include <mutex>
#include <shared_mutex>
#include <thread>
#include <unordered_map>
namespace rtc {
class Certificate;
class Processor;
class IceTransport;
class DtlsTransport;
class SctpTransport;
namespace impl {
using certificate_ptr = std::shared_ptr<Certificate>;
using future_certificate_ptr = std::shared_future<certificate_ptr>;
struct PeerConnection;
}
struct RTC_CPP_EXPORT DataChannelInit {
Reliability reliability = {};
@ -58,7 +48,7 @@ struct RTC_CPP_EXPORT DataChannelInit {
string protocol = "";
};
class RTC_CPP_EXPORT PeerConnection final : public std::enable_shared_from_this<PeerConnection> {
class RTC_CPP_EXPORT PeerConnection final : CheshireCat<impl::PeerConnection> {
public:
enum class State : int {
New = RTC_NEW,
@ -84,7 +74,7 @@ public:
} rtcSignalingState;
PeerConnection();
PeerConnection(const Configuration &config);
PeerConnection(Configuration config);
~PeerConnection();
void close();
@ -128,80 +118,6 @@ public:
// Track media support requires compiling with libSRTP
std::shared_ptr<Track> addTrack(Description::Media description);
void onTrack(std::function<void(std::shared_ptr<Track> track)> callback);
private:
std::shared_ptr<IceTransport> initIceTransport();
std::shared_ptr<DtlsTransport> initDtlsTransport();
std::shared_ptr<SctpTransport> initSctpTransport();
void closeTransports();
void endLocalCandidates();
bool checkFingerprint(const std::string &fingerprint) const;
void forwardMessage(message_ptr message);
void forwardMedia(message_ptr message);
void forwardBufferedAmount(uint16_t stream, size_t amount);
std::optional<std::string> getMidFromSsrc(uint32_t ssrc);
std::shared_ptr<DataChannel> emplaceDataChannel(Description::Role role, string label,
DataChannelInit init);
std::shared_ptr<DataChannel> findDataChannel(uint16_t stream);
void iterateDataChannels(std::function<void(std::shared_ptr<DataChannel> channel)> func);
void openDataChannels();
void closeDataChannels();
void remoteCloseDataChannels();
void incomingTrack(Description::Media description);
void openTracks();
void validateRemoteDescription(const Description &description);
void processLocalDescription(Description description);
void processLocalCandidate(Candidate candidate);
void processRemoteDescription(Description description);
void processRemoteCandidate(Candidate candidate);
string localBundleMid() const;
void triggerDataChannel(std::weak_ptr<DataChannel> weakDataChannel);
void triggerTrack(std::shared_ptr<Track> track);
bool changeState(State state);
bool changeGatheringState(GatheringState state);
bool changeSignalingState(SignalingState state);
void resetCallbacks();
void outgoingMedia(message_ptr message);
const init_token mInitToken = Init::Token();
const Configuration mConfig;
const future_certificate_ptr mCertificate;
const std::unique_ptr<Processor> mProcessor;
std::optional<Description> mLocalDescription, mRemoteDescription;
std::optional<Description> mCurrentLocalDescription;
mutable std::mutex mLocalDescriptionMutex, mRemoteDescriptionMutex;
std::shared_ptr<IceTransport> mIceTransport;
std::shared_ptr<DtlsTransport> mDtlsTransport;
std::shared_ptr<SctpTransport> mSctpTransport;
std::unordered_map<uint16_t, std::weak_ptr<DataChannel>> mDataChannels; // by stream ID
std::unordered_map<string, std::weak_ptr<Track>> mTracks; // by mid
std::vector<std::weak_ptr<Track>> mTrackLines; // by SDP order
std::shared_mutex mDataChannelsMutex, mTracksMutex;
std::unordered_map<uint32_t, string> mMidFromSsrc; // cache
std::atomic<State> mState;
std::atomic<GatheringState> mGatheringState;
std::atomic<SignalingState> mSignalingState;
std::atomic<bool> mNegotiationNeeded;
synchronized_callback<std::shared_ptr<DataChannel>> mDataChannelCallback;
synchronized_callback<Description> mLocalDescriptionCallback;
synchronized_callback<Candidate> mLocalCandidateCallback;
synchronized_callback<State> mStateChangeCallback;
synchronized_callback<GatheringState> mGatheringStateChangeCallback;
synchronized_callback<SignalingState> mSignalingStateChangeCallback;
synchronized_callback<std::shared_ptr<Track>> mTrackCallback;
};
} // namespace rtc

42
include/rtc/track.hpp
View File

@ -22,28 +22,30 @@
#include "channel.hpp"
#include "description.hpp"
#include "include.hpp"
#include "mediahandler.hpp"
#include "message.hpp"
#include "queue.hpp"
#include "mediahandler.hpp"
#include <atomic>
#include <variant>
#include <shared_mutex>
#include <variant>
namespace rtc {
#if RTC_ENABLE_MEDIA
class DtlsSrtpTransport;
#endif
namespace impl {
class RTC_CPP_EXPORT Track final : public std::enable_shared_from_this<Track>, public Channel {
class Track;
} // namespace impl
class RTC_CPP_EXPORT Track final : private CheshireCat<impl::Track>, public Channel {
public:
Track(Description::Media description);
Track(impl_ptr<impl::Track> impl);
~Track() = default;
string mid() const;
Description::Media description() const;
Description::Direction direction() const;
Description::Media description() const;
void setDescription(Description::Media description);
@ -55,11 +57,6 @@ public:
bool isClosed(void) const override;
size_t maxMessageSize() const override;
// Extended API
size_t availableAmount() const override;
std::optional<message_variant> receive() override;
std::optional<message_variant> peek() override;
bool requestKeyframe();
// RTCP handler
@ -67,24 +64,7 @@ public:
std::shared_ptr<MediaHandler> getRtcpHandler();
private:
#if RTC_ENABLE_MEDIA
void open(std::shared_ptr<DtlsSrtpTransport> transport);
std::weak_ptr<DtlsSrtpTransport> mDtlsSrtpTransport;
#endif
void incoming(message_ptr message);
bool outgoing(message_ptr message);
Description::Media mMediaDescription;
std::shared_ptr<MediaHandler> mRtcpHandler;
mutable std::shared_mutex mMutex;
std::atomic<bool> mIsClosed = false;
Queue<message_ptr> mRecvQueue;
friend class PeerConnection;
using CheshireCat<impl::Track>::impl;
};
} // namespace rtc

58
include/rtc/websocket.hpp
View File

@ -1,5 +1,5 @@
/**
* Copyright (c) 2020 Paul-Louis Ageneau
* Copyright (c) 2020-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@ -27,19 +27,15 @@
#include "message.hpp"
#include "queue.hpp"
#include <atomic>
#include <optional>
#include <thread>
#include <variant>
namespace rtc {
class TcpTransport;
class TlsTransport;
class WsTransport;
namespace impl {
class RTC_CPP_EXPORT WebSocket final : public Channel,
public std::enable_shared_from_this<WebSocket> {
struct WebSocket;
}
class RTC_CPP_EXPORT WebSocket final : private CheshireCat<impl::WebSocket>, public Channel {
public:
enum class State : int {
Connecting = 0,
@ -53,49 +49,25 @@ public:
std::vector<string> protocols;
};
WebSocket(std::optional<Configuration> config = nullopt);
WebSocket();
WebSocket(Configuration config);
~WebSocket();
State readyState() const;
bool isOpen() const override;
bool isClosed() const override;
size_t maxMessageSize() const override;
void open(const string &url);
void close() override;
bool send(const message_variant data) override;
bool send(const byte *data, size_t size) override;
bool isOpen() const override;
bool isClosed() const override;
size_t maxMessageSize() const override;
// Extended API
std::optional<message_variant> receive() override;
std::optional<message_variant> peek() override;
size_t availableAmount() const override; // total size available to receive
private:
bool changeState(State state);
void remoteClose();
bool outgoing(message_ptr message);
void incoming(message_ptr message);
std::shared_ptr<TcpTransport> initTcpTransport();
std::shared_ptr<TlsTransport> initTlsTransport();
std::shared_ptr<WsTransport> initWsTransport();
void closeTransports();
init_token mInitToken = Init::Token();
std::shared_ptr<TcpTransport> mTcpTransport;
std::shared_ptr<TlsTransport> mTlsTransport;
std::shared_ptr<WsTransport> mWsTransport;
std::recursive_mutex mInitMutex;
const Configuration mConfig;
string mScheme, mHost, mHostname, mService, mPath;
std::atomic<State> mState = State::Closed;
Queue<message_ptr> mRecvQueue;
using CheshireCat<impl::WebSocket>::impl;
};
} // namespace rtc
#endif

72
src/channel.cpp
View File

@ -1,5 +1,5 @@
/**
* Copyright (c) 2019 Paul-Louis Ageneau
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@ -18,26 +18,34 @@
#include "channel.hpp"
#include "impl/channel.hpp"
namespace rtc {
Channel::~Channel() {
impl()->resetCallbacks();
}
Channel::Channel(impl_ptr<impl::Channel> impl) : CheshireCat<impl::Channel>(std::move(impl)) {}
size_t Channel::maxMessageSize() const { return DEFAULT_MAX_MESSAGE_SIZE; }
size_t Channel::bufferedAmount() const { return mBufferedAmount; }
size_t Channel::bufferedAmount() const { return impl()->bufferedAmount; }
size_t Channel::availableAmount() const { return 0; }
void Channel::onOpen(std::function<void()> callback) { impl()->openCallback = callback; }
void Channel::onOpen(std::function<void()> callback) { mOpenCallback = callback; }
void Channel::onClosed(std::function<void()> callback) { impl()->closedCallback = callback; }
void Channel::onClosed(std::function<void()> callback) { mClosedCallback = callback; }
void Channel::onError(std::function<void(string error)> callback) { mErrorCallback = callback; }
void Channel::onError(std::function<void(string error)> callback) {
impl()->errorCallback = callback;
}
void Channel::onMessage(std::function<void(message_variant data)> callback) {
mMessageCallback = callback;
impl()->messageCallback = callback;
// Pass pending messages
while (auto message = receive())
mMessageCallback(*message);
impl()->messageCallback(*message);
}
void Channel::onMessage(std::function<void(binary data)> binaryCallback,
@ -48,45 +56,25 @@ void Channel::onMessage(std::function<void(binary data)> binaryCallback,
}
void Channel::onBufferedAmountLow(std::function<void()> callback) {
mBufferedAmountLowCallback = callback;
impl()->bufferedAmountLowCallback = callback;
}
void Channel::setBufferedAmountLowThreshold(size_t amount) { mBufferedAmountLowThreshold = amount; }
void Channel::onAvailable(std::function<void()> callback) { mAvailableCallback = callback; }
void Channel::triggerOpen() { mOpenCallback(); }
void Channel::triggerClosed() { mClosedCallback(); }
void Channel::triggerError(string error) { mErrorCallback(error); }
void Channel::triggerAvailable(size_t count) {
if (count == 1)
mAvailableCallback();
while (mMessageCallback && count--) {
auto message = receive();
if (!message)
break;
mMessageCallback(*message);
}
void Channel::setBufferedAmountLowThreshold(size_t amount) {
impl()->bufferedAmountLowThreshold = amount;
}
void Channel::triggerBufferedAmount(size_t amount) {
size_t previous = mBufferedAmount.exchange(amount);
size_t threshold = mBufferedAmountLowThreshold.load();
if (previous > threshold && amount <= threshold)
mBufferedAmountLowCallback();
std::optional<message_variant> Channel::receive() {
return impl()->receive();
}
void Channel::resetCallbacks() {
mOpenCallback = nullptr;
mClosedCallback = nullptr;
mErrorCallback = nullptr;
mMessageCallback = nullptr;
mAvailableCallback = nullptr;
mBufferedAmountLowCallback = nullptr;
std::optional<message_variant> Channel::peek() {
return impl()->peek();
}
size_t Channel::availableAmount() const {
return impl()->availableAmount();
}
void Channel::onAvailable(std::function<void()> callback) { impl()->availableCallback = callback; }
} // namespace rtc

349
src/datachannel.cpp
View File

@ -1,5 +1,5 @@
/**
* Copyright (c) 2019 Paul-Louis Ageneau
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@ -18,9 +18,10 @@
#include "datachannel.hpp"
#include "include.hpp"
#include "logcounter.hpp"
#include "peerconnection.hpp"
#include "sctptransport.hpp"
#include "impl/datachannel.hpp"
#include "impl/peerconnection.hpp"
#ifdef _WIN32
#include <winsock2.h>
@ -30,334 +31,36 @@
namespace rtc {
LogCounter COUNTER_USERNEG_OPEN_MESSAGE(
plog::warning, "Number of open messages for a user-negotiated DataChannel received");
using std::shared_ptr;
using std::weak_ptr;
using std::chrono::milliseconds;
// Messages for the DataChannel establishment protocol
// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09
enum MessageType : uint8_t {
MESSAGE_OPEN_REQUEST = 0x00,
MESSAGE_OPEN_RESPONSE = 0x01,
MESSAGE_ACK = 0x02,
MESSAGE_OPEN = 0x03,
MESSAGE_CLOSE = 0x04
};
enum ChannelType : uint8_t {
CHANNEL_RELIABLE = 0x00,
CHANNEL_PARTIAL_RELIABLE_REXMIT = 0x01,
CHANNEL_PARTIAL_RELIABLE_TIMED = 0x02
};
#pragma pack(push, 1)
struct OpenMessage {
uint8_t type = MESSAGE_OPEN;
uint8_t channelType;
uint16_t priority;
uint32_t reliabilityParameter;
uint16_t labelLength;
uint16_t protocolLength;
// The following fields are:
// uint8_t[labelLength] label
// uint8_t[protocolLength] protocol
};
struct AckMessage {
uint8_t type = MESSAGE_ACK;
};
struct CloseMessage {
uint8_t type = MESSAGE_CLOSE;
};
#pragma pack(pop)
DataChannel::DataChannel(weak_ptr<PeerConnection> pc, uint16_t stream, string label,
string protocol, Reliability reliability)
: mPeerConnection(pc), mStream(stream), mLabel(std::move(label)),
mProtocol(std::move(protocol)),
mReliability(std::make_shared<Reliability>(std::move(reliability))),
mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
DataChannel::DataChannel(impl_ptr<impl::DataChannel> impl)
: CheshireCat<impl::DataChannel>(impl),
Channel(std::dynamic_pointer_cast<impl::Channel>(impl)) {}
DataChannel::~DataChannel() { close(); }
uint16_t DataChannel::stream() const { return mStream; }
void DataChannel::close() { return impl()->close(); }
uint16_t DataChannel::id() const { return mStream; }
uint16_t DataChannel::stream() const { return impl()->stream(); }
string DataChannel::label() const {
std::shared_lock lock(mMutex);
return mLabel;
uint16_t DataChannel::id() const { return impl()->stream(); }
string DataChannel::label() const { return impl()->label(); }
string DataChannel::protocol() const { return impl()->protocol(); }
Reliability DataChannel::reliability() const { return impl()->reliability(); }
bool DataChannel::isOpen(void) const { return impl()->isOpen(); }
bool DataChannel::isClosed(void) const { return impl()->isClosed(); }
size_t DataChannel::maxMessageSize() const { return impl()->maxMessageSize(); }
bool DataChannel::send(message_variant data) {
return impl()->outgoing(make_message(std::move(data)));
}
string DataChannel::protocol() const {
std::shared_lock lock(mMutex);
return mProtocol;
}
Reliability DataChannel::reliability() const {
std::shared_lock lock(mMutex);
return *mReliability;
}
void DataChannel::close() {
std::shared_ptr<SctpTransport> transport;
{
std::shared_lock lock(mMutex);
transport = mSctpTransport.lock();
}
mIsClosed = true;
if (mIsOpen.exchange(false) && transport)
transport->closeStream(mStream);
resetCallbacks();
}
void DataChannel::remoteClose() {
if (!mIsClosed.exchange(true))
triggerClosed();
mIsOpen = false;
}
bool DataChannel::send(message_variant data) { return outgoing(make_message(std::move(data))); }
bool DataChannel::send(const byte *data, size_t size) {
return outgoing(std::make_shared<Message>(data, data + size, Message::Binary));
}
std::optional<message_variant> DataChannel::receive() {
while (auto next = mRecvQueue.tryPop()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
auto raw = reinterpret_cast<const uint8_t *>(message->data());
if (!message->empty() && raw[0] == MESSAGE_CLOSE)
remoteClose();
}
return nullopt;
}
std::optional<message_variant> DataChannel::peek() {
while (auto next = mRecvQueue.peek()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
auto raw = reinterpret_cast<const uint8_t *>(message->data());
if (!message->empty() && raw[0] == MESSAGE_CLOSE)
remoteClose();
mRecvQueue.tryPop();
}
return nullopt;
}
bool DataChannel::isOpen(void) const { return mIsOpen; }
bool DataChannel::isClosed(void) const { return mIsClosed; }
size_t DataChannel::maxMessageSize() const {
size_t remoteMax = DEFAULT_MAX_MESSAGE_SIZE;
if (auto pc = mPeerConnection.lock())
if (auto description = pc->remoteDescription())
if (auto *application = description->application())
if (auto maxMessageSize = application->maxMessageSize())
remoteMax = *maxMessageSize > 0 ? *maxMessageSize : LOCAL_MAX_MESSAGE_SIZE;
return std::min(remoteMax, LOCAL_MAX_MESSAGE_SIZE);
}
size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
void DataChannel::open(shared_ptr<SctpTransport> transport) {
{
std::unique_lock lock(mMutex);
mSctpTransport = transport;
}
if (!mIsOpen.exchange(true))
triggerOpen();
}
void DataChannel::processOpenMessage(message_ptr) {
PLOG_DEBUG << "Received an open message for a user-negotiated DataChannel, ignoring";
COUNTER_USERNEG_OPEN_MESSAGE++;
}
bool DataChannel::outgoing(message_ptr message) {
std::shared_ptr<SctpTransport> transport;
{
std::shared_lock lock(mMutex);
transport = mSctpTransport.lock();
if (!transport || mIsClosed)
throw std::runtime_error("DataChannel is closed");
if (message->size() > maxMessageSize())
throw std::runtime_error("Message size exceeds limit");
// Before the ACK has been received on a DataChannel, all messages must be sent ordered
message->reliability = mIsOpen ? mReliability : nullptr;
message->stream = mStream;
}
return transport->send(message);
}
void DataChannel::incoming(message_ptr message) {
if (!message)
return;
switch (message->type) {
case Message::Control: {
if (message->size() == 0)
break; // Ignore
auto raw = reinterpret_cast<const uint8_t *>(message->data());
switch (raw[0]) {
case MESSAGE_OPEN:
processOpenMessage(message);
break;
case MESSAGE_ACK:
if (!mIsOpen.exchange(true)) {
triggerOpen();
}
break;
case MESSAGE_CLOSE:
// The close message will be processed in-order in receive()
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
break;
default:
// Ignore
break;
}
break;
}
case Message::String:
case Message::Binary:
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
break;
default:
// Ignore
break;
}
}
NegotiatedDataChannel::NegotiatedDataChannel(std::weak_ptr<PeerConnection> pc, uint16_t stream,
string label, string protocol, Reliability reliability)
: DataChannel(pc, stream, std::move(label), std::move(protocol), std::move(reliability)) {}
NegotiatedDataChannel::NegotiatedDataChannel(std::weak_ptr<PeerConnection> pc,
std::weak_ptr<SctpTransport> transport,
uint16_t stream)
: DataChannel(pc, stream, "", "", {}) {
mSctpTransport = transport;
}
NegotiatedDataChannel::~NegotiatedDataChannel() {}
void NegotiatedDataChannel::open(shared_ptr<SctpTransport> transport) {
std::unique_lock lock(mMutex);
mSctpTransport = transport;
uint8_t channelType;
uint32_t reliabilityParameter;
switch (mReliability->type) {
case Reliability::Type::Rexmit:
channelType = CHANNEL_PARTIAL_RELIABLE_REXMIT;
reliabilityParameter = uint32_t(std::get<int>(mReliability->rexmit));
break;
case Reliability::Type::Timed:
channelType = CHANNEL_PARTIAL_RELIABLE_TIMED;
reliabilityParameter = uint32_t(std::get<milliseconds>(mReliability->rexmit).count());
break;
default:
channelType = CHANNEL_RELIABLE;
reliabilityParameter = 0;
break;
}
if (mReliability->unordered)
channelType |= 0x80;
const size_t len = sizeof(OpenMessage) + mLabel.size() + mProtocol.size();
binary buffer(len, byte(0));
auto &open = *reinterpret_cast<OpenMessage *>(buffer.data());
open.type = MESSAGE_OPEN;
open.channelType = channelType;
open.priority = htons(0);
open.reliabilityParameter = htonl(reliabilityParameter);
open.labelLength = htons(uint16_t(mLabel.size()));
open.protocolLength = htons(uint16_t(mProtocol.size()));
auto end = reinterpret_cast<char *>(buffer.data() + sizeof(OpenMessage));
std::copy(mLabel.begin(), mLabel.end(), end);
std::copy(mProtocol.begin(), mProtocol.end(), end + mLabel.size());
lock.unlock();
transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
}
void NegotiatedDataChannel::processOpenMessage(message_ptr message) {
std::unique_lock lock(mMutex);
auto transport = mSctpTransport.lock();
if (!transport)
throw std::runtime_error("DataChannel has no transport");
if (message->size() < sizeof(OpenMessage))
throw std::invalid_argument("DataChannel open message too small");
OpenMessage open = *reinterpret_cast<const OpenMessage *>(message->data());
open.priority = ntohs(open.priority);
open.reliabilityParameter = ntohl(open.reliabilityParameter);
open.labelLength = ntohs(open.labelLength);
open.protocolLength = ntohs(open.protocolLength);
if (message->size() < sizeof(OpenMessage) + size_t(open.labelLength + open.protocolLength))
throw std::invalid_argument("DataChannel open message truncated");
auto end = reinterpret_cast<const char *>(message->data() + sizeof(OpenMessage));
mLabel.assign(end, open.labelLength);
mProtocol.assign(end + open.labelLength, open.protocolLength);
mReliability->unordered = (open.channelType & 0x80) != 0;
switch (open.channelType & 0x7F) {
case CHANNEL_PARTIAL_RELIABLE_REXMIT:
mReliability->type = Reliability::Type::Rexmit;
mReliability->rexmit = int(open.reliabilityParameter);
break;
case CHANNEL_PARTIAL_RELIABLE_TIMED:
mReliability->type = Reliability::Type::Timed;
mReliability->rexmit = milliseconds(open.reliabilityParameter);
break;
default:
mReliability->type = Reliability::Type::Reliable;
mReliability->rexmit = int(0);
}
lock.unlock();
binary buffer(sizeof(AckMessage), byte(0));
auto &ack = *reinterpret_cast<AckMessage *>(buffer.data());
ack.type = MESSAGE_ACK;
transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
if (!mIsOpen.exchange(true))
triggerOpen();
return impl()->outgoing(std::make_shared<Message>(data, data + size, Message::Binary));
}
} // namespace rtc

4
src/base64.cpp → src/impl/base64.cpp
View File

@ -20,7 +20,7 @@
#include "base64.hpp"
namespace rtc {
namespace rtc::impl {
using std::to_integer;
@ -59,6 +59,6 @@ string to_base64(const binary &data) {
return out;
}
} // namespace rtc
} // namespace rtc::impl
#endif

6
src/base64.hpp → src/impl/base64.hpp
View File

@ -16,14 +16,14 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_BASE64_H
#define RTC_BASE64_H
#ifndef RTC_IMPL_BASE64_H
#define RTC_IMPL_BASE64_H
#if RTC_ENABLE_WEBSOCKET
#include "include.hpp"
namespace rtc {
namespace rtc::impl {
string to_base64(const binary &data);

16
src/certificate.cpp → src/impl/certificate.cpp
View File

@ -26,14 +26,10 @@
#include <sstream>
#include <unordered_map>
using std::shared_ptr;
using std::string;
using std::unique_ptr;
namespace rtc::impl {
#if USE_GNUTLS
namespace rtc {
Certificate::Certificate(string crt_pem, string key_pem)
: mCredentials(gnutls::new_credentials(), gnutls::free_credentials) {
@ -129,12 +125,8 @@ certificate_ptr make_certificate_impl(string commonName) {
} // namespace
} // namespace rtc
#else // USE_GNUTLS==0
namespace rtc {
Certificate::Certificate(string crt_pem, string key_pem) {
BIO *bio = BIO_new(BIO_s_mem());
BIO_write(bio, crt_pem.data(), int(crt_pem.size()));
@ -230,14 +222,10 @@ certificate_ptr make_certificate_impl(string commonName) {
} // namespace
} // namespace rtc
#endif
// Common for GnuTLS and OpenSSL
namespace rtc {
namespace {
static std::unordered_map<string, future_certificate_ptr> CertificateCache;
@ -262,4 +250,4 @@ void CleanupCertificateCache() {
CertificateCache.clear();
}
} // namespace rtc
} // namespace rtc::impl

8
src/certificate.hpp → src/impl/certificate.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_CERTIFICATE_H
#define RTC_CERTIFICATE_H
#ifndef RTC_IMPL_CERTIFICATE_H
#define RTC_IMPL_CERTIFICATE_H
#include "include.hpp"
#include "tls.hpp"
@ -25,7 +25,7 @@
#include <future>
#include <tuple>
namespace rtc {
namespace rtc::impl {
class Certificate {
public:
@ -65,6 +65,6 @@ future_certificate_ptr make_certificate(string commonName = "libdatachannel"); /
void CleanupCertificateCache();
} // namespace rtc
} // namespace rtc::impl
#endif

57
src/impl/channel.cpp Normal file
View File

@ -0,0 +1,57 @@
/**
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "channel.hpp"
namespace rtc::impl {
void Channel::triggerOpen() { openCallback(); }
void Channel::triggerClosed() { closedCallback(); }
void Channel::triggerError(string error) { errorCallback(error); }
void Channel::triggerAvailable(size_t count) {
if (count == 1)
availableCallback();
while (messageCallback && count--) {
auto message = receive();
if (!message)
break;
messageCallback(*message);
}
}
void Channel::triggerBufferedAmount(size_t amount) {
size_t previous = bufferedAmount.exchange(amount);
size_t threshold = bufferedAmountLowThreshold.load();
if (previous > threshold && amount <= threshold)
bufferedAmountLowCallback();
}
void Channel::resetCallbacks() {
openCallback = nullptr;
closedCallback = nullptr;
errorCallback = nullptr;
messageCallback = nullptr;
availableCallback = nullptr;
bufferedAmountLowCallback = nullptr;
}
} // namespace rtc::impl

57
src/impl/channel.hpp Normal file
View File

@ -0,0 +1,57 @@
/**
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_IMPL_CHANNEL_H
#define RTC_IMPL_CHANNEL_H
#include "include.hpp"
#include "message.hpp"
#include <atomic>
#include <functional>
#include <variant>
namespace rtc::impl {
struct Channel {
virtual std::optional<message_variant> receive() = 0;
virtual std::optional<message_variant> peek() = 0;
virtual size_t availableAmount() const = 0;
virtual void triggerOpen();
virtual void triggerClosed();
virtual void triggerError(string error);
virtual void triggerAvailable(size_t count);
virtual void triggerBufferedAmount(size_t amount);
virtual void resetCallbacks();
synchronized_callback<> openCallback;
synchronized_callback<> closedCallback;
synchronized_callback<string> errorCallback;
synchronized_callback<message_variant> messageCallback;
synchronized_callback<> availableCallback;
synchronized_callback<> bufferedAmountLowCallback;
std::atomic<size_t> bufferedAmount = 0;
std::atomic<size_t> bufferedAmountLowThreshold = 0;
};
} // namespace rtc::impl
#endif

365
src/impl/datachannel.cpp Normal file
View File

@ -0,0 +1,365 @@
/**
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "datachannel.hpp"
#include "include.hpp"
#include "logcounter.hpp"
#include "peerconnection.hpp"
#include "sctptransport.hpp"
#include "rtc/datachannel.hpp"
#include "rtc/track.hpp"
#ifdef _WIN32
#include <winsock2.h>
#else
#include <arpa/inet.h>
#endif
using std::chrono::milliseconds;
namespace rtc::impl {
// Messages for the DataChannel establishment protocol
// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09
enum MessageType : uint8_t {
MESSAGE_OPEN_REQUEST = 0x00,
MESSAGE_OPEN_RESPONSE = 0x01,
MESSAGE_ACK = 0x02,
MESSAGE_OPEN = 0x03,
MESSAGE_CLOSE = 0x04
};
enum ChannelType : uint8_t {
CHANNEL_RELIABLE = 0x00,
CHANNEL_PARTIAL_RELIABLE_REXMIT = 0x01,
CHANNEL_PARTIAL_RELIABLE_TIMED = 0x02
};
#pragma pack(push, 1)
struct OpenMessage {
uint8_t type = MESSAGE_OPEN;
uint8_t channelType;
uint16_t priority;
uint32_t reliabilityParameter;
uint16_t labelLength;
uint16_t protocolLength;
// The following fields are:
// uint8_t[labelLength] label
// uint8_t[protocolLength] protocol
};
struct AckMessage {
uint8_t type = MESSAGE_ACK;
};
struct CloseMessage {
uint8_t type = MESSAGE_CLOSE;
};
#pragma pack(pop)
LogCounter COUNTER_USERNEG_OPEN_MESSAGE(
plog::warning, "Number of open messages for a user-negotiated DataChannel received");
DataChannel::DataChannel(weak_ptr<PeerConnection> pc, uint16_t stream, string label,
string protocol, Reliability reliability)
: mPeerConnection(pc), mStream(stream), mLabel(std::move(label)),
mProtocol(std::move(protocol)),
mReliability(std::make_shared<Reliability>(std::move(reliability))),
mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
DataChannel::~DataChannel() { close(); }
void DataChannel::close() {
std::shared_ptr<SctpTransport> transport;
{
std::shared_lock lock(mMutex);
transport = mSctpTransport.lock();
}
mIsClosed = true;
if (mIsOpen.exchange(false) && transport)
transport->closeStream(mStream);
resetCallbacks();
}
void DataChannel::remoteClose() {
if (!mIsClosed.exchange(true))
triggerClosed();
mIsOpen = false;
}
std::optional<message_variant> DataChannel::receive() {
while (auto next = mRecvQueue.tryPop()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
auto raw = reinterpret_cast<const uint8_t *>(message->data());
if (!message->empty() && raw[0] == MESSAGE_CLOSE)
remoteClose();
}
return nullopt;
}
std::optional<message_variant> DataChannel::peek() {
while (auto next = mRecvQueue.peek()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
auto raw = reinterpret_cast<const uint8_t *>(message->data());
if (!message->empty() && raw[0] == MESSAGE_CLOSE)
remoteClose();
mRecvQueue.tryPop();
}
return nullopt;
}
size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
uint16_t DataChannel::stream() const {
std::shared_lock lock(mMutex);
return mStream;
}
string DataChannel::label() const {
std::shared_lock lock(mMutex);
return mLabel;
}
string DataChannel::protocol() const {
std::shared_lock lock(mMutex);
return mProtocol;
}
Reliability DataChannel::reliability() const {
std::shared_lock lock(mMutex);
return *mReliability;
}
bool DataChannel::isOpen(void) const { return mIsOpen; }
bool DataChannel::isClosed(void) const { return mIsClosed; }
size_t DataChannel::maxMessageSize() const {
size_t remoteMax = DEFAULT_MAX_MESSAGE_SIZE;
if (auto pc = mPeerConnection.lock())
if (auto description = pc->remoteDescription())
if (auto *application = description->application())
if (auto maxMessageSize = application->maxMessageSize())
remoteMax = *maxMessageSize > 0 ? *maxMessageSize : LOCAL_MAX_MESSAGE_SIZE;
return std::min(remoteMax, LOCAL_MAX_MESSAGE_SIZE);
}
void DataChannel::shiftStream() {
if (mStream % 2 == 1)
mStream -= 1;
}
void DataChannel::open(shared_ptr<SctpTransport> transport) {
{
std::unique_lock lock(mMutex);
mSctpTransport = transport;
}
if (!mIsOpen.exchange(true))
triggerOpen();
}
void DataChannel::processOpenMessage(message_ptr) {
PLOG_DEBUG << "Received an open message for a user-negotiated DataChannel, ignoring";
COUNTER_USERNEG_OPEN_MESSAGE++;
}
bool DataChannel::outgoing(message_ptr message) {
std::shared_ptr<SctpTransport> transport;
{
std::shared_lock lock(mMutex);
transport = mSctpTransport.lock();
if (!transport || mIsClosed)
throw std::runtime_error("DataChannel is closed");
if (message->size() > maxMessageSize())
throw std::runtime_error("Message size exceeds limit");
// Before the ACK has been received on a DataChannel, all messages must be sent ordered
message->reliability = mIsOpen ? mReliability : nullptr;
message->stream = mStream;
}
return transport->send(message);
}
void DataChannel::incoming(message_ptr message) {
if (!message)
return;
switch (message->type) {
case Message::Control: {
if (message->size() == 0)
break; // Ignore
auto raw = reinterpret_cast<const uint8_t *>(message->data());
switch (raw[0]) {
case MESSAGE_OPEN:
processOpenMessage(message);
break;
case MESSAGE_ACK:
if (!mIsOpen.exchange(true)) {
triggerOpen();
}
break;
case MESSAGE_CLOSE:
// The close message will be processed in-order in receive()
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
break;
default:
// Ignore
break;
}
break;
}
case Message::String:
case Message::Binary:
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
break;
default:
// Ignore
break;
}
}
NegotiatedDataChannel::NegotiatedDataChannel(std::weak_ptr<impl::PeerConnection> pc,
uint16_t stream, string label, string protocol,
Reliability reliability)
: DataChannel(pc, stream, std::move(label), std::move(protocol), std::move(reliability)) {}
NegotiatedDataChannel::NegotiatedDataChannel(std::weak_ptr<impl::PeerConnection> pc,
std::weak_ptr<impl::SctpTransport> transport,
uint16_t stream)
: DataChannel(pc, stream, "", "", {}) {
mSctpTransport = transport;
}
NegotiatedDataChannel::~NegotiatedDataChannel() {}
void NegotiatedDataChannel::open(shared_ptr<impl::SctpTransport> transport) {
std::unique_lock lock(mMutex);
mSctpTransport = transport;
uint8_t channelType;
uint32_t reliabilityParameter;
switch (mReliability->type) {
case Reliability::Type::Rexmit:
channelType = CHANNEL_PARTIAL_RELIABLE_REXMIT;
reliabilityParameter = uint32_t(std::get<int>(mReliability->rexmit));
break;
case Reliability::Type::Timed:
channelType = CHANNEL_PARTIAL_RELIABLE_TIMED;
reliabilityParameter = uint32_t(std::get<milliseconds>(mReliability->rexmit).count());
break;
default:
channelType = CHANNEL_RELIABLE;
reliabilityParameter = 0;
break;
}
if (mReliability->unordered)
channelType |= 0x80;
const size_t len = sizeof(OpenMessage) + mLabel.size() + mProtocol.size();
binary buffer(len, byte(0));
auto &open = *reinterpret_cast<OpenMessage *>(buffer.data());
open.type = MESSAGE_OPEN;
open.channelType = channelType;
open.priority = htons(0);
open.reliabilityParameter = htonl(reliabilityParameter);
open.labelLength = htons(uint16_t(mLabel.size()));
open.protocolLength = htons(uint16_t(mProtocol.size()));
auto end = reinterpret_cast<char *>(buffer.data() + sizeof(OpenMessage));
std::copy(mLabel.begin(), mLabel.end(), end);
std::copy(mProtocol.begin(), mProtocol.end(), end + mLabel.size());
lock.unlock();
transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
}
void NegotiatedDataChannel::processOpenMessage(message_ptr message) {
std::unique_lock lock(mMutex);
auto transport = mSctpTransport.lock();
if (!transport)
throw std::runtime_error("DataChannel has no transport");
if (message->size() < sizeof(OpenMessage))
throw std::invalid_argument("DataChannel open message too small");
OpenMessage open = *reinterpret_cast<const OpenMessage *>(message->data());
open.priority = ntohs(open.priority);
open.reliabilityParameter = ntohl(open.reliabilityParameter);
open.labelLength = ntohs(open.labelLength);
open.protocolLength = ntohs(open.protocolLength);
if (message->size() < sizeof(OpenMessage) + size_t(open.labelLength + open.protocolLength))
throw std::invalid_argument("DataChannel open message truncated");
auto end = reinterpret_cast<const char *>(message->data() + sizeof(OpenMessage));
mLabel.assign(end, open.labelLength);
mProtocol.assign(end + open.labelLength, open.protocolLength);
mReliability->unordered = (open.channelType & 0x80) != 0;
switch (open.channelType & 0x7F) {
case CHANNEL_PARTIAL_RELIABLE_REXMIT:
mReliability->type = Reliability::Type::Rexmit;
mReliability->rexmit = int(open.reliabilityParameter);
break;
case CHANNEL_PARTIAL_RELIABLE_TIMED:
mReliability->type = Reliability::Type::Timed;
mReliability->rexmit = milliseconds(open.reliabilityParameter);
break;
default:
mReliability->type = Reliability::Type::Reliable;
mReliability->rexmit = int(0);
}
lock.unlock();
binary buffer(sizeof(AckMessage), byte(0));
auto &ack = *reinterpret_cast<AckMessage *>(buffer.data());
ack.type = MESSAGE_ACK;
transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
if (!mIsOpen.exchange(true))
triggerOpen();
}
} // namespace rtc::impl

94
src/impl/datachannel.hpp Normal file
View File

@ -0,0 +1,94 @@
/**
* Copyright (c) 2019 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_IMPL_DATA_CHANNEL_H
#define RTC_IMPL_DATA_CHANNEL_H
#include "channel.hpp"
#include "include.hpp"
#include "message.hpp"
#include "peerconnection.hpp"
#include "queue.hpp"
#include "reliability.hpp"
#include "sctptransport.hpp"
#include <atomic>
namespace rtc::impl {
struct PeerConnection;
struct DataChannel : Channel, std::enable_shared_from_this<DataChannel> {
DataChannel(weak_ptr<impl::PeerConnection> pc, uint16_t stream, string label, string protocol,
Reliability reliability);
~DataChannel();
void close();
void remoteClose();
bool outgoing(message_ptr message);
void incoming(message_ptr message);
std::optional<message_variant> receive() override;
std::optional<message_variant> peek() override;
size_t availableAmount() const override;
uint16_t stream() const;
string label() const;
string protocol() const;
Reliability reliability() const;
bool isOpen(void) const;
bool isClosed(void) const;
size_t maxMessageSize() const;
void shiftStream();
virtual void open(shared_ptr<SctpTransport> transport);
virtual void processOpenMessage(message_ptr);
protected:
const weak_ptr<impl::PeerConnection> mPeerConnection;
weak_ptr<SctpTransport> mSctpTransport;
uint16_t mStream;
string mLabel;
string mProtocol;
shared_ptr<Reliability> mReliability;
mutable std::shared_mutex mMutex;
Queue<message_ptr> mRecvQueue;
std::atomic<bool> mIsOpen = false;
std::atomic<bool> mIsClosed = false;
};
struct NegotiatedDataChannel final : public DataChannel {
NegotiatedDataChannel(weak_ptr<impl::PeerConnection> pc, uint16_t stream, string label,
string protocol, Reliability reliability);
NegotiatedDataChannel(weak_ptr<impl::PeerConnection> pc,
weak_ptr<impl::SctpTransport> transport, uint16_t stream);
~NegotiatedDataChannel();
void open(impl_ptr<impl::SctpTransport> transport) override;
void processOpenMessage(message_ptr message) override;
};
} // namespace rtc::impl
#endif

6
src/dtlssrtptransport.cpp → src/impl/dtlssrtptransport.cpp
View File

@ -19,17 +19,17 @@
#include "dtlssrtptransport.hpp"
#include "logcounter.hpp"
#include "tls.hpp"
#include "rtp.hpp"
#if RTC_ENABLE_MEDIA
#include <cstring>
#include <exception>
using std::shared_ptr;
using std::to_integer;
using std::to_string;
namespace rtc {
namespace rtc::impl {
static LogCounter COUNTER_MEDIA_TRUNCATED(plog::warning,
"Number of truncated SRT(C)P packets received");
@ -48,7 +48,7 @@ static LogCounter COUNTER_SRTP_REPLAY(plog::warning, "Number of SRTP replay pack
static LogCounter
COUNTER_SRTP_AUTH_FAIL(plog::warning,
"Number of SRTP packets received that failed authentication checks");
static rtc::LogCounter
static LogCounter
COUNTER_SRTP_FAIL(plog::warning,
"Number of SRTP packets received that had an unknown libSRTP failure");

6
src/dtlssrtptransport.hpp → src/impl/dtlssrtptransport.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_DTLS_SRTP_TRANSPORT_H
#define RTC_DTLS_SRTP_TRANSPORT_H
#ifndef RTC_IMPL_DTLS_SRTP_TRANSPORT_H
#define RTC_IMPL_DTLS_SRTP_TRANSPORT_H
#include "dtlstransport.hpp"
#include "include.hpp"
@ -32,7 +32,7 @@
#include <atomic>
namespace rtc {
namespace rtc::impl {
class DtlsSrtpTransport final : public DtlsTransport {
public:

2
src/dtlstransport.cpp → src/impl/dtlstransport.cpp
View File

@ -39,7 +39,7 @@ using std::string;
using std::unique_ptr;
using std::weak_ptr;
namespace rtc {
namespace rtc::impl {
#if USE_GNUTLS

7
src/dtlstransport.hpp → src/impl/dtlstransport.hpp
View File

@ -16,12 +16,11 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_DTLS_TRANSPORT_H
#define RTC_DTLS_TRANSPORT_H
#ifndef RTC_IMPL_DTLS_TRANSPORT_H
#define RTC_IMPL_DTLS_TRANSPORT_H
#include "certificate.hpp"
#include "include.hpp"
#include "peerconnection.hpp"
#include "queue.hpp"
#include "tls.hpp"
#include "transport.hpp"
@ -32,7 +31,7 @@
#include <mutex>
#include <thread>
namespace rtc {
namespace rtc::impl {
class IceTransport;

25
src/icetransport.cpp → src/impl/icetransport.cpp
View File

@ -37,16 +37,13 @@
#include <sys/types.h>
using namespace std::chrono_literals;
using std::shared_ptr;
using std::weak_ptr;
using std::chrono::system_clock;
#if !USE_NICE
#define MAX_TURN_SERVERS_COUNT 2
namespace rtc {
namespace rtc::impl {
IceTransport::IceTransport(const Configuration &config, candidate_callback candidateCallback,
state_callback stateChangeCallback,
@ -282,7 +279,7 @@ void IceTransport::processCandidate(const string &candidate) {
void IceTransport::processGatheringDone() { changeGatheringState(GatheringState::Complete); }
void IceTransport::StateChangeCallback(juice_agent_t *, juice_state_t state, void *user_ptr) {
auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(user_ptr);
try {
iceTransport->processStateChange(static_cast<unsigned int>(state));
} catch (const std::exception &e) {
@ -291,7 +288,7 @@ void IceTransport::StateChangeCallback(juice_agent_t *, juice_state_t state, voi
}
void IceTransport::CandidateCallback(juice_agent_t *, const char *sdp, void *user_ptr) {
auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(user_ptr);
try {
iceTransport->processCandidate(sdp);
} catch (const std::exception &e) {
@ -300,7 +297,7 @@ void IceTransport::CandidateCallback(juice_agent_t *, const char *sdp, void *use
}
void IceTransport::GatheringDoneCallback(juice_agent_t *, void *user_ptr) {
auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(user_ptr);
try {
iceTransport->processGatheringDone();
} catch (const std::exception &e) {
@ -309,7 +306,7 @@ void IceTransport::GatheringDoneCallback(juice_agent_t *, void *user_ptr) {
}
void IceTransport::RecvCallback(juice_agent_t *, const char *data, size_t size, void *user_ptr) {
auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(user_ptr);
try {
PLOG_VERBOSE << "Incoming size=" << size;
auto b = reinterpret_cast<const byte *>(data);
@ -341,7 +338,7 @@ void IceTransport::LogCallback(juice_log_level_t level, const char *message) {
PLOG(severity) << "juice: " << message;
}
} // namespace rtc
} // namespace rtc::impl
#else // USE_NICE == 1
@ -717,7 +714,7 @@ string IceTransport::AddressToString(const NiceAddress &addr) {
void IceTransport::CandidateCallback(NiceAgent *agent, NiceCandidate *candidate,
gpointer userData) {
auto iceTransport = static_cast<rtc::IceTransport *>(userData);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(userData);
gchar *cand = nice_agent_generate_local_candidate_sdp(agent, candidate);
try {
iceTransport->processCandidate(cand);
@ -729,7 +726,7 @@ void IceTransport::CandidateCallback(NiceAgent *agent, NiceCandidate *candidate,
void IceTransport::GatheringDoneCallback(NiceAgent * /*agent*/, guint /*streamId*/,
gpointer userData) {
auto iceTransport = static_cast<rtc::IceTransport *>(userData);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(userData);
try {
iceTransport->processGatheringDone();
} catch (const std::exception &e) {
@ -739,7 +736,7 @@ void IceTransport::GatheringDoneCallback(NiceAgent * /*agent*/, guint /*streamId
void IceTransport::StateChangeCallback(NiceAgent * /*agent*/, guint /*streamId*/,
guint /*componentId*/, guint state, gpointer userData) {
auto iceTransport = static_cast<rtc::IceTransport *>(userData);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(userData);
try {
iceTransport->processStateChange(state);
} catch (const std::exception &e) {
@ -749,7 +746,7 @@ void IceTransport::StateChangeCallback(NiceAgent * /*agent*/, guint /*streamId*/
void IceTransport::RecvCallback(NiceAgent * /*agent*/, guint /*streamId*/, guint /*componentId*/,
guint len, gchar *buf, gpointer userData) {
auto iceTransport = static_cast<rtc::IceTransport *>(userData);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(userData);
try {
PLOG_VERBOSE << "Incoming size=" << len;
auto b = reinterpret_cast<byte *>(buf);
@ -760,7 +757,7 @@ void IceTransport::RecvCallback(NiceAgent * /*agent*/, guint /*streamId*/, guint
}
gboolean IceTransport::TimeoutCallback(gpointer userData) {
auto iceTransport = static_cast<rtc::IceTransport *>(userData);
auto iceTransport = static_cast<rtc::impl::IceTransport *>(userData);
try {
iceTransport->processTimeout();
} catch (const std::exception &e) {

6
src/icetransport.hpp → src/impl/icetransport.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_ICE_TRANSPORT_H
#define RTC_ICE_TRANSPORT_H
#ifndef RTC_IMPL_ICE_TRANSPORT_H
#define RTC_IMPL_ICE_TRANSPORT_H
#include "candidate.hpp"
#include "configuration.hpp"
@ -37,7 +37,7 @@
#include <mutex>
#include <thread>
namespace rtc {
namespace rtc::impl {
class IceTransport : public Transport {
public:

2
src/logcounter.cpp → src/impl/logcounter.cpp
View File

@ -18,7 +18,7 @@
#include "logcounter.hpp"
namespace rtc {
namespace rtc::impl {
LogCounter::LogCounter(plog::Severity severity, const std::string &text,
std::chrono::seconds duration) {

2
src/logcounter.hpp → src/impl/logcounter.hpp
View File

@ -25,7 +25,7 @@
#include <atomic>
#include <chrono>
namespace rtc {
namespace rtc::impl {
class LogCounter {
private:

1009
src/impl/peerconnection.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

129
src/impl/peerconnection.hpp Normal file
View File

@ -0,0 +1,129 @@
/**
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_IMPL_PEER_CONNECTION_H
#define RTC_IMPL_PEER_CONNECTION_H
#include "datachannel.hpp"
#include "dtlstransport.hpp"
#include "icetransport.hpp"
#include "include.hpp"
#include "sctptransport.hpp"
#include "track.hpp"
#include "rtc/peerconnection.hpp"
namespace rtc::impl {
struct PeerConnection : std::enable_shared_from_this<PeerConnection> {
using State = rtc::PeerConnection::State;
using GatheringState = rtc::PeerConnection::GatheringState;
using SignalingState = rtc::PeerConnection::SignalingState;
PeerConnection(Configuration config_);
~PeerConnection();
void close();
std::optional<Description> localDescription() const;
std::optional<Description> remoteDescription() const;
std::shared_ptr<IceTransport> initIceTransport();
std::shared_ptr<DtlsTransport> initDtlsTransport();
std::shared_ptr<SctpTransport> initSctpTransport();
std::shared_ptr<IceTransport> getIceTransport() const;
std::shared_ptr<DtlsTransport> getDtlsTransport() const;
std::shared_ptr<SctpTransport> getSctpTransport() const;
void closeTransports();
void endLocalCandidates();
void rollbackLocalDescription();
bool checkFingerprint(const std::string &fingerprint) const;
void forwardMessage(message_ptr message);
void forwardMedia(message_ptr message);
void forwardBufferedAmount(uint16_t stream, size_t amount);
std::optional<string> getMidFromSsrc(uint32_t ssrc);
shared_ptr<DataChannel> emplaceDataChannel(Description::Role role, string label,
DataChannelInit init);
shared_ptr<DataChannel> findDataChannel(uint16_t stream);
void shiftDataChannels();
void iterateDataChannels(std::function<void(std::shared_ptr<DataChannel> channel)> func);
void openDataChannels();
void closeDataChannels();
void remoteCloseDataChannels();
shared_ptr<Track> emplaceTrack(Description::Media description);
void incomingTrack(Description::Media description);
void openTracks();
void validateRemoteDescription(const Description &description);
void processLocalDescription(Description description);
void processLocalCandidate(Candidate candidate);
void processRemoteDescription(Description description);
void processRemoteCandidate(Candidate candidate);
string localBundleMid() const;
void triggerDataChannel(std::weak_ptr<DataChannel> weakDataChannel);
void triggerTrack(std::shared_ptr<Track> track);
bool changeState(State newState);
bool changeGatheringState(GatheringState newState);
bool changeSignalingState(SignalingState newState);
void resetCallbacks();
void outgoingMedia(message_ptr message);
const Configuration config;
std::atomic<State> state = State::New;
std::atomic<GatheringState> gatheringState = GatheringState::New;
std::atomic<SignalingState> signalingState = SignalingState::Stable;
std::atomic<bool> negotiationNeeded = false;
synchronized_callback<std::shared_ptr<rtc::DataChannel>> dataChannelCallback;
synchronized_callback<Description> localDescriptionCallback;
synchronized_callback<Candidate> localCandidateCallback;
synchronized_callback<State> stateChangeCallback;
synchronized_callback<GatheringState> gatheringStateChangeCallback;
synchronized_callback<SignalingState> signalingStateChangeCallback;
synchronized_callback<std::shared_ptr<rtc::Track>> trackCallback;
private:
const init_token mInitToken = Init::Token();
const future_certificate_ptr mCertificate;
const std::unique_ptr<Processor> mProcessor;
std::optional<Description> mLocalDescription, mRemoteDescription;
std::optional<Description> mCurrentLocalDescription;
mutable std::mutex mLocalDescriptionMutex, mRemoteDescriptionMutex;
std::shared_ptr<IceTransport> mIceTransport;
std::shared_ptr<DtlsTransport> mDtlsTransport;
std::shared_ptr<SctpTransport> mSctpTransport;
std::unordered_map<uint16_t, std::weak_ptr<DataChannel>> mDataChannels; // by stream ID
std::unordered_map<string, std::weak_ptr<Track>> mTracks; // by mid
std::vector<std::weak_ptr<Track>> mTrackLines; // by SDP order
std::shared_mutex mDataChannelsMutex, mTracksMutex;
std::unordered_map<uint32_t, string> mMidFromSsrc; // cache
};
} // namespace rtc::impl
#endif

4
src/processor.cpp → src/impl/processor.cpp
View File

@ -18,7 +18,7 @@
#include "processor.hpp"
namespace rtc {
namespace rtc::impl {
Processor::Processor(size_t limit) : mTasks(limit) {}
@ -40,4 +40,4 @@ void Processor::schedule() {
}
}
} // namespace rtc
} // namespace rtc::impl

8
src/processor.hpp → src/impl/processor.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_PROCESSOR_H
#define RTC_PROCESSOR_H
#ifndef RTC_IMPL_PROCESSOR_H
#define RTC_IMPL_PROCESSOR_H
#include "include.hpp"
#include "init.hpp"
@ -30,7 +30,7 @@
#include <mutex>
#include <queue>
namespace rtc {
namespace rtc::impl {
// Processed tasks in order by delegating them to the thread pool
class Processor final {
@ -76,6 +76,6 @@ template <class F, class... Args> void Processor::enqueue(F &&f, Args &&...args)
}
}
} // namespace rtc
} // namespace rtc::impl
#endif

2
src/sctptransport.cpp → src/impl/sctptransport.cpp
View File

@ -54,7 +54,7 @@
using namespace std::chrono_literals;
using namespace std::chrono;
namespace rtc {
namespace rtc::impl {
static LogCounter COUNTER_UNKNOWN_PPID(plog::warning,
"Number of SCTP packets received with an unknown PPID");

11
src/sctptransport.hpp → src/impl/sctptransport.hpp
View File

@ -1,5 +1,5 @@
/**
* Copyright (c) 2019 Paul-Louis Ageneau
* Copyright (c) 2019-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@ -16,11 +16,10 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_SCTP_TRANSPORT_H
#define RTC_SCTP_TRANSPORT_H
#ifndef RTC_IMPL_SCTP_TRANSPORT_H
#define RTC_IMPL_SCTP_TRANSPORT_H
#include "include.hpp"
#include "peerconnection.hpp"
#include "processor.hpp"
#include "queue.hpp"
#include "transport.hpp"
@ -34,7 +33,7 @@
#include "usrsctp.h"
namespace rtc {
namespace rtc::impl {
class SctpTransport final : public Transport {
public:
@ -120,6 +119,6 @@ private:
static std::shared_mutex InstancesMutex;
};
} // namespace rtc
} // namespace rtc::impl
#endif

4
src/tcptransport.cpp → src/impl/tcptransport.cpp
View File

@ -27,7 +27,7 @@
#include <unistd.h>
#endif
namespace rtc {
namespace rtc::impl {
using std::to_string;
@ -398,6 +398,6 @@ int TcpTransport::prepareSelect(fd_set &readfds, fd_set &writefds) {
void TcpTransport::interruptSelect() { mInterrupter.interrupt(); }
} // namespace rtc
} // namespace rtc::impl
#endif

8
src/tcptransport.hpp → src/impl/tcptransport.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_TCP_TRANSPORT_H
#define RTC_TCP_TRANSPORT_H
#ifndef RTC_IMPL_TCP_TRANSPORT_H
#define RTC_IMPL_TCP_TRANSPORT_H
#include "include.hpp"
#include "queue.hpp"
@ -31,7 +31,7 @@
// Use the socket defines from libjuice
#include "../deps/libjuice/src/socket.h"
namespace rtc {
namespace rtc::impl {
// Utility class to interrupt select()
class SelectInterrupter {
@ -85,7 +85,7 @@ private:
Queue<message_ptr> mSendQueue;
};
} // namespace rtc
} // namespace rtc::impl
#endif

6
src/threadpool.cpp → src/impl/threadpool.cpp
View File

@ -22,11 +22,11 @@
namespace {
void joinThreadPoolInstance() { rtc::ThreadPool::Instance().join(); }
void joinThreadPoolInstance() { rtc::impl::ThreadPool::Instance().join(); }
} // namespace
namespace rtc {
namespace rtc::impl {
ThreadPool &ThreadPool::Instance() {
static ThreadPool *instance = new ThreadPool;
@ -103,4 +103,4 @@ std::function<void()> ThreadPool::dequeue() {
return nullptr;
}
} // namespace rtc
} // namespace rtc::impl

8
src/threadpool.hpp → src/impl/threadpool.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_THREADPOOL_H
#define RTC_THREADPOOL_H
#ifndef RTC_IMPL_THREADPOOL_H
#define RTC_IMPL_THREADPOOL_H
#include "include.hpp"
#include "init.hpp"
@ -34,7 +34,7 @@
#include <thread>
#include <vector>
namespace rtc {
namespace rtc::impl {
template <class F, class... Args>
using invoke_future_t = std::future<std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>>;
@ -119,6 +119,6 @@ auto ThreadPool::schedule(clock::time_point time, F &&f, Args &&...args)
return result;
}
} // namespace rtc
} // namespace rtc::impl
#endif

0
src/tls.cpp → src/impl/tls.cpp
View File

0
src/tls.hpp → src/impl/tls.hpp
View File

4
src/tlstransport.cpp → src/impl/tlstransport.cpp
View File

@ -33,7 +33,7 @@ using std::string;
using std::unique_ptr;
using std::weak_ptr;
namespace rtc {
namespace rtc::impl {
#if USE_GNUTLS
@ -443,6 +443,6 @@ void TlsTransport::InfoCallback(const SSL *ssl, int where, int ret) {
#endif
} // namespace rtc
} // namespace rtc::impl
#endif

8
src/tlstransport.hpp → src/impl/tlstransport.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_TLS_TRANSPORT_H
#define RTC_TLS_TRANSPORT_H
#ifndef RTC_IMPL_TLS_TRANSPORT_H
#define RTC_IMPL_TLS_TRANSPORT_H
#include "include.hpp"
#include "queue.hpp"
@ -28,7 +28,7 @@
#include <thread>
namespace rtc {
namespace rtc::impl {
class TcpTransport;
@ -76,7 +76,7 @@ protected:
#endif
};
} // namespace rtc
} // namespace rtc::impl
#endif

181
src/impl/track.cpp Normal file
View File

@ -0,0 +1,181 @@
/**
* Copyright (c) 2020 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "track.hpp"
#include "logcounter.hpp"
namespace rtc::impl {
static LogCounter COUNTER_MEDIA_BAD_DIRECTION(plog::warning,
"Number of media packets sent in invalid directions");
static LogCounter COUNTER_QUEUE_FULL(plog::warning,
"Number of media packets dropped due to a full queue");
Track::Track(Description::Media description)
: mMediaDescription(std::move(description)), mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
string Track::mid() const {
std::shared_lock lock(mMutex);
return mMediaDescription.mid();
}
Description::Direction Track::direction() const {
std::shared_lock lock(mMutex);
return mMediaDescription.direction();
}
Description::Media Track::description() const {
std::shared_lock lock(mMutex);
return mMediaDescription;
}
void Track::setDescription(Description::Media description) {
std::unique_lock lock(mMutex);
if (description.mid() != mMediaDescription.mid())
throw std::logic_error("Media description mid does not match track mid");
mMediaDescription = std::move(description);
}
void Track::close() {
mIsClosed = true;
setRtcpHandler(nullptr);
resetCallbacks();
}
std::optional<message_variant> Track::receive() {
if (auto next = mRecvQueue.tryPop())
return to_variant(std::move(**next));
return nullopt;
}
std::optional<message_variant> Track::peek() {
if (auto next = mRecvQueue.peek())
return to_variant(std::move(**next));
return nullopt;
}
bool Track::isOpen(void) const {
#if RTC_ENABLE_MEDIA
std::shared_lock lock(mMutex);
return !mIsClosed && mDtlsSrtpTransport.lock();
#else
return !mIsClosed;
#endif
}
bool Track::isClosed(void) const { return mIsClosed; }
size_t Track::availableAmount() const { return mRecvQueue.amount(); }
#if RTC_ENABLE_MEDIA
void Track::open(shared_ptr<DtlsSrtpTransport> transport) {
{
std::lock_guard lock(mMutex);
mDtlsSrtpTransport = transport;
}
triggerOpen();
}
#endif
void Track::incoming(message_ptr message) {
if (!message)
return;
// TODO
auto dir = direction();
if ((dir == Description::Direction::SendOnly || dir == Description::Direction::Inactive) &&
message->type != Message::Control) {
COUNTER_MEDIA_BAD_DIRECTION++;
return;
}
if (auto handler = getRtcpHandler()) {
message = handler->incoming(message);
if (!message)
return;
}
// Tail drop if queue is full
if (mRecvQueue.full()) {
COUNTER_QUEUE_FULL++;
return;
}
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
}
bool Track::outgoing([[maybe_unused]] message_ptr message) {
if (mIsClosed)
throw std::runtime_error("Track is closed");
auto dir = direction();
if ((dir == Description::Direction::RecvOnly || dir == Description::Direction::Inactive)) {
COUNTER_MEDIA_BAD_DIRECTION++;
return false;
}
if (auto handler = getRtcpHandler()) {
message = handler->outgoing(message);
if (!message)
return false;
}
#if RTC_ENABLE_MEDIA
std::shared_ptr<DtlsSrtpTransport> transport;
{
std::shared_lock lock(mMutex);
transport = mDtlsSrtpTransport.lock();
if (!transport)
throw std::runtime_error("Track is closed");
// Set recommended medium-priority DSCP value
// See https://tools.ietf.org/html/draft-ietf-tsvwg-rtcweb-qos-18
if (mMediaDescription.type() == "audio")
message->dscp = 46; // EF: Expedited Forwarding
else
message->dscp = 36; // AF42: Assured Forwarding class 4, medium drop probability
}
return transport->sendMedia(message);
#else
PLOG_WARNING << "Ignoring track send (not compiled with media support)";
return false;
#endif
}
void Track::setRtcpHandler(std::shared_ptr<MediaHandler> handler) {
{
std::unique_lock lock(mMutex);
mRtcpHandler = handler;
}
handler->onOutgoing(std::bind(&Track::outgoing, this, std::placeholders::_1));
}
std::shared_ptr<MediaHandler> Track::getRtcpHandler() {
std::shared_lock lock(mMutex);
return mRtcpHandler;
}
} // namespace rtc::impl

82
src/impl/track.hpp Normal file
View File

@ -0,0 +1,82 @@
/**
* Copyright (c) 2020-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_IMPL_TRACK_H
#define RTC_IMPL_TRACK_H
#include "channel.hpp"
#include "description.hpp"
#include "include.hpp"
#include "mediahandler.hpp"
#if RTC_ENABLE_MEDIA
#include "dtlssrtptransport.hpp"
#endif
#include <atomic>
#include <shared_mutex>
#include <variant>
namespace rtc::impl {
class Track final : public std::enable_shared_from_this<Track>, public Channel {
public:
Track(Description::Media description);
~Track() = default;
void close();
void incoming(message_ptr message);
bool outgoing(message_ptr message);
std::optional<message_variant> receive() override;
std::optional<message_variant> peek() override;
size_t availableAmount() const override;
bool isOpen() const;
bool isClosed() const;
string mid() const;
Description::Direction direction() const;
Description::Media description() const;
void setDescription(Description::Media description);
std::shared_ptr<MediaHandler> getRtcpHandler();
void setRtcpHandler(shared_ptr<MediaHandler> handler);
#if RTC_ENABLE_MEDIA
void open(std::shared_ptr<DtlsSrtpTransport> transport);
#endif
private:
#if RTC_ENABLE_MEDIA
weak_ptr<DtlsSrtpTransport> mDtlsSrtpTransport;
#endif
Description::Media mMediaDescription;
shared_ptr<MediaHandler> mRtcpHandler;
mutable std::shared_mutex mMutex;
std::atomic<bool> mIsClosed = false;
Queue<message_ptr> mRecvQueue;
};
} // namespace rtc::impl
#endif

8
src/transport.hpp → src/impl/transport.hpp
View File

@ -16,8 +16,8 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_TRANSPORT_H
#define RTC_TRANSPORT_H
#ifndef RTC_IMPL_TRANSPORT_H
#define RTC_IMPL_TRANSPORT_H
#include "include.hpp"
#include "message.hpp"
@ -26,7 +26,7 @@
#include <functional>
#include <memory>
namespace rtc {
namespace rtc::impl {
class Transport {
public:
@ -98,6 +98,6 @@ private:
std::atomic<bool> mStopped = true;
};
} // namespace rtc
} // namespace rtc::impl
#endif

4
src/verifiedtlstransport.cpp → src/impl/verifiedtlstransport.cpp
View File

@ -26,7 +26,7 @@ using std::string;
using std::unique_ptr;
using std::weak_ptr;
namespace rtc {
namespace rtc::impl {
VerifiedTlsTransport::VerifiedTlsTransport(shared_ptr<TcpTransport> lower, string host,
state_callback callback)
@ -44,6 +44,6 @@ VerifiedTlsTransport::VerifiedTlsTransport(shared_ptr<TcpTransport> lower, strin
VerifiedTlsTransport::~VerifiedTlsTransport() {}
} // namespace rtc
} // namespace rtc::impl
#endif

8
src/verifiedtlstransport.hpp → src/impl/verifiedtlstransport.hpp
View File

@ -16,14 +16,14 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_VERIFIED_TLS_TRANSPORT_H
#define RTC_VERIFIED_TLS_TRANSPORT_H
#ifndef RTC_IMPL_VERIFIED_TLS_TRANSPORT_H
#define RTC_IMPL_VERIFIED_TLS_TRANSPORT_H
#include "tlstransport.hpp"
#if RTC_ENABLE_WEBSOCKET
namespace rtc {
namespace rtc::impl {
class VerifiedTlsTransport final : public TlsTransport {
public:
@ -31,7 +31,7 @@ public:
~VerifiedTlsTransport();
};
} // namespace rtc
} // namespace rtc::impl
#endif

370
src/impl/websocket.cpp Normal file
View File

@ -0,0 +1,370 @@
/**
* Copyright (c) 2020-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#if RTC_ENABLE_WEBSOCKET
#include "websocket.hpp"
#include "include.hpp"
#include "threadpool.hpp"
#include "tcptransport.hpp"
#include "tlstransport.hpp"
#include "verifiedtlstransport.hpp"
#include "wstransport.hpp"
#include <regex>
#ifdef _WIN32
#include <winsock2.h>
#endif
namespace rtc::impl {
using namespace std::placeholders;
WebSocket::WebSocket(Configuration config_)
: config(std::move(config_)), mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {
PLOG_VERBOSE << "Creating WebSocket";
}
WebSocket::~WebSocket() {
PLOG_VERBOSE << "Destroying WebSocket";
remoteClose();
}
void WebSocket::parse(const string &url) {
PLOG_VERBOSE << "Opening WebSocket to URL: " << url;
if (state != State::Closed)
throw std::logic_error("WebSocket must be closed before opening");
// Modified regex from RFC 3986, see https://tools.ietf.org/html/rfc3986#appendix-B
static const char *rs =
R"(^(([^:.@/?#]+):)?(/{0,2}((([^:@]*)(:([^@]*))?)@)?(([^:/?#]*)(:([^/?#]*))?))?([^?#]*)(\?([^#]*))?(#(.*))?)";
static const std::regex r(rs, std::regex::extended);
std::smatch m;
if (!std::regex_match(url, m, r) || m[10].length() == 0)
throw std::invalid_argument("Invalid WebSocket URL: " + url);
mScheme = m[2];
if (mScheme.empty())
mScheme = "ws";
else if (mScheme != "ws" && mScheme != "wss")
throw std::invalid_argument("Invalid WebSocket scheme: " + mScheme);
mHostname = m[10];
mService = m[12];
if (mService.empty()) {
mService = mScheme == "ws" ? "80" : "443";
mHost = mHostname;
} else {
mHost = mHostname + ':' + mService;
}
while (!mHostname.empty() && mHostname.front() == '[')
mHostname.erase(mHostname.begin());
while (!mHostname.empty() && mHostname.back() == ']')
mHostname.pop_back();
mPath = m[13];
if (mPath.empty())
mPath += '/';
if (string query = m[15]; !query.empty())
mPath += "?" + query;
changeState(State::Connecting);
initTcpTransport();
}
void WebSocket::close() {
auto s = state.load();
if (s == State::Connecting || s == State::Open) {
PLOG_VERBOSE << "Closing WebSocket";
changeState(State::Closing);
if (auto transport = std::atomic_load(&mWsTransport))
transport->close();
else
changeState(State::Closed);
}
}
void WebSocket::remoteClose() {
if (state.load() != State::Closed) {
close();
closeTransports();
}
}
bool WebSocket::isOpen() const { return state == State::Open; }
bool WebSocket::isClosed() const { return state == State::Closed; }
size_t WebSocket::maxMessageSize() const { return DEFAULT_MAX_MESSAGE_SIZE; }
std::optional<message_variant> WebSocket::receive() {
while (auto next = mRecvQueue.tryPop()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
}
return nullopt;
}
std::optional<message_variant> WebSocket::peek() {
while (auto next = mRecvQueue.peek()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
mRecvQueue.tryPop();
}
return nullopt;
}
size_t WebSocket::availableAmount() const { return mRecvQueue.amount(); }
bool WebSocket::changeState(State newState) { return state.exchange(newState) != newState; }
bool WebSocket::outgoing(message_ptr message) {
if (state != State::Open || !mWsTransport)
throw std::runtime_error("WebSocket is not open");
if (message->size() > maxMessageSize())
throw std::runtime_error("Message size exceeds limit");
return mWsTransport->send(message);
}
void WebSocket::incoming(message_ptr message) {
if (!message) {
remoteClose();
return;
}
if (message->type == Message::String || message->type == Message::Binary) {
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
}
}
shared_ptr<TcpTransport> WebSocket::initTcpTransport() {
PLOG_VERBOSE << "Starting TCP transport";
using State = TcpTransport::State;
try {
if (auto transport = std::atomic_load(&mTcpTransport))
return transport;
auto transport = std::make_shared<TcpTransport>(
mHostname, mService, [this, weak_this = weak_from_this()](State transportState) {
auto shared_this = weak_this.lock();
if (!shared_this)
return;
switch (transportState) {
case State::Connected:
if (mScheme == "ws")
initWsTransport();
else
initTlsTransport();
break;
case State::Failed:
triggerError("TCP connection failed");
remoteClose();
break;
case State::Disconnected:
remoteClose();
break;
default:
// Ignore
break;
}
});
std::atomic_store(&mTcpTransport, transport);
if (state == WebSocket::State::Closed) {
mTcpTransport.reset();
throw std::runtime_error("Connection is closed");
}
transport->start();
return transport;
} catch (const std::exception &e) {
PLOG_ERROR << e.what();
remoteClose();
throw std::runtime_error("TCP transport initialization failed");
}
}
shared_ptr<TlsTransport> WebSocket::initTlsTransport() {
PLOG_VERBOSE << "Starting TLS transport";
using State = TlsTransport::State;
try {
if (auto transport = std::atomic_load(&mTlsTransport))
return transport;
auto lower = std::atomic_load(&mTcpTransport);
auto stateChangeCallback = [this, weak_this = weak_from_this()](State transportState) {
auto shared_this = weak_this.lock();
if (!shared_this)
return;
switch (transportState) {
case State::Connected:
initWsTransport();
break;
case State::Failed:
triggerError("TCP connection failed");
remoteClose();
break;
case State::Disconnected:
remoteClose();
break;
default:
// Ignore
break;
}
};
shared_ptr<TlsTransport> transport;
#ifdef _WIN32
if (!config.disableTlsVerification) {
PLOG_WARNING << "TLS certificate verification with root CA is not supported on Windows";
}
transport = std::make_shared<TlsTransport>(lower, mHostname, stateChangeCallback);
#else
if (config.disableTlsVerification)
transport = std::make_shared<TlsTransport>(lower, mHostname, stateChangeCallback);
else
transport =
std::make_shared<VerifiedTlsTransport>(lower, mHostname, stateChangeCallback);
#endif
std::atomic_store(&mTlsTransport, transport);
if (state == WebSocket::State::Closed) {
mTlsTransport.reset();
throw std::runtime_error("Connection is closed");
}
transport->start();
return transport;
} catch (const std::exception &e) {
PLOG_ERROR << e.what();
remoteClose();
throw std::runtime_error("TLS transport initialization failed");
}
}
shared_ptr<WsTransport> WebSocket::initWsTransport() {
PLOG_VERBOSE << "Starting WebSocket transport";
using State = WsTransport::State;
try {
if (auto transport = std::atomic_load(&mWsTransport))
return transport;
shared_ptr<Transport> lower = std::atomic_load(&mTlsTransport);
if (!lower)
lower = std::atomic_load(&mTcpTransport);
WsTransport::Configuration wsConfig = {};
wsConfig.host = mHost;
wsConfig.path = mPath;
wsConfig.protocols = config.protocols;
auto transport = std::make_shared<WsTransport>(
lower, wsConfig, weak_bind(&WebSocket::incoming, this, _1),
[this, weak_this = weak_from_this()](State transportState) {
auto shared_this = weak_this.lock();
if (!shared_this)
return;
switch (transportState) {
case State::Connected:
if (state == WebSocket::State::Connecting) {
PLOG_DEBUG << "WebSocket open";
changeState(WebSocket::State::Open);
triggerOpen();
}
break;
case State::Failed:
triggerError("WebSocket connection failed");
remoteClose();
break;
case State::Disconnected:
remoteClose();
break;
default:
// Ignore
break;
}
});
std::atomic_store(&mWsTransport, transport);
if (state == WebSocket::State::Closed) {
mWsTransport.reset();
throw std::runtime_error("Connection is closed");
}
transport->start();
return transport;
} catch (const std::exception &e) {
PLOG_ERROR << e.what();
remoteClose();
throw std::runtime_error("WebSocket transport initialization failed");
}
}
std::shared_ptr<TcpTransport> WebSocket::getTcpTransport() const {
return std::atomic_load(&mTcpTransport);
}
std::shared_ptr<TlsTransport> WebSocket::getTlsTransport() const {
return std::atomic_load(&mTlsTransport);
}
std::shared_ptr<WsTransport> WebSocket::getWsTransport() const {
return std::atomic_load(&mWsTransport);
}
void WebSocket::closeTransports() {
PLOG_VERBOSE << "Closing transports";
if (state.load() != State::Closed) {
changeState(State::Closed);
triggerClosed();
}
// Reset callbacks now that state is changed
resetCallbacks();
// Pass the pointers to a thread, allowing to terminate a transport from its own thread
auto ws = std::atomic_exchange(&mWsTransport, decltype(mWsTransport)(nullptr));
auto tls = std::atomic_exchange(&mTlsTransport, decltype(mTlsTransport)(nullptr));
auto tcp = std::atomic_exchange(&mTcpTransport, decltype(mTcpTransport)(nullptr));
ThreadPool::Instance().enqueue([ws, tls, tcp]() mutable {
if (ws)
ws->stop();
if (tls)
tls->stop();
if (tcp)
tcp->stop();
ws.reset();
tls.reset();
tcp.reset();
});
}
} // namespace rtc::impl
#endif

93
src/impl/websocket.hpp Normal file
View File

@ -0,0 +1,93 @@
/**
* Copyright (c) 2020-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_IMPL_WEBSOCKET_H
#define RTC_IMPL_WEBSOCKET_H
#if RTC_ENABLE_WEBSOCKET
#include "channel.hpp"
#include "include.hpp"
#include "init.hpp"
#include "message.hpp"
#include "queue.hpp"
#include "tcptransport.hpp"
#include "tlstransport.hpp"
#include "wstransport.hpp"
#include "rtc/websocket.hpp"
#include <atomic>
#include <optional>
#include <thread>
#include <variant>
namespace rtc::impl {
struct WebSocket final : public Channel, public std::enable_shared_from_this<WebSocket> {
using State = rtc::WebSocket::State;
using Configuration = rtc::WebSocket::Configuration;
WebSocket(Configuration config_);
~WebSocket();
void parse(const string &url);
void close();
bool outgoing(message_ptr message);
void incoming(message_ptr message);
std::optional<message_variant> receive() override;
std::optional<message_variant> peek() override;
size_t availableAmount() const override;
bool isOpen() const;
bool isClosed() const;
size_t maxMessageSize() const;
bool changeState(State state);
void remoteClose();
std::shared_ptr<TcpTransport> initTcpTransport();
std::shared_ptr<TlsTransport> initTlsTransport();
std::shared_ptr<WsTransport> initWsTransport();
std::shared_ptr<TcpTransport> getTcpTransport() const;
std::shared_ptr<TlsTransport> getTlsTransport() const;
std::shared_ptr<WsTransport> getWsTransport() const;
void closeTransports();
const Configuration config;
std::atomic<State> state = State::Closed;
private:
const init_token mInitToken = Init::Token();
std::shared_ptr<TcpTransport> mTcpTransport;
std::shared_ptr<TlsTransport> mTlsTransport;
std::shared_ptr<WsTransport> mWsTransport;
string mScheme, mHost, mHostname, mService, mPath;
Queue<message_ptr> mRecvQueue;
};
} // namespace rtc::impl
#endif
#endif // RTC_IMPL_WEBSOCKET_H

2
src/wstransport.cpp → src/impl/wstransport.cpp
View File

@ -45,7 +45,7 @@
#define ntohll(x) htonll(x)
#endif
namespace rtc {
namespace rtc::impl {
using namespace std::chrono;
using std::to_integer;

8
src/wstransport.hpp → src/impl/wstransport.hpp
View File

@ -16,15 +16,15 @@
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef RTC_WS_TRANSPORT_H
#define RTC_WS_TRANSPORT_H
#ifndef RTC_IMPL_WS_TRANSPORT_H
#define RTC_IMPL_WS_TRANSPORT_H
#include "include.hpp"
#include "transport.hpp"
#if RTC_ENABLE_WEBSOCKET
namespace rtc {
namespace rtc::impl {
class TcpTransport;
class TlsTransport;
@ -81,7 +81,7 @@ private:
Opcode mPartialOpcode;
};
} // namespace rtc
} // namespace rtc::impl
#endif

40
src/init.cpp
View File

@ -18,26 +18,24 @@
#include "init.hpp"
#include "certificate.hpp"
#include "dtlstransport.hpp"
#include "sctptransport.hpp"
#include "threadpool.hpp"
#include "tls.hpp"
#include "impl/certificate.hpp"
#include "impl/dtlstransport.hpp"
#include "impl/sctptransport.hpp"
#include "impl/threadpool.hpp"
#include "impl/tls.hpp"
#if RTC_ENABLE_WEBSOCKET
#include "tlstransport.hpp"
#include "impl/tlstransport.hpp"
#endif
#if RTC_ENABLE_MEDIA
#include "dtlssrtptransport.hpp"
#include "impl/dtlssrtptransport.hpp"
#endif
#ifdef _WIN32
#include <winsock2.h>
#endif
using std::shared_ptr;
namespace rtc {
namespace {
@ -51,7 +49,7 @@ void doInit() {
throw std::runtime_error("WSAStartup failed, error=" + std::to_string(WSAGetLastError()));
#endif
ThreadPool::Instance().spawn(THREADPOOL_SIZE);
impl::ThreadPool::Instance().spawn(THREADPOOL_SIZE);
#if USE_GNUTLS
// Nothing to do
@ -59,29 +57,29 @@ void doInit() {
openssl::init();
#endif
SctpTransport::Init();
DtlsTransport::Init();
impl::SctpTransport::Init();
impl::DtlsTransport::Init();
#if RTC_ENABLE_WEBSOCKET
TlsTransport::Init();
impl::TlsTransport::Init();
#endif
#if RTC_ENABLE_MEDIA
DtlsSrtpTransport::Init();
impl::DtlsSrtpTransport::Init();
#endif
}
void doCleanup() {
PLOG_DEBUG << "Global cleanup";
ThreadPool::Instance().join();
CleanupCertificateCache();
impl::ThreadPool::Instance().join();
impl::CleanupCertificateCache();
SctpTransport::Cleanup();
DtlsTransport::Cleanup();
impl::SctpTransport::Cleanup();
impl::DtlsTransport::Cleanup();
#if RTC_ENABLE_WEBSOCKET
TlsTransport::Cleanup();
impl::TlsTransport::Cleanup();
#endif
#if RTC_ENABLE_MEDIA
DtlsSrtpTransport::Cleanup();
impl::DtlsSrtpTransport::Cleanup();
#endif
#ifdef _WIN32
@ -114,7 +112,7 @@ void Init::Preload() {
Global = new shared_ptr<void>(token);
PLOG_DEBUG << "Preloading certificate";
make_certificate().wait();
impl::make_certificate().wait();
}
void Init::Cleanup() {

1082
src/peerconnection.cpp
View File

File diff suppressed because it is too large Load Diff

11
src/rtcpreceivingsession.cpp
View File

@ -20,9 +20,10 @@
#if RTC_ENABLE_MEDIA
#include "rtcpreceivingsession.hpp"
#include "logcounter.hpp"
#include "track.hpp"
#include "impl/logcounter.hpp"
#include <cmath>
#include <utility>
@ -34,10 +35,10 @@
namespace rtc {
static LogCounter COUNTER_BAD_RTP_HEADER(plog::warning, "Number of malformed RTP headers");
static LogCounter COUNTER_UNKNOWN_PPID(plog::warning, "Number of Unknown PPID messages");
static LogCounter COUNTER_BAD_NOTIF_LEN(plog::warning, "Number of Bad-Lengthed notifications");
static LogCounter COUNTER_BAD_SCTP_STATUS(plog::warning, "Number of unknown SCTP_STATUS errors");
static impl::LogCounter COUNTER_BAD_RTP_HEADER(plog::warning, "Number of malformed RTP headers");
static impl::LogCounter COUNTER_UNKNOWN_PPID(plog::warning, "Number of Unknown PPID messages");
static impl::LogCounter COUNTER_BAD_NOTIF_LEN(plog::warning, "Number of Bad-Lengthed notifications");
static impl::LogCounter COUNTER_BAD_SCTP_STATUS(plog::warning, "Number of unknown SCTP_STATUS errors");
message_ptr RtcpReceivingSession::outgoing(message_ptr ptr) { return ptr; }

171
src/track.cpp
View File

@ -1,5 +1,5 @@
/**
* Copyright (c) 2020 Paul-Louis Ageneau
* Copyright (c) 2020-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@ -17,189 +17,50 @@
*/
#include "track.hpp"
#include "dtlssrtptransport.hpp"
#include "include.hpp"
#include "logcounter.hpp"
static rtc::LogCounter
COUNTER_MEDIA_BAD_DIRECTION(plog::warning,
"Number of media packets sent in invalid directions");
static rtc::LogCounter COUNTER_QUEUE_FULL(plog::warning,
"Number of media packets dropped due to a full queue");
#include "impl/track.hpp"
namespace rtc {
using std::shared_ptr;
using std::weak_ptr;
Track::Track(impl_ptr<impl::Track> impl)
: CheshireCat<impl::Track>(impl), Channel(std::dynamic_pointer_cast<impl::Channel>(impl)) {}
Track::Track(Description::Media description)
: mMediaDescription(std::move(description)), mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
string Track::mid() const { return impl()->mid(); }
string Track::mid() const {
std::shared_lock lock(mMutex);
return mMediaDescription.mid();
}
Description::Direction Track::direction() const { return impl()->direction(); }
Description::Media Track::description() const {
std::shared_lock lock(mMutex);
return mMediaDescription;
}
Description::Direction Track::direction() const {
std::shared_lock lock(mMutex);
return mMediaDescription.direction();
}
Description::Media Track::description() const { return impl()->description(); }
void Track::setDescription(Description::Media description) {
std::unique_lock lock(mMutex);
if (description.mid() != mMediaDescription.mid())
throw std::logic_error("Media description mid does not match track mid");
mMediaDescription = std::move(description);
impl()->setDescription(std::move(description));
}
void Track::close() {
mIsClosed = true;
void Track::close() { impl()->close(); }
setRtcpHandler(nullptr);
resetCallbacks();
}
bool Track::send(message_variant data) {
if (mIsClosed)
throw std::runtime_error("Track is closed");
auto dir = direction();
if ((dir == Description::Direction::RecvOnly || dir == Description::Direction::Inactive)) {
COUNTER_MEDIA_BAD_DIRECTION++;
return false;
}
auto message = make_message(std::move(data));
if (auto handler = getRtcpHandler()) {
message = handler->outgoing(message);
if (!message)
return false;
}
return outgoing(std::move(message));
}
bool Track::send(message_variant data) { return impl()->outgoing(make_message(std::move(data))); }
bool Track::send(const byte *data, size_t size) { return send(binary(data, data + size)); }
std::optional<message_variant> Track::receive() {
if (auto next = mRecvQueue.tryPop())
return to_variant(std::move(**next));
bool Track::isOpen(void) const { return impl()->isOpen(); }
return nullopt;
}
std::optional<message_variant> Track::peek() {
if (auto next = mRecvQueue.peek())
return to_variant(std::move(**next));
return nullopt;
}
bool Track::isOpen(void) const {
#if RTC_ENABLE_MEDIA
std::shared_lock lock(mMutex);
return !mIsClosed && mDtlsSrtpTransport.lock();
#else
return !mIsClosed;
#endif
}
bool Track::isClosed(void) const { return mIsClosed; }
bool Track::isClosed(void) const { return impl()->isClosed(); }
size_t Track::maxMessageSize() const {
// TODO
return 65535 - 12 - 4; // SRTP/UDP
}
size_t Track::availableAmount() const { return mRecvQueue.amount(); }
#if RTC_ENABLE_MEDIA
void Track::open(shared_ptr<DtlsSrtpTransport> transport) {
{
std::lock_guard lock(mMutex);
mDtlsSrtpTransport = transport;
}
triggerOpen();
}
#endif
void Track::incoming(message_ptr message) {
if (!message)
return;
auto dir = direction();
if ((dir == Description::Direction::SendOnly || dir == Description::Direction::Inactive) &&
message->type != Message::Control) {
COUNTER_MEDIA_BAD_DIRECTION++;
return;
}
if (auto handler = getRtcpHandler()) {
message = handler->incoming(message);
if (!message)
return;
}
// Tail drop if queue is full
if (mRecvQueue.full()) {
COUNTER_QUEUE_FULL++;
return;
}
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
}
bool Track::outgoing([[maybe_unused]] message_ptr message) {
#if RTC_ENABLfiE_MEDIA
std::shared_ptr<DtlsSrtpTransport> transport;
{
std::shared_lock lock(mMutex);
transport = mDtlsSrtpTransport.lock();
if (!transport)
throw std::runtime_error("Track is closed");
// Set recommended medium-priority DSCP value
// See https://tools.ietf.org/html/draft-ietf-tsvwg-rtcweb-qos-18
if (mMediaDescription.type() == "audio")
message->dscp = 46; // EF: Expedited Forwarding
else
message->dscp = 36; // AF42: Assured Forwarding class 4, medium drop probability
}
return transport->sendMedia(message);
#else
PLOG_WARNING << "Ignoring track send (not compiled with media support)";
return false;
#endif
}
void Track::setRtcpHandler(std::shared_ptr<MediaHandler> handler) {
{
std::unique_lock lock(mMutex);
mRtcpHandler = handler;
}
handler->onOutgoing(std::bind(&Track::outgoing, this, std::placeholders::_1));
impl()->setRtcpHandler(std::move(handler));
}
bool Track::requestKeyframe() {
if (auto handler = getRtcpHandler())
if (auto handler = impl()->getRtcpHandler())
return handler->requestKeyframe();
return false;
}
std::shared_ptr<MediaHandler> Track::getRtcpHandler() {
std::shared_lock lock(mMutex);
return mRtcpHandler;
}
std::shared_ptr<MediaHandler> Track::getRtcpHandler() { return impl()->getRtcpHandler(); }
} // namespace rtc

339
src/websocket.cpp
View File

@ -1,5 +1,5 @@
/**
* Copyright (c) 2020 Paul-Louis Ageneau
* Copyright (c) 2020-2021 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
@ -20,12 +20,8 @@
#include "websocket.hpp"
#include "include.hpp"
#include "threadpool.hpp"
#include "tcptransport.hpp"
#include "tlstransport.hpp"
#include "verifiedtlstransport.hpp"
#include "wstransport.hpp"
#include "impl/websocket.hpp"
#include <regex>
@ -35,335 +31,50 @@
namespace rtc {
using std::shared_ptr;
using namespace std::placeholders;
WebSocket::WebSocket(std::optional<Configuration> config)
: mConfig(config ? std::move(*config) : Configuration()),
mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {
PLOG_VERBOSE << "Creating WebSocket";
}
WebSocket::WebSocket() : WebSocket(Configuration()) {}
WebSocket::~WebSocket() {
PLOG_VERBOSE << "Destroying WebSocket";
remoteClose();
}
WebSocket::WebSocket(Configuration config)
: CheshireCat<impl::WebSocket>(std::move(config)),
Channel(std::dynamic_pointer_cast<impl::Channel>(CheshireCat<impl::WebSocket>::impl())) {}
WebSocket::State WebSocket::readyState() const { return mState; }
WebSocket::~WebSocket() { impl()->remoteClose(); }
WebSocket::State WebSocket::readyState() const { return impl()->state; }
bool WebSocket::isOpen() const { return impl()->state.load() == State::Open; }
bool WebSocket::isClosed() const { return impl()->state.load() == State::Closed; }
size_t WebSocket::maxMessageSize() const { return DEFAULT_MAX_MESSAGE_SIZE; }
void WebSocket::open(const string &url) {
PLOG_VERBOSE << "Opening WebSocket to URL: " << url;
if (mState != State::Closed)
throw std::logic_error("WebSocket must be closed before opening");
// Modified regex from RFC 3986, see https://tools.ietf.org/html/rfc3986#appendix-B
static const char *rs =
R"(^(([^:.@/?#]+):)?(/{0,2}((([^:@]*)(:([^@]*))?)@)?(([^:/?#]*)(:([^/?#]*))?))?([^?#]*)(\?([^#]*))?(#(.*))?)";
static const std::regex r(rs, std::regex::extended);
std::smatch m;
if (!std::regex_match(url, m, r) || m[10].length() == 0)
throw std::invalid_argument("Invalid WebSocket URL: " + url);
mScheme = m[2];
if (mScheme.empty())
mScheme = "ws";
else if (mScheme != "ws" && mScheme != "wss")
throw std::invalid_argument("Invalid WebSocket scheme: " + mScheme);
mHostname = m[10];
mService = m[12];
if (mService.empty()) {
mService = mScheme == "ws" ? "80" : "443";
mHost = mHostname;
} else {
mHost = mHostname + ':' + mService;
}
while (!mHostname.empty() && mHostname.front() == '[')
mHostname.erase(mHostname.begin());
while (!mHostname.empty() && mHostname.back() == ']')
mHostname.pop_back();
mPath = m[13];
if (mPath.empty())
mPath += '/';
if (string query = m[15]; !query.empty())
mPath += "?" + query;
changeState(State::Connecting);
initTcpTransport();
impl()->parse(url);
impl()->changeState(State::Connecting);
impl()->initTcpTransport();
}
void WebSocket::close() {
auto state = mState.load();
auto state = impl()->state.load();
if (state == State::Connecting || state == State::Open) {
PLOG_VERBOSE << "Closing WebSocket";
changeState(State::Closing);
if (auto transport = std::atomic_load(&mWsTransport))
impl()->changeState(State::Closing);
if (auto transport = impl()->getWsTransport())
transport->close();
else
changeState(State::Closed);
impl()->changeState(State::Closed);
}
}
void WebSocket::remoteClose() {
if (mState.load() != State::Closed) {
close();
closeTransports();
}
bool WebSocket::send(message_variant data) {
return impl()->outgoing(make_message(std::move(data)));
}
bool WebSocket::send(message_variant data) { return outgoing(make_message(std::move(data))); }
bool WebSocket::send(const byte *data, size_t size) {
return outgoing(make_message(data, data + size));
}
bool WebSocket::isOpen() const { return mState == State::Open; }
bool WebSocket::isClosed() const { return mState == State::Closed; }
size_t WebSocket::maxMessageSize() const { return DEFAULT_MAX_MESSAGE_SIZE; }
std::optional<message_variant> WebSocket::receive() {
while (auto next = mRecvQueue.tryPop()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
}
return nullopt;
}
std::optional<message_variant> WebSocket::peek() {
while (auto next = mRecvQueue.peek()) {
message_ptr message = *next;
if (message->type != Message::Control)
return to_variant(std::move(*message));
mRecvQueue.tryPop();
}
return nullopt;
}
size_t WebSocket::availableAmount() const { return mRecvQueue.amount(); }
bool WebSocket::changeState(State state) { return mState.exchange(state) != state; }
bool WebSocket::outgoing(message_ptr message) {
if (mState != State::Open || !mWsTransport)
throw std::runtime_error("WebSocket is not open");
if (message->size() > maxMessageSize())
throw std::runtime_error("Message size exceeds limit");
return mWsTransport->send(message);
}
void WebSocket::incoming(message_ptr message) {
if (!message) {
remoteClose();
return;
}
if (message->type == Message::String || message->type == Message::Binary) {
mRecvQueue.push(message);
triggerAvailable(mRecvQueue.size());
}
}
shared_ptr<TcpTransport> WebSocket::initTcpTransport() {
PLOG_VERBOSE << "Starting TCP transport";
using State = TcpTransport::State;
try {
std::lock_guard lock(mInitMutex);
if (auto transport = std::atomic_load(&mTcpTransport))
return transport;
auto transport = std::make_shared<TcpTransport>(
mHostname, mService, [this, weak_this = weak_from_this()](State state) {
auto shared_this = weak_this.lock();
if (!shared_this)
return;
switch (state) {
case State::Connected:
if (mScheme == "ws")
initWsTransport();
else
initTlsTransport();
break;
case State::Failed:
triggerError("TCP connection failed");
remoteClose();
break;
case State::Disconnected:
remoteClose();
break;
default:
// Ignore
break;
}
});
std::atomic_store(&mTcpTransport, transport);
if (mState == WebSocket::State::Closed) {
mTcpTransport.reset();
throw std::runtime_error("Connection is closed");
}
transport->start();
return transport;
} catch (const std::exception &e) {
PLOG_ERROR << e.what();
remoteClose();
throw std::runtime_error("TCP transport initialization failed");
}
}
shared_ptr<TlsTransport> WebSocket::initTlsTransport() {
PLOG_VERBOSE << "Starting TLS transport";
using State = TlsTransport::State;
try {
std::lock_guard lock(mInitMutex);
if (auto transport = std::atomic_load(&mTlsTransport))
return transport;
auto lower = std::atomic_load(&mTcpTransport);
auto stateChangeCallback = [this, weak_this = weak_from_this()](State state) {
auto shared_this = weak_this.lock();
if (!shared_this)
return;
switch (state) {
case State::Connected:
initWsTransport();
break;
case State::Failed:
triggerError("TCP connection failed");
remoteClose();
break;
case State::Disconnected:
remoteClose();
break;
default:
// Ignore
break;
}
};
shared_ptr<TlsTransport> transport;
#ifdef _WIN32
if (!mConfig.disableTlsVerification) {
PLOG_WARNING << "TLS certificate verification with root CA is not supported on Windows";
}
transport = std::make_shared<TlsTransport>(lower, mHostname, stateChangeCallback);
#else
if (mConfig.disableTlsVerification)
transport = std::make_shared<TlsTransport>(lower, mHostname, stateChangeCallback);
else
transport =
std::make_shared<VerifiedTlsTransport>(lower, mHostname, stateChangeCallback);
#endif
std::atomic_store(&mTlsTransport, transport);
if (mState == WebSocket::State::Closed) {
mTlsTransport.reset();
throw std::runtime_error("Connection is closed");
}
transport->start();
return transport;
} catch (const std::exception &e) {
PLOG_ERROR << e.what();
remoteClose();
throw std::runtime_error("TLS transport initialization failed");
}
}
shared_ptr<WsTransport> WebSocket::initWsTransport() {
PLOG_VERBOSE << "Starting WebSocket transport";
using State = WsTransport::State;
try {
std::lock_guard lock(mInitMutex);
if (auto transport = std::atomic_load(&mWsTransport))
return transport;
shared_ptr<Transport> lower = std::atomic_load(&mTlsTransport);
if (!lower)
lower = std::atomic_load(&mTcpTransport);
WsTransport::Configuration wsConfig = {};
wsConfig.host = mHost;
wsConfig.path = mPath;
wsConfig.protocols = mConfig.protocols;
auto transport = std::make_shared<WsTransport>(
lower, wsConfig, weak_bind(&WebSocket::incoming, this, _1),
[this, weak_this = weak_from_this()](State state) {
auto shared_this = weak_this.lock();
if (!shared_this)
return;
switch (state) {
case State::Connected:
if (mState == WebSocket::State::Connecting) {
PLOG_DEBUG << "WebSocket open";
changeState(WebSocket::State::Open);
triggerOpen();
}
break;
case State::Failed:
triggerError("WebSocket connection failed");
remoteClose();
break;
case State::Disconnected:
remoteClose();
break;
default:
// Ignore
break;
}
});
std::atomic_store(&mWsTransport, transport);
if (mState == WebSocket::State::Closed) {
mWsTransport.reset();
throw std::runtime_error("Connection is closed");
}
transport->start();
return transport;
} catch (const std::exception &e) {
PLOG_ERROR << e.what();
remoteClose();
throw std::runtime_error("WebSocket transport initialization failed");
}
}
void WebSocket::closeTransports() {
PLOG_VERBOSE << "Closing transports";
if (mState.load() != State::Closed) {
changeState(State::Closed);
triggerClosed();
}
// Reset callbacks now that state is changed
resetCallbacks();
// Pass the pointers to a thread, allowing to terminate a transport from its own thread
auto ws = std::atomic_exchange(&mWsTransport, decltype(mWsTransport)(nullptr));
auto tls = std::atomic_exchange(&mTlsTransport, decltype(mTlsTransport)(nullptr));
auto tcp = std::atomic_exchange(&mTcpTransport, decltype(mTcpTransport)(nullptr));
ThreadPool::Instance().enqueue([ws, tls, tcp]() mutable {
if (ws)
ws->stop();
if (tls)
tls->stop();
if (tcp)
tcp->stop();
ws.reset();
tls.reset();
tcp.reset();
});
return impl()->outgoing(make_message(data, data + size));
}
} // namespace rtc