Added .travis.yml

Manually handle OpenSSL handshake timeout

.travis.yml

@@ -0,0 +1,4 @@
+os: osx
+osx_image: xcode11.3
+language: cpp
+script: cmake -B build -DUSE_JUICE=1 -DUSE_GNUTLS=1 && cd build && make && ./tests

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.7)
 project (libdatachannel
 	DESCRIPTION "WebRTC DataChannels Library"
-	VERSION 0.4.8
+	VERSION 0.4.9
 	LANGUAGES CXX)
 
 option(USE_GNUTLS "Use GnuTLS instead of OpenSSL" OFF)
@@ -17,6 +17,7 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 set(CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/cmake/Modules)
 
 if(WIN32)
+	add_definitions(-DWIN32_LEAN_AND_MEAN)
 	if (MSYS OR MINGW)
 		add_definitions(-DSCTP_STDINT_INCLUDE=<stdint.h>)
 	endif()
@@ -31,11 +32,31 @@ set(LIBDATACHANNEL_SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/src/description.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/peerconnection.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/rtc.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/sctptransport.cpp
 )
 
+set(LIBDATACHANNEL_HEADERS
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/candidate.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/channel.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/configuration.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/configuration.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/datachannel.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/description.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/include.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/init.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/log.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/message.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/peerconnection.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/queue.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/reliability.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtc.h
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtc.hpp
+)
+
 set(TESTS_SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/test/main.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/test/connectivity.cpp
@@ -54,6 +75,10 @@ set(THREADS_PREFER_PTHREAD_FLAG ON)
 find_package(Threads REQUIRED)
 
 add_subdirectory(deps/usrsctp EXCLUDE_FROM_ALL)
+if (CMAKE_CXX_COMPILER_ID MATCHES "GNU")
+    target_compile_options(usrsctp PRIVATE -Wno-error=format-truncation)
+	target_compile_options(usrsctp-static PRIVATE -Wno-error=format-truncation)
+endif()
 add_library(Usrsctp::Usrsctp ALIAS usrsctp)
 add_library(Usrsctp::UsrsctpStatic ALIAS usrsctp-static)
 
@@ -123,6 +148,9 @@ endif()
 add_library(LibDataChannel::LibDataChannel ALIAS datachannel)
 add_library(LibDataChannel::LibDataChannelStatic ALIAS datachannel-static)
 
+install(TARGETS datachannel LIBRARY DESTINATION lib)
+install(FILES ${LIBDATACHANNEL_HEADERS} DESTINATION include/rtc)
+
 # Main Test
 add_executable(datachannel-tests ${TESTS_SOURCES})
 set_target_properties(datachannel-tests PROPERTIES

Makefile

@@ -5,7 +5,7 @@ CXX=$(CROSS)g++
 AR=$(CROSS)ar
 RM=rm -f
 CXXFLAGS=-std=c++17
-CPPFLAGS=-O2 -pthread -fPIC -Wall -Wno-address-of-packed-member
+CPPFLAGS=-O2 -pthread -fPIC -Wall
 LDFLAGS=-pthread
 LIBS=
 LOCALLIBS=libusrsctp.a
@@ -86,7 +86,7 @@ dist-clean: clean
 libusrsctp.a:
 	cd $(USRSCTP_DIR) && \
 		./bootstrap && \
-		./configure --enable-static --disable-debug CFLAGS="$(CPPFLAGS)" && \
+		./configure --enable-static --disable-debug CFLAGS="$(CPPFLAGS) -Wno-error=format-truncation" && \
 		make
 	cp $(USRSCTP_DIR)/usrsctplib/.libs/libusrsctp.a .
 

include/rtc/candidate.hpp

@@ -25,6 +25,15 @@
 
 namespace rtc {
 
+enum class CandidateType { Host = 0, ServerReflexive, PeerReflexive, Relayed };
+enum class CandidateTransportType { Udp = 0, TcpActive, TcpPassive, TcpSo };
+struct CandidateInfo {
+	string address;
+	int port;
+	CandidateType type;
+	CandidateTransportType transportType;
+};
+
 class Candidate {
 public:
 	Candidate(string candidate, string mid = "");
@@ -46,6 +55,8 @@ private:
 } // namespace rtc
 
 std::ostream &operator<<(std::ostream &out, const rtc::Candidate &candidate);
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateType &type);
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateTransportType &transportType);
 
 #endif
 

include/rtc/channel.hpp

@@ -60,6 +60,8 @@ protected:
 	virtual void triggerAvailable(size_t count);
 	virtual void triggerBufferedAmount(size_t amount);
 
+	void resetCallbacks();
+
 private:
 	synchronized_callback<> mOpenCallback;
 	synchronized_callback<> mClosedCallback;

include/rtc/configuration.hpp

@@ -51,8 +51,22 @@ struct IceServer {
 	RelayType relayType;
 };
 
+struct ProxyServer {
+	enum class Type { None = 0, Socks5, Http, Last = Http };
+
+	ProxyServer(Type type_, string ip_, uint16_t port_, string username_ = "",
+	            string password_ = "");
+
+	Type type;
+	string ip;
+	uint16_t port;
+	string username;
+	string password;
+};
+
 struct Configuration {
 	std::vector<IceServer> iceServers;
+	std::optional<ProxyServer> proxyServer;
 	bool enableIceTcp = false;
 	uint16_t portRangeBegin = 1024;
 	uint16_t portRangeEnd = 65535;

include/rtc/datachannel.hpp

@@ -38,9 +38,9 @@ class PeerConnection;
 
 class DataChannel : public std::enable_shared_from_this<DataChannel>, public Channel {
 public:
-	DataChannel(std::shared_ptr<PeerConnection> pc, unsigned int stream, string label,
+	DataChannel(std::weak_ptr<PeerConnection> pc, unsigned int stream, string label,
 	            string protocol, Reliability reliability);
-	DataChannel(std::shared_ptr<PeerConnection> pc, std::shared_ptr<SctpTransport> transport,
+	DataChannel(std::weak_ptr<PeerConnection> pc, std::weak_ptr<SctpTransport> transport,
 	            unsigned int stream);
 	~DataChannel();
 
@@ -65,13 +65,13 @@ public:
 
 private:
 	void remoteClose();
-	void open(std::shared_ptr<SctpTransport> sctpTransport);
+	void open(std::shared_ptr<SctpTransport> transport);
 	bool outgoing(mutable_message_ptr message);
 	void incoming(message_ptr message);
 	void processOpenMessage(message_ptr message);
 
-	const std::shared_ptr<PeerConnection> mPeerConnection;
+	const std::weak_ptr<PeerConnection> mPeerConnection;
-	std::shared_ptr<SctpTransport> mSctpTransport;
+	std::weak_ptr<SctpTransport> mSctpTransport;
 
 	unsigned int mStream;
 	string mLabel;

include/rtc/include.hpp

@@ -20,7 +20,6 @@
 #define RTC_INCLUDE_H
 
 #ifdef _WIN32
-#define WIN32_LEAN_AND_MEAN
 #ifndef _WIN32_WINNT
 #define _WIN32_WINNT 0x0602
 #endif

include/rtc/init.hpp

@@ -0,0 +1,50 @@
+/**
+ * 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
+ */
+
+#ifndef RTC_INIT_H
+#define RTC_INIT_H
+
+#include "include.hpp"
+
+#include <mutex>
+
+namespace rtc {
+
+class Init;
+using init_token = std::shared_ptr<Init>;
+
+class Init {
+public:
+	static init_token Token();
+	static void Cleanup();
+
+	~Init();
+
+private:
+	Init();
+
+	static std::weak_ptr<Init> Weak;
+	static init_token Global;
+	static std::mutex Mutex;
+};
+
+inline void Cleanup() { Init::Cleanup(); }
+
+} // namespace rtc
+
+#endif

include/rtc/log.hpp

@@ -19,9 +19,7 @@
 #ifndef RTC_LOG_H
 #define RTC_LOG_H
 
-#include "plog/Appenders/ColorConsoleAppender.h"
 #include "plog/Log.h"
-#include "plog/Logger.h"
 
 namespace rtc {
 
@@ -35,21 +33,8 @@ enum class LogLevel { // Don't change, it must match plog severity
 	Verbose = 6
 };
 
-inline void InitLogger(plog::Severity severity, plog::IAppender *appender = nullptr) {
+void InitLogger(LogLevel level);
-	static plog::ColorConsoleAppender<plog::TxtFormatter> consoleAppender;
+void InitLogger(plog::Severity severity, plog::IAppender *appender = nullptr);
-	static plog::Logger<0> *logger = nullptr;
-	if (!logger) {
-		logger = &plog::init(severity, appender ? appender : &consoleAppender);
-		PLOG_DEBUG << "Logger initialized";
-	} else {
-		logger->setMaxSeverity(severity);
-		if (appender)
-			logger->addAppender(appender);
-	}
-}
-
-inline void InitLogger(LogLevel level) { InitLogger(static_cast<plog::Severity>(level)); }
-
 }
 
 #endif

include/rtc/peerconnection.hpp

@@ -24,6 +24,7 @@
 #include "datachannel.hpp"
 #include "description.hpp"
 #include "include.hpp"
+#include "init.hpp"
 #include "message.hpp"
 #include "reliability.hpp"
 #include "rtc.hpp"
@@ -51,14 +52,13 @@ public:
 		Connected = RTC_CONNECTED,
 		Disconnected = RTC_DISCONNECTED,
 		Failed = RTC_FAILED,
-		Closed = RTC_CLOSED,
+		Closed = RTC_CLOSED
-		Destroying = RTC_DESTROYING
 	};
 
 	enum class GatheringState : int {
 		New = RTC_GATHERING_NEW,
 		InProgress = RTC_GATHERING_INPROGRESS,
-		Complete = RTC_GATHERING_COMPLETE,
+		Complete = RTC_GATHERING_COMPLETE
 	};
 
 	PeerConnection(void);
@@ -87,10 +87,21 @@ public:
 	void onStateChange(std::function<void(State state)> callback);
 	void onGatheringStateChange(std::function<void(GatheringState state)> callback);
 
+	bool getSelectedCandidatePair(CandidateInfo *local, CandidateInfo *remote);
+
+	// Stats
+	void clearStats();
+	size_t bytesSent();
+	size_t bytesReceived();
+	std::optional<std::chrono::milliseconds> rtt();
+
 private:
+	init_token mInitToken = Init::Token();
+
 	std::shared_ptr<IceTransport> initIceTransport(Description::Role role);
 	std::shared_ptr<DtlsTransport> initDtlsTransport();
 	std::shared_ptr<SctpTransport> initSctpTransport();
+	void closeTransports();
 
 	void endLocalCandidates();
 	bool checkFingerprint(const std::string &fingerprint) const;
@@ -109,8 +120,10 @@ private:
 	void processLocalDescription(Description description);
 	void processLocalCandidate(Candidate candidate);
 	void triggerDataChannel(std::weak_ptr<DataChannel> weakDataChannel);
-	void changeState(State state);
+	bool changeState(State state);
-	void changeGatheringState(GatheringState state);
+	bool changeGatheringState(GatheringState state);
+
+	void resetCallbacks();
 
 	const Configuration mConfig;
 	const std::shared_ptr<Certificate> mCertificate;
@@ -121,7 +134,6 @@ private:
 	std::shared_ptr<IceTransport> mIceTransport;
 	std::shared_ptr<DtlsTransport> mDtlsTransport;
 	std::shared_ptr<SctpTransport> mSctpTransport;
-	std::recursive_mutex mInitMutex;
 
 	std::unordered_map<unsigned int, std::weak_ptr<DataChannel>> mDataChannels;
 	std::shared_mutex mDataChannelsMutex;

include/rtc/queue.hpp

@@ -41,12 +41,10 @@ public:
 	bool empty() const;
 	size_t size() const;   // elements
 	size_t amount() const; // amount
-	void push(const T &element);
+	void push(T element);
-	void push(T &&element);
 	std::optional<T> pop();
 	std::optional<T> peek();
-	void wait();
+	bool wait(const std::optional<std::chrono::milliseconds> &duration = nullopt);
-	void wait(const std::chrono::milliseconds &duration);
 
 private:
 	const size_t mLimit;
@@ -88,9 +86,7 @@ template <typename T> size_t Queue<T>::amount() const {
 	return mAmount;
 }
 
-template <typename T> void Queue<T>::push(const T &element) { push(T{element}); }
+template <typename T> void Queue<T>::push(T element) {
-
-template <typename T> void Queue<T>::push(T &&element) {
 	std::unique_lock lock(mMutex);
 	mPushCondition.wait(lock, [this]() { return !mLimit || mQueue.size() < mLimit || mStopping; });
 	if (!mStopping) {
@@ -122,14 +118,14 @@ template <typename T> std::optional<T> Queue<T>::peek() {
 	}
 }
 
-template <typename T> void Queue<T>::wait() {
+template <typename T>
+bool Queue<T>::wait(const std::optional<std::chrono::milliseconds> &duration) {
 	std::unique_lock lock(mMutex);
-	mPopCondition.wait(lock, [this]() { return !mQueue.empty() || mStopping; });
+	if (duration)
-}
+		mPopCondition.wait_for(lock, *duration, [this]() { return !mQueue.empty() || mStopping; });
-
+	else
-template <typename T> void Queue<T>::wait(const std::chrono::milliseconds &duration) {
+		mPopCondition.wait(lock, [this]() { return !mQueue.empty() || mStopping; });
-	std::unique_lock lock(mMutex);
+	return !mStopping;
-	mPopCondition.wait_for(lock, duration, [this]() { return !mQueue.empty() || mStopping; });
 }
 
 } // namespace rtc

include/rtc/rtc.h

@@ -23,6 +23,8 @@
 extern "C" {
 #endif
 
+#include <stdint.h>
+
 // libdatachannel C API
 
 typedef enum {
@@ -31,8 +33,7 @@ typedef enum {
 	RTC_CONNECTED = 2,
 	RTC_DISCONNECTED = 3,
 	RTC_FAILED = 4,
-	RTC_CLOSED = 5,
+	RTC_CLOSED = 5
-	RTC_DESTROYING = 6 // internal
 } rtcState;
 
 typedef enum {
@@ -55,6 +56,8 @@ typedef enum {
 typedef struct {
 	const char **iceServers;
 	int iceServersCount;
+	uint16_t portRangeBegin;
+	uint16_t portRangeEnd;
 } rtcConfiguration;
 
 typedef void (*dataChannelCallbackFunc)(int dc, void *ptr);

include/rtc/rtc.hpp

@@ -17,8 +17,11 @@
  */
 
 // C++ API
-#include "datachannel.hpp"
+#include "include.hpp"
+#include "init.hpp" // for rtc::Cleanup()
 #include "log.hpp"
+//
+#include "datachannel.hpp"
 #include "peerconnection.hpp"
 
 // C API

src/candidate.cpp

@@ -131,3 +131,33 @@ Candidate::operator string() const {
 std::ostream &operator<<(std::ostream &out, const rtc::Candidate &candidate) {
 	return out << std::string(candidate);
 }
+
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateType &type) {
+	switch (type) {
+	case rtc::CandidateType::Host:
+		return out << "Host";
+	case rtc::CandidateType::PeerReflexive:
+		return out << "PeerReflexive";
+	case rtc::CandidateType::Relayed:
+		return out << "Relayed";
+	case rtc::CandidateType::ServerReflexive:
+		return out << "ServerReflexive";
+	default:
+		return out << "Unknown";
+	}
+}
+
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateTransportType &transportType) {
+	switch (transportType) {
+	case rtc::CandidateTransportType::TcpActive:
+		return out << "TcpActive";
+	case rtc::CandidateTransportType::TcpPassive:
+		return out << "TcpPassive";
+	case rtc::CandidateTransportType::TcpSo:
+		return out << "TcpSo";
+	case rtc::CandidateTransportType::Udp:
+		return out << "Udp";
+	default:
+		return out << "Unknown";
+	}
+}

src/certificate.cpp

@@ -245,13 +245,6 @@ shared_ptr<Certificate> make_certificate(const string &commonName) {
 	if (auto it = cache.find(commonName); it != cache.end())
 		return it->second;
 
-	if (cache.empty()) {
-		// This is the first call to OpenSSL
-		OPENSSL_init_ssl(0, NULL);
-		SSL_load_error_strings();
-		ERR_load_crypto_strings();
-	}
-
 	shared_ptr<X509> x509(X509_new(), X509_free);
 	shared_ptr<EVP_PKEY> pkey(EVP_PKEY_new(), EVP_PKEY_free);
 

src/channel.cpp

@@ -88,5 +88,14 @@ void Channel::triggerBufferedAmount(size_t amount) {
 		mBufferedAmountLowCallback();
 }
 
+void Channel::resetCallbacks() {
+	mOpenCallback = nullptr;
+	mClosedCallback = nullptr;
+	mErrorCallback = nullptr;
+	mMessageCallback = nullptr;
+	mAvailableCallback = nullptr;
+	mBufferedAmountLowCallback = nullptr;
+}
+
 } // namespace rtc
 

src/configuration.cpp

@@ -38,16 +38,17 @@ IceServer::IceServer(const string &url) {
 	});
 
 	string scheme = opt[2].value_or("stun");
+	relayType = RelayType::TurnUdp;
 	if (scheme == "stun" || scheme == "STUN")
 		type = Type::Stun;
 	else if (scheme == "turn" || scheme == "TURN")
 		type = Type::Turn;
-	else if (scheme == "turns" || scheme == "TURNS")
+	else if (scheme == "turns" || scheme == "TURNS") {
 		type = Type::Turn;
-	else
+		relayType = RelayType::TurnTls;
+	} else
 		throw std::invalid_argument("Unknown ICE server protocol: " + scheme);
 
-	relayType = RelayType::TurnUdp;
 	if (auto &query = opt[15]) {
 		if (query->find("transport=udp") != string::npos)
 			relayType = RelayType::TurnUdp;
@@ -84,4 +85,7 @@ IceServer::IceServer(string hostname_, string service_, string username_, string
     : hostname(std::move(hostname_)), service(std::move(service_)), type(Type::Turn),
       username(std::move(username_)), password(std::move(password_)), relayType(relayType_) {}
 
+ProxyServer::ProxyServer(Type type_, string ip_, uint16_t port_, string username_, string password_)
+    : type(type_), ip(ip_), port(port_), username(username_), password(password_) {}
+
 } // namespace rtc

src/datachannel.cpp

@@ -30,6 +30,7 @@
 namespace rtc {
 
 using std::shared_ptr;
+using std::weak_ptr;
 
 // Messages for the DataChannel establishment protocol
 // See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09
@@ -66,16 +67,16 @@ struct CloseMessage {
 
 const size_t RECV_QUEUE_LIMIT = 1024 * 1024; // 1 MiB
 
-DataChannel::DataChannel(shared_ptr<PeerConnection> pc, unsigned int stream, string label,
+DataChannel::DataChannel(weak_ptr<PeerConnection> pc, unsigned int stream, string label,
                          string protocol, Reliability reliability)
-    : mPeerConnection(std::move(pc)), mStream(stream), mLabel(std::move(label)),
+    : 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(shared_ptr<PeerConnection> pc, shared_ptr<SctpTransport> transport,
+DataChannel::DataChannel(weak_ptr<PeerConnection> pc, weak_ptr<SctpTransport> transport,
                          unsigned int stream)
-    : mPeerConnection(std::move(pc)), mSctpTransport(transport), mStream(stream),
+    : mPeerConnection(pc), mSctpTransport(transport), mStream(stream),
       mReliability(std::make_shared<Reliability>()),
       mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
 
@@ -92,10 +93,13 @@ string DataChannel::protocol() const { return mProtocol; }
 Reliability DataChannel::reliability() const { return *mReliability; }
 
 void DataChannel::close() {
-	if (mIsOpen.exchange(false) && mSctpTransport)
+	if (mIsOpen.exchange(false))
-		mSctpTransport->reset(mStream);
+		if (auto transport = mSctpTransport.lock())
+			transport->reset(mStream);
 	mIsClosed = true;
 	mSctpTransport.reset();
+
+	resetCallbacks();
 }
 
 void DataChannel::remoteClose() {
@@ -147,17 +151,18 @@ bool DataChannel::isClosed(void) const { return mIsClosed; }
 
 size_t DataChannel::maxMessageSize() const {
 	size_t max = DEFAULT_MAX_MESSAGE_SIZE;
-	if (auto description = mPeerConnection->remoteDescription())
+	if (auto pc = mPeerConnection.lock())
-		if (auto maxMessageSize = description->maxMessageSize())
+		if (auto description = pc->remoteDescription())
-			return *maxMessageSize > 0 ? *maxMessageSize : LOCAL_MAX_MESSAGE_SIZE;
+			if (auto maxMessageSize = description->maxMessageSize())
+				return *maxMessageSize > 0 ? *maxMessageSize : LOCAL_MAX_MESSAGE_SIZE;
 
 	return std::min(max, LOCAL_MAX_MESSAGE_SIZE);
 }
 
 size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
 
-void DataChannel::open(shared_ptr<SctpTransport> sctpTransport) {
+void DataChannel::open(shared_ptr<SctpTransport> transport) {
-	mSctpTransport = sctpTransport;
+	mSctpTransport = transport;
 
 	uint8_t channelType = static_cast<uint8_t>(mReliability->type);
 	if (mReliability->unordered)
@@ -184,20 +189,24 @@ void DataChannel::open(shared_ptr<SctpTransport> sctpTransport) {
 	std::copy(mLabel.begin(), mLabel.end(), end);
 	std::copy(mProtocol.begin(), mProtocol.end(), end + mLabel.size());
 
-	mSctpTransport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
+	transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
 }
 
 bool DataChannel::outgoing(mutable_message_ptr message) {
-	if (mIsClosed || !mSctpTransport)
+	if (mIsClosed)
 		throw std::runtime_error("DataChannel is closed");
 
 	if (message->size() > maxMessageSize())
 		throw std::runtime_error("Message size exceeds limit");
 
+	auto transport = mSctpTransport.lock();
+	if (!transport)
+		throw std::runtime_error("DataChannel has no transport");
+
 	// Before the ACK has been received on a DataChannel, all messages must be sent ordered
 	message->reliability = mIsOpen ? mReliability : nullptr;
 	message->stream = mStream;
-	return mSctpTransport->send(message);
+	return transport->send(message);
 }
 
 void DataChannel::incoming(message_ptr message) {
@@ -236,6 +245,10 @@ void DataChannel::incoming(message_ptr message) {
 }
 
 void DataChannel::processOpenMessage(message_ptr message) {
+	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");
 
@@ -272,7 +285,7 @@ void DataChannel::processOpenMessage(message_ptr message) {
 	auto &ack = *reinterpret_cast<AckMessage *>(buffer.data());
 	ack.type = MESSAGE_ACK;
 
-	mSctpTransport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
+	transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
 
 	mIsOpen = true;
 	triggerOpen();

src/dtlstransport.cpp

@@ -55,6 +55,14 @@ static bool check_gnutls(int ret, const string &message = "GnuTLS error") {
 
 namespace rtc {
 
+void DtlsTransport::Init() {
+	// Nothing to do
+}
+
+void DtlsTransport::Cleanup() {
+	// Nothing to do
+}
+
 DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certificate> certificate,
                              verifier_callback verifierCallback,
                              state_callback stateChangeCallback)
@@ -64,8 +72,6 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 
 	PLOG_DEBUG << "Initializing DTLS transport (GnuTLS)";
 
-	gnutls_certificate_set_verify_function(mCertificate->credentials(), CertificateCallback);
-
 	bool active = lower->role() == Description::Role::Active;
 	unsigned int flags = GNUTLS_DATAGRAM | (active ? GNUTLS_CLIENT : GNUTLS_SERVER);
 	check_gnutls(gnutls_init(&mSession, flags));
@@ -78,12 +84,14 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 	check_gnutls(gnutls_priority_set_direct(mSession, priorities, &err_pos),
 	             "Unable to set TLS priorities");
 
+	gnutls_certificate_set_verify_function(mCertificate->credentials(), CertificateCallback);
 	check_gnutls(
 	    gnutls_credentials_set(mSession, GNUTLS_CRD_CERTIFICATE, mCertificate->credentials()));
 
-	gnutls_dtls_set_mtu(mSession, 1280 - 40 - 8); // min MTU over UDP/IPv6 (only for handshake)
+	gnutls_dtls_set_timeouts(mSession,
-	gnutls_dtls_set_timeouts(mSession, 400, 60000);
+	                         1000,   // 1s retransmission timeout recommended by RFC 6347
-	gnutls_handshake_set_timeout(mSession, 60000);
+	                         30000); // 30s total timeout
+	gnutls_handshake_set_timeout(mSession, 30000);
 
 	gnutls_session_set_ptr(mSession, this);
 	gnutls_transport_set_ptr(mSession, this);
@@ -102,15 +110,16 @@ DtlsTransport::~DtlsTransport() {
 
 DtlsTransport::State DtlsTransport::state() const { return mState; }
 
-void DtlsTransport::stop() {
+bool DtlsTransport::stop() {
-	Transport::stop();
+	if (!Transport::stop())
+		return false;
 
-	if (mRecvThread.joinable()) {
+	PLOG_DEBUG << "Stopping DTLS recv thread";
-		PLOG_DEBUG << "Stopping DTLS recv thread";
+	mIncomingQueue.stop();
-		mIncomingQueue.stop();
+	gnutls_bye(mSession, GNUTLS_SHUT_RDWR);
-		gnutls_bye(mSession, GNUTLS_SHUT_RDWR);
+	mRecvThread.join();
-		mRecvThread.join();
+	onRecv(nullptr);
-	}
+	return true;
 }
 
 bool DtlsTransport::send(message_ptr message) {
@@ -131,10 +140,13 @@ bool DtlsTransport::send(message_ptr message) {
 }
 
 void DtlsTransport::incoming(message_ptr message) {
-	if (message)
+	if (!message) {
-		mIncomingQueue.push(message);
-	else
 		mIncomingQueue.stop();
+		return;
+	}
+
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	mIncomingQueue.push(message);
 }
 
 void DtlsTransport::changeState(State state) {
@@ -148,6 +160,7 @@ void DtlsTransport::runRecvLoop() {
 	// Handshake loop
 	try {
 		changeState(State::Connecting);
+		gnutls_dtls_set_mtu(mSession, 1280 - 40 - 8); // min MTU over UDP/IPv6
 
 		int ret;
 		do {
@@ -171,6 +184,7 @@ void DtlsTransport::runRecvLoop() {
 
 	// Receive loop
 	try {
+		PLOG_INFO << "DTLS handshake done";
 		changeState(State::Connected);
 
 		const size_t bufferSize = maxMtu;
@@ -262,10 +276,8 @@ ssize_t DtlsTransport::ReadCallback(gnutls_transport_ptr_t ptr, void *data, size
 
 int DtlsTransport::TimeoutCallback(gnutls_transport_ptr_t ptr, unsigned int ms) {
 	DtlsTransport *t = static_cast<DtlsTransport *>(ptr);
-	if (ms != GNUTLS_INDEFINITE_TIMEOUT)
+	t->mIncomingQueue.wait(ms != GNUTLS_INDEFINITE_TIMEOUT ? std::make_optional(milliseconds(ms))
-		t->mIncomingQueue.wait(milliseconds(ms));
+	                                                       : nullopt);
-	else
-		t->mIncomingQueue.wait();
 	return !t->mIncomingQueue.empty() ? 1 : 0;
 }
 
@@ -323,7 +335,7 @@ BIO_METHOD *DtlsTransport::BioMethods = NULL;
 int DtlsTransport::TransportExIndex = -1;
 std::mutex DtlsTransport::GlobalMutex;
 
-void DtlsTransport::GlobalInit() {
+void DtlsTransport::Init() {
 	std::lock_guard lock(GlobalMutex);
 	if (!BioMethods) {
 		BioMethods = BIO_meth_new(BIO_TYPE_BIO, "DTLS writer");
@@ -339,6 +351,10 @@ void DtlsTransport::GlobalInit() {
 	}
 }
 
+void DtlsTransport::Cleanup() {
+	// Nothing to do
+}
+
 DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certificate> certificate,
                              verifier_callback verifierCallback, state_callback stateChangeCallback)
     : Transport(lower), mCertificate(certificate), mState(State::Disconnected),
@@ -346,7 +362,6 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
       mStateChangeCallback(std::move(stateChangeCallback)) {
 
 	PLOG_DEBUG << "Initializing DTLS transport (OpenSSL)";
-	GlobalInit();
 
 	if (!(mCtx = SSL_CTX_new(DTLS_method())))
 		throw std::runtime_error("Unable to create SSL context");
@@ -376,7 +391,6 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 		throw std::runtime_error("Unable to create SSL instance");
 
 	SSL_set_ex_data(mSsl, TransportExIndex, this);
-	SSL_set_mtu(mSsl, 1280 - 40 - 8); // min MTU over UDP/IPv6
 
 	if (lower->role() == Description::Role::Active)
 		SSL_set_connect_state(mSsl);
@@ -405,16 +419,16 @@ DtlsTransport::~DtlsTransport() {
 	SSL_CTX_free(mCtx);
 }
 
-void DtlsTransport::stop() {
+bool DtlsTransport::stop() {
-	Transport::stop();
+	if (!Transport::stop())
+		return false;
 
-	if (mRecvThread.joinable()) {
+	PLOG_DEBUG << "Stopping DTLS recv thread";
-		PLOG_DEBUG << "Stopping DTLS recv thread";
+	mIncomingQueue.stop();
-		mIncomingQueue.stop();
+	mRecvThread.join();
-		mRecvThread.join();
+	SSL_shutdown(mSsl);
-
+	onRecv(nullptr);
-		SSL_shutdown(mSsl);
+	return true;
-	}
 }
 
 DtlsTransport::State DtlsTransport::state() const { return mState; }
@@ -432,10 +446,13 @@ bool DtlsTransport::send(message_ptr message) {
 }
 
 void DtlsTransport::incoming(message_ptr message) {
-	if (message)
+	if (!message) {
-		mIncomingQueue.push(message);
-	else
 		mIncomingQueue.stop();
+		return;
+	}
+
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	mIncomingQueue.push(message);
 }
 
 void DtlsTransport::changeState(State state) {
@@ -447,35 +464,71 @@ void DtlsTransport::runRecvLoop() {
 	const size_t maxMtu = 4096;
 	try {
 		changeState(State::Connecting);
+		SSL_set_mtu(mSsl, 1280 - 40 - 8); // min MTU over UDP/IPv6
 
-		SSL_do_handshake(mSsl);
+		// Initiate the handshake
+		int ret = SSL_do_handshake(mSsl);
+		check_openssl_ret(mSsl, ret, "Handshake failed");
 
 		const size_t bufferSize = maxMtu;
 		byte buffer[bufferSize];
-		while (auto next = mIncomingQueue.pop()) {
+		while (true) {
-			auto message = *next;
+			// Process pending messages
-			BIO_write(mInBio, message->data(), message->size());
+			while (!mIncomingQueue.empty()) {
-			int ret = SSL_read(mSsl, buffer, bufferSize);
+				auto message = *mIncomingQueue.pop();
-			if (!check_openssl_ret(mSsl, ret))
+				BIO_write(mInBio, message->data(), message->size());
-				break;
 
-			auto decrypted = ret > 0 ? make_message(buffer, buffer + ret) : nullptr;
+				if (mState == State::Connecting) {
+					// Continue the handshake
+					int ret = SSL_do_handshake(mSsl);
+					if (!check_openssl_ret(mSsl, ret, "Handshake failed"))
+						break;
 
-			if (mState == State::Connecting) {
+					if (SSL_is_init_finished(mSsl)) {
-				if (unsigned long err = ERR_get_error())
+						// RFC 8261: DTLS MUST support sending messages larger than the current path
-					throw std::runtime_error("handshake failed: " + openssl_error_string(err));
+						// MTU See https://tools.ietf.org/html/rfc8261#section-5
+						SSL_set_mtu(mSsl, maxMtu + 1);
 
-				if (SSL_is_init_finished(mSsl)) {
+						PLOG_INFO << "DTLS handshake done";
-					changeState(State::Connected);
+						changeState(State::Connected);
-
+					}
-					// RFC 8261: DTLS MUST support sending messages larger than the current path MTU
+				} else {
-					// See https://tools.ietf.org/html/rfc8261#section-5
+					int ret = SSL_read(mSsl, buffer, bufferSize);
-					SSL_set_mtu(mSsl, maxMtu + 1);
+					if (!check_openssl_ret(mSsl, ret))
+						break;
+					if (ret > 0)
+						recv(make_message(buffer, buffer + ret));
 				}
 			}
 
-			if (decrypted)
+			// No more messages pending, retransmit and rearm timeout if connecting
-				recv(decrypted);
+			std::optional<milliseconds> duration;
+			if (mState == State::Connecting) {
+				// Warning: This function breaks the usual return value convention
+				int ret = DTLSv1_handle_timeout(mSsl);
+				if (ret < 0) {
+					throw std::runtime_error("Handshake timeout"); // write BIO can't fail
+				} else if (ret > 0) {
+					LOG_VERBOSE << "OpenSSL did DTLS retransmit";
+				}
+
+				struct timeval timeout = {};
+				if (mState == State::Connecting && DTLSv1_get_timeout(mSsl, &timeout)) {
+					duration = milliseconds(timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
+					// Also handle handshake timeout manually because OpenSSL actually doesn't...
+					// OpenSSL backs off exponentially in base 2 starting from the recommended 1s
+					// so this allows for 5 retransmissions and fails after roughly 30s.
+					if (duration > 30s) {
+						throw std::runtime_error("Handshake timeout");
+					} else {
+						LOG_VERBOSE << "OpenSSL DTLS retransmit timeout is " << duration->count()
+						            << "ms";
+					}
+				}
+			}
+
+			if (!mIncomingQueue.wait(duration))
+				break; // queue is stopped
 		}
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "DTLS recv: " << e.what();
@@ -486,7 +539,7 @@ void DtlsTransport::runRecvLoop() {
 		changeState(State::Disconnected);
 		recv(nullptr);
 	} else {
-		PLOG_INFO << "DTLS handshake failed";
+		PLOG_ERROR << "DTLS handshake failed";
 		changeState(State::Failed);
 	}
 }
@@ -536,7 +589,8 @@ int DtlsTransport::BioMethodWrite(BIO *bio, const char *in, int inl) {
 	if (!transport)
 		return -1;
 	auto b = reinterpret_cast<const byte *>(in);
-	return transport->outgoing(make_message(b, b + inl)) ? inl : 0;
+	transport->outgoing(make_message(b, b + inl));
+	return inl; // can't fail
 }
 
 long DtlsTransport::BioMethodCtrl(BIO *bio, int cmd, long num, void *ptr) {

src/dtlstransport.hpp

@@ -43,6 +43,9 @@ class IceTransport;
 
 class DtlsTransport : public Transport {
 public:
+	static void Init();
+	static void Cleanup();
+
 	enum class State { Disconnected, Connecting, Connected, Failed };
 
 	using verifier_callback = std::function<bool(const std::string &fingerprint)>;
@@ -54,7 +57,7 @@ public:
 
 	State state() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if dropped
 
 private:
@@ -87,7 +90,6 @@ private:
 	static int TransportExIndex;
 	static std::mutex GlobalMutex;
 
-	static void GlobalInit();
 	static int CertificateCallback(int preverify_ok, X509_STORE_CTX *ctx);
 	static void InfoCallback(const SSL *ssl, int where, int ret);
 

src/icetransport.cpp

@@ -18,6 +18,7 @@
 
 #include "icetransport.hpp"
 #include "configuration.hpp"
+#include "transport.hpp"
 
 #include <iostream>
 #include <random>
@@ -72,7 +73,7 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 	unsigned seed = std::chrono::system_clock::now().time_since_epoch().count();
 	std::shuffle(servers.begin(), servers.end(), std::default_random_engine(seed));
 
-	// Pick a STUN server
+	// Pick a STUN server (TURN support is not implemented in libjuice yet)
 	for (auto &server : servers) {
 		if (!server.hostname.empty() && server.type == IceServer::Type::Stun) {
 			if (server.service.empty())
@@ -86,7 +87,12 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 		}
 	}
 
-	// TURN support is not implemented yet
+	// Port range
+	if (config.portRangeBegin > 1024 ||
+	    (config.portRangeEnd != 0 && config.portRangeEnd != 65535)) {
+		jconfig.local_port_range_begin = config.portRangeBegin;
+		jconfig.local_port_range_end = config.portRangeEnd;
+	}
 
 	// Create agent
 	mAgent = decltype(mAgent)(juice_create(&jconfig), juice_destroy);
@@ -96,8 +102,9 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 IceTransport::~IceTransport() { stop(); }
 
-void IceTransport::stop() {
+bool IceTransport::stop() {
-	// Nothing to do
+	onRecv(nullptr);
+	return Transport::stop();
 }
 
 Description::Role IceTransport::role() const { return mRole; }
@@ -162,7 +169,10 @@ bool IceTransport::send(message_ptr message) {
 	return outgoing(message);
 }
 
-void IceTransport::incoming(message_ptr message) { recv(message); }
+void IceTransport::incoming(message_ptr message) {
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	recv(message);
+}
 
 void IceTransport::incoming(const byte *data, int size) {
 	incoming(make_message(data, data + size));
@@ -309,6 +319,18 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 	g_object_set(G_OBJECT(mNiceAgent.get()), "upnp", FALSE, nullptr);
 	g_object_set(G_OBJECT(mNiceAgent.get()), "upnp-timeout", 200, nullptr);
 
+	// Proxy
+	if (config.proxyServer.has_value()) {
+		ProxyServer proxyServer = config.proxyServer.value();
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-type", proxyServer.type, nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-ip", proxyServer.ip.c_str(), nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-port", proxyServer.port, nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-username", proxyServer.username.c_str(),
+		             nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-password", proxyServer.password.c_str(),
+		             nullptr);
+	}
+
 	// Randomize order
 	std::vector<IceServer> servers = config.iceServers;
 	unsigned seed = std::chrono::system_clock::now().time_since_epoch().count();
@@ -423,16 +445,19 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 IceTransport::~IceTransport() { stop(); }
 
-void IceTransport::stop() {
+bool IceTransport::stop() {
 	if (mTimeoutId) {
 		g_source_remove(mTimeoutId);
 		mTimeoutId = 0;
 	}
-	if (mMainLoopThread.joinable()) {
+
-		PLOG_DEBUG << "Stopping ICE thread";
+	if (!Transport::stop())
-		g_main_loop_quit(mMainLoop.get());
+		return false;
-		mMainLoopThread.join();
+
-	}
+	PLOG_DEBUG << "Stopping ICE thread";
+	g_main_loop_quit(mMainLoop.get());
+	mMainLoopThread.join();
+	return true;
 }
 
 Description::Role IceTransport::role() const { return mRole; }
@@ -516,7 +541,10 @@ bool IceTransport::send(message_ptr message) {
 	return outgoing(message);
 }
 
-void IceTransport::incoming(message_ptr message) { recv(message); }
+void IceTransport::incoming(message_ptr message) {
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	recv(message);
+}
 
 void IceTransport::incoming(const byte *data, int size) {
 	incoming(make_message(data, data + size));
@@ -645,6 +673,58 @@ void IceTransport::LogCallback(const gchar *logDomain, GLogLevelFlags logLevel,
 	PLOG(severity) << "nice: " << message;
 }
 
+bool IceTransport::getSelectedCandidatePair(CandidateInfo *localInfo, CandidateInfo *remoteInfo) {
+	NiceCandidate *local, *remote;
+	gboolean result = nice_agent_get_selected_pair(mNiceAgent.get(), mStreamId, 1, &local, &remote);
+
+	if (!result)
+		return false;
+
+	char ipaddr[INET6_ADDRSTRLEN];
+	nice_address_to_string(&local->addr, ipaddr);
+	localInfo->address = std::string(ipaddr);
+	localInfo->port = nice_address_get_port(&local->addr);
+	localInfo->type = IceTransport::NiceTypeToCandidateType(local->type);
+	localInfo->transportType =
+	    IceTransport::NiceTransportTypeToCandidateTransportType(local->transport);
+
+	nice_address_to_string(&remote->addr, ipaddr);
+	remoteInfo->address = std::string(ipaddr);
+	remoteInfo->port = nice_address_get_port(&remote->addr);
+	remoteInfo->type = IceTransport::NiceTypeToCandidateType(remote->type);
+	remoteInfo->transportType =
+	    IceTransport::NiceTransportTypeToCandidateTransportType(remote->transport);
+
+	return true;
+}
+
+const CandidateType IceTransport::NiceTypeToCandidateType(NiceCandidateType type) {
+	switch (type) {
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_HOST:
+		return CandidateType::Host;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_PEER_REFLEXIVE:
+		return CandidateType::PeerReflexive;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_RELAYED:
+		return CandidateType::Relayed;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_SERVER_REFLEXIVE:
+		return CandidateType::ServerReflexive;
+	}
+}
+
+const CandidateTransportType
+IceTransport::NiceTransportTypeToCandidateTransportType(NiceCandidateTransport type) {
+	switch (type) {
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_TCP_ACTIVE:
+		return CandidateTransportType::TcpActive;
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_TCP_PASSIVE:
+		return CandidateTransportType::TcpPassive;
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_TCP_SO:
+		return CandidateTransportType::TcpSo;
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_UDP:
+		return CandidateTransportType::Udp;
+	}
+}
+
 } // namespace rtc
 
 #endif

src/icetransport.hpp

@@ -37,7 +37,7 @@
 #include <thread>
 
 namespace rtc {
-
+	
 class IceTransport : public Transport {
 public:
 #if USE_JUICE
@@ -56,6 +56,8 @@ public:
 		Completed = NICE_COMPONENT_STATE_READY,
 		Failed = NICE_COMPONENT_STATE_FAILED,
 	};
+
+	bool getSelectedCandidatePair(CandidateInfo *local, CandidateInfo *remote);
 #endif
 	enum class GatheringState { New = 0, InProgress = 1, Complete = 2 };
 
@@ -79,7 +81,7 @@ public:
 	std::optional<string> getLocalAddress() const;
 	std::optional<string> getRemoteAddress() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if dropped
 
 private:
@@ -133,6 +135,8 @@ private:
 	static gboolean TimeoutCallback(gpointer userData);
 	static void LogCallback(const gchar *log_domain, GLogLevelFlags log_level, const gchar *message,
 	                        gpointer user_data);
+	static const CandidateType NiceTypeToCandidateType(NiceCandidateType type);
+	static const CandidateTransportType NiceTransportTypeToCandidateTransportType(NiceCandidateTransport type);
 #endif
 };
 

src/init.cpp

@@ -0,0 +1,86 @@
+/**
+ * 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 "init.hpp"
+
+#include "dtlstransport.hpp"
+#include "sctptransport.hpp"
+
+#ifdef _WIN32
+#include <winsock2.h>
+#endif
+
+#if USE_GNUTLS
+// Nothing to do
+#else
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+#endif
+
+using std::shared_ptr;
+
+namespace rtc {
+
+std::weak_ptr<Init> Init::Weak;
+init_token Init::Global;
+std::mutex Init::Mutex;
+
+init_token Init::Token() {
+	std::lock_guard lock(Mutex);
+
+	if (!Global) {
+		if (auto token = Weak.lock())
+			Global = token;
+		else
+			Global = shared_ptr<Init>(new Init());
+	}
+	return Global;
+}
+
+void Init::Cleanup() { Global.reset(); }
+
+Init::Init() {
+#ifdef _WIN32
+	WSADATA wsaData;
+	if (WSAStartup(MAKEWORD(2, 2), &wsaData))
+		throw std::runtime_error("WSAStartup failed, error=" + std::to_string(WSAGetLastError()));
+#endif
+
+#if USE_GNUTLS
+		// Nothing to do
+#else
+	OPENSSL_init_ssl(0, NULL);
+	SSL_load_error_strings();
+	ERR_load_crypto_strings();
+#endif
+
+	DtlsTransport::Init();
+	SctpTransport::Init();
+}
+
+Init::~Init() {
+	DtlsTransport::Cleanup();
+	SctpTransport::Cleanup();
+
+#ifdef _WIN32
+	WSACleanup();
+#endif
+}
+
+} // namespace rtc
+

src/log.cpp

@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2019-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 "log.hpp"
+
+#include "plog/Appenders/ColorConsoleAppender.h"
+#include "plog/Log.h"
+#include "plog/Logger.h"
+
+namespace rtc {
+
+void InitLogger(LogLevel level) { InitLogger(static_cast<plog::Severity>(level)); }
+
+void InitLogger(plog::Severity severity, plog::IAppender *appender) {
+	static plog::ColorConsoleAppender<plog::TxtFormatter> consoleAppender;
+	static plog::Logger<0> *logger = nullptr;
+	if (!logger) {
+		logger = &plog::init(severity, appender ? appender : &consoleAppender);
+		PLOG_DEBUG << "Logger initialized";
+	} else {
+		logger->setMaxSeverity(severity);
+		if (appender)
+			logger->addAppender(appender);
+	}
+}
+}
+

src/peerconnection.cpp

@@ -24,10 +24,7 @@
 #include "sctptransport.hpp"
 
 #include <iostream>
-
+#include <thread>
-#ifdef _WIN32
-#include <winsock2.h>
-#endif
 
 namespace rtc {
 
@@ -36,43 +33,33 @@ using namespace std::placeholders;
 using std::shared_ptr;
 using std::weak_ptr;
 
-PeerConnection::PeerConnection() : PeerConnection(Configuration()) {
+template <typename F, typename T, typename... Args> auto weak_bind(F &&f, T *t, Args &&... _args) {
-#ifdef _WIN32
+	return [bound = std::bind(f, t, _args...), weak_this = t->weak_from_this()](auto &&... args) {
-	WSADATA wsaData;
+		if (auto shared_this = weak_this.lock())
-	if (WSAStartup(MAKEWORD(2, 2), &wsaData))
+			bound(args...);
-		throw std::runtime_error("WSAStartup failed, error=" + std::to_string(WSAGetLastError()));
+	};
-#endif
 }
 
+template <typename F, typename T, typename... Args>
+auto weak_bind_verifier(F &&f, T *t, Args &&... _args) {
+	return [bound = std::bind(f, t, _args...), weak_this = t->weak_from_this()](auto &&... args) {
+		if (auto shared_this = weak_this.lock())
+			return bound(args...);
+		else
+			return false;
+	};
+}
+
+PeerConnection::PeerConnection() : PeerConnection(Configuration()) {}
+
 PeerConnection::PeerConnection(const Configuration &config)
     : mConfig(config), mCertificate(make_certificate("libdatachannel")), mState(State::New) {}
 
-PeerConnection::~PeerConnection() {
+PeerConnection::~PeerConnection() { close(); }
-	changeState(State::Destroying);
-	close();
-	mSctpTransport.reset();
-	mDtlsTransport.reset();
-	mIceTransport.reset();
-
-#ifdef _WIN32
-	WSACleanup();
-#endif
-}
 
 void PeerConnection::close() {
-	// Close DataChannels
 	closeDataChannels();
-
+	closeTransports();
-	// Close Transports
-	for (int i = 0; i < 2; ++i) { // Make sure a transport wasn't spawn behind our back
-		if (auto transport = std::atomic_load(&mSctpTransport))
-			transport->stop();
-		if (auto transport = std::atomic_load(&mDtlsTransport))
-			transport->stop();
-		if (auto transport = std::atomic_load(&mIceTransport))
-			transport->stop();
-	}
-	changeState(State::Closed);
 }
 
 const Configuration *PeerConnection::config() const { return &mConfig; }
@@ -114,7 +101,7 @@ void PeerConnection::setRemoteDescription(Description description) {
 		if (!sctpTransport && iceTransport->role() == Description::Role::Active) {
 			// Since we assumed passive role during DataChannel creation, we need to shift the
 			// stream numbers by one to shift them from odd to even.
-			std::unique_lock lock(mDataChannelsMutex);
+			std::unique_lock lock(mDataChannelsMutex); // we are going to swap the container
 			decltype(mDataChannels) newDataChannels;
 			auto it = mDataChannels.begin();
 			while (it != mDataChannels.end()) {
@@ -216,13 +203,15 @@ void PeerConnection::onGatheringStateChange(std::function<void(GatheringState st
 
 shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role) {
 	try {
-		std::lock_guard lock(mInitMutex);
 		if (auto transport = std::atomic_load(&mIceTransport))
 			return transport;
 
 		auto transport = std::make_shared<IceTransport>(
-		    mConfig, role, std::bind(&PeerConnection::processLocalCandidate, this, _1),
+		    mConfig, role, weak_bind(&PeerConnection::processLocalCandidate, this, _1),
-		    [this](IceTransport::State state) {
+		    [this, weak_this = weak_from_this()](IceTransport::State state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case IceTransport::State::Connecting:
 				    changeState(State::Connecting);
@@ -241,7 +230,10 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role
 				    break;
 			    }
 		    },
-		    [this](IceTransport::GatheringState state) {
+		    [this, weak_this = weak_from_this()](IceTransport::GatheringState state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case IceTransport::GatheringState::InProgress:
 				    changeGatheringState(GatheringState::InProgress);
@@ -255,8 +247,15 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role
 				    break;
 			    }
 		    });
+
 		std::atomic_store(&mIceTransport, transport);
+		if (mState == State::Closed) {
+			mIceTransport.reset();
+			transport->stop();
+			throw std::runtime_error("Connection is closed");
+		}
 		return transport;
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
 		changeState(State::Failed);
@@ -266,14 +265,16 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role
 
 shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 	try {
-		std::lock_guard lock(mInitMutex);
 		if (auto transport = std::atomic_load(&mDtlsTransport))
 			return transport;
 
 		auto lower = std::atomic_load(&mIceTransport);
 		auto transport = std::make_shared<DtlsTransport>(
-		    lower, mCertificate, std::bind(&PeerConnection::checkFingerprint, this, _1),
+		    lower, mCertificate, weak_bind_verifier(&PeerConnection::checkFingerprint, this, _1),
-		    [this](DtlsTransport::State state) {
+		    [this, weak_this = weak_from_this()](DtlsTransport::State state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case DtlsTransport::State::Connected:
 				    initSctpTransport();
@@ -289,8 +290,15 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 				    break;
 			    }
 		    });
+
 		std::atomic_store(&mDtlsTransport, transport);
+		if (mState == State::Closed) {
+			mDtlsTransport.reset();
+			transport->stop();
+			throw std::runtime_error("Connection is closed");
+		}
 		return transport;
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
 		changeState(State::Failed);
@@ -300,16 +308,18 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 
 shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 	try {
-		std::lock_guard lock(mInitMutex);
 		if (auto transport = std::atomic_load(&mSctpTransport))
 			return transport;
 
 		uint16_t sctpPort = remoteDescription()->sctpPort().value_or(DEFAULT_SCTP_PORT);
 		auto lower = std::atomic_load(&mDtlsTransport);
 		auto transport = std::make_shared<SctpTransport>(
-		    lower, sctpPort, std::bind(&PeerConnection::forwardMessage, this, _1),
+		    lower, sctpPort, weak_bind(&PeerConnection::forwardMessage, this, _1),
-		    std::bind(&PeerConnection::forwardBufferedAmount, this, _1, _2),
+		    weak_bind(&PeerConnection::forwardBufferedAmount, this, _1, _2),
-		    [this](SctpTransport::State state) {
+		    [this, weak_this = weak_from_this()](SctpTransport::State state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case SctpTransport::State::Connected:
 				    changeState(State::Connected);
@@ -328,8 +338,15 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 				    break;
 			    }
 		    });
+
 		std::atomic_store(&mSctpTransport, transport);
+		if (mState == State::Closed) {
+			mSctpTransport.reset();
+			transport->stop();
+			throw std::runtime_error("Connection is closed");
+		}
 		return transport;
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
 		changeState(State::Failed);
@@ -337,6 +354,34 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 	}
 }
 
+void PeerConnection::closeTransports() {
+	// Change state to sink state Closed to block init methods
+	changeState(State::Closed);
+
+	// Reset callbacks now that state is changed
+	resetCallbacks();
+
+	// Pass the references to a thread, allowing to terminate a transport from its own thread
+	auto sctp = std::atomic_exchange(&mSctpTransport, decltype(mSctpTransport)(nullptr));
+	auto dtls = std::atomic_exchange(&mDtlsTransport, decltype(mDtlsTransport)(nullptr));
+	auto ice = std::atomic_exchange(&mIceTransport, decltype(mIceTransport)(nullptr));
+	if (sctp || dtls || ice) {
+		std::thread t([sctp, dtls, ice]() mutable {
+			if (sctp)
+				sctp->stop();
+			if (dtls)
+				dtls->stop();
+			if (ice)
+				ice->stop();
+
+			sctp.reset();
+			dtls.reset();
+			ice.reset();
+		});
+		t.detach();
+	}
+}
+
 void PeerConnection::endLocalCandidates() {
 	std::lock_guard lock(mLocalDescriptionMutex);
 	if (mLocalDescription)
@@ -372,7 +417,7 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		    message->stream % 2 == remoteParity) {
 			channel =
 			    std::make_shared<DataChannel>(shared_from_this(), sctpTransport, message->stream);
-			channel->onOpen(std::bind(&PeerConnection::triggerDataChannel, this,
+			channel->onOpen(weak_bind(&PeerConnection::triggerDataChannel, this,
 			                          weak_ptr<DataChannel>{channel}));
 			mDataChannels.insert(std::make_pair(message->stream, channel));
 		} else {
@@ -397,7 +442,7 @@ shared_ptr<DataChannel> PeerConnection::emplaceDataChannel(Description::Role rol
 	// The active side must use streams with even identifiers, whereas the passive side must use
 	// streams with odd identifiers.
 	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09#section-6
-	std::unique_lock lock(mDataChannelsMutex);
+	std::unique_lock lock(mDataChannelsMutex); // we are going to emplace
 	unsigned int stream = (role == Description::Role::Active) ? 0 : 1;
 	while (mDataChannels.find(stream) != mDataChannels.end()) {
 		stream += 2;
@@ -411,30 +456,41 @@ shared_ptr<DataChannel> PeerConnection::emplaceDataChannel(Description::Role rol
 }
 
 shared_ptr<DataChannel> PeerConnection::findDataChannel(uint16_t stream) {
-	std::shared_lock lock(mDataChannelsMutex);
+	std::shared_lock lock(mDataChannelsMutex); // read-only
-	shared_ptr<DataChannel> channel;
+	if (auto it = mDataChannels.find(stream); it != mDataChannels.end())
-	if (auto it = mDataChannels.find(stream); it != mDataChannels.end()) {
+		if (auto channel = it->second.lock())
-		channel = it->second.lock();
+			return channel;
-		if (!channel || channel->isClosed()) {
+
-			mDataChannels.erase(it);
+	return nullptr;
-			channel.reset();
-		}
-	}
-	return channel;
 }
 
 void PeerConnection::iterateDataChannels(
     std::function<void(shared_ptr<DataChannel> channel)> func) {
-	std::shared_lock lock(mDataChannelsMutex);
+	// Iterate
-	auto it = mDataChannels.begin();
+	{
-	while (it != mDataChannels.end()) {
+		std::shared_lock lock(mDataChannelsMutex); // read-only
-		auto channel = it->second.lock();
+		auto it = mDataChannels.begin();
-		if (!channel || channel->isClosed()) {
+		while (it != mDataChannels.end()) {
-			it = mDataChannels.erase(it);
+			auto channel = it->second.lock();
-			continue;
+			if (channel && !channel->isClosed())
+				func(channel);
+
+			++it;
+		}
+	}
+
+	// Cleanup
+	{
+		std::unique_lock lock(mDataChannelsMutex); // we are going to erase
+		auto it = mDataChannels.begin();
+		while (it != mDataChannels.end()) {
+			if (!it->second.lock()) {
+				it = mDataChannels.erase(it);
+				continue;
+			}
+
+			++it;
 		}
-		func(channel);
-		++it;
 	}
 }
 
@@ -483,21 +539,72 @@ void PeerConnection::triggerDataChannel(weak_ptr<DataChannel> weakDataChannel) {
 	mDataChannelCallback(dataChannel);
 }
 
-void PeerConnection::changeState(State state) {
+bool PeerConnection::changeState(State state) {
 	State current;
 	do {
 		current = mState.load();
-		if (current == state || current == State::Destroying)
+		if (current == state)
-			return;
+			return true;
+		if (current == State::Closed)
+			return false;
+
 	} while (!mState.compare_exchange_weak(current, state));
 
-	if (state != State::Destroying)
+	mStateChangeCallback(state);
-		mStateChangeCallback(state);
+	return true;
 }
 
-void PeerConnection::changeGatheringState(GatheringState state) {
+bool PeerConnection::changeGatheringState(GatheringState state) {
 	if (mGatheringState.exchange(state) != state)
 		mGatheringStateChangeCallback(state);
+	return true;
+}
+
+void PeerConnection::resetCallbacks() {
+	// Unregister all callbacks
+	mDataChannelCallback = nullptr;
+	mLocalDescriptionCallback = nullptr;
+	mLocalCandidateCallback = nullptr;
+	mStateChangeCallback = nullptr;
+	mGatheringStateChangeCallback = nullptr;
+}
+
+bool PeerConnection::getSelectedCandidatePair(CandidateInfo *local, CandidateInfo *remote) {
+#if not USE_JUICE
+	auto iceTransport = std::atomic_load(&mIceTransport);
+	return iceTransport->getSelectedCandidatePair(local, remote);
+#else
+	PLOG_WARNING << "getSelectedCandidatePair is not implemented for libjuice";
+	return false;
+#endif
+}
+
+void PeerConnection::clearStats() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->clearStats();
+}
+
+size_t PeerConnection::bytesSent() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->bytesSent();
+	return 0;
+}
+
+size_t PeerConnection::bytesReceived() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->bytesReceived();
+	return 0;
+}
+
+std::optional<std::chrono::milliseconds> PeerConnection::rtt() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->rtt();
+	PLOG_WARNING << "Could not load sctpTransport";
+	return std::nullopt;
 }
 
 } // namespace rtc
@@ -524,9 +631,6 @@ std::ostream &operator<<(std::ostream &out, const rtc::PeerConnection::State &st
 	case State::Closed:
 		str = "closed";
 		break;
-	case State::Destroying:
-		str = "destroying";
-		break;
 	default:
 		str = "unknown";
 		break;
@@ -553,4 +657,3 @@ std::ostream &operator<<(std::ostream &out, const rtc::PeerConnection::Gathering
 	}
 	return out << str;
 }
-

src/rtc.cpp

@@ -53,6 +53,14 @@ void *getUserPointer(int id) {
 	return it != userPointerMap.end() ? it->second : nullptr;
 }
 
+void setUserPointer(int i, void *ptr) {
+	std::lock_guard lock(mutex);
+	if (ptr)
+		userPointerMap.insert(std::make_pair(i, ptr));
+	else
+		userPointerMap.erase(i);
+}
+
 shared_ptr<PeerConnection> getPeerConnection(int id) {
 	std::lock_guard lock(mutex);
 	auto it = peerConnectionMap.find(id);
@@ -99,18 +107,18 @@ bool eraseDataChannel(int dc) {
 
 void rtcInitLogger(rtcLogLevel level) { InitLogger(static_cast<LogLevel>(level)); }
 
-void rtcSetUserPointer(int i, void *ptr) {
+void rtcSetUserPointer(int i, void *ptr) { setUserPointer(i, ptr); }
-	if (ptr)
-		userPointerMap.insert(std::make_pair(i, ptr));
-	else
-		userPointerMap.erase(i);
-}
 
 int rtcCreatePeerConnection(const rtcConfiguration *config) {
 	Configuration c;
 	for (int i = 0; i < config->iceServersCount; ++i)
 		c.iceServers.emplace_back(string(config->iceServers[i]));
 
+	if (config->portRangeBegin || config->portRangeEnd) {
+		c.portRangeBegin = config->portRangeBegin;
+		c.portRangeEnd = config->portRangeEnd;
+	}
+
 	return emplacePeerConnection(std::make_shared<PeerConnection>(c));
 }
 

src/sctptransport.cpp

@@ -49,31 +49,20 @@ using std::shared_ptr;
 
 namespace rtc {
 
-std::mutex SctpTransport::GlobalMutex;
+void SctpTransport::Init() {
-int SctpTransport::InstancesCount = 0;
+	usrsctp_init(0, &SctpTransport::WriteCallback, nullptr);
-
+	usrsctp_sysctl_set_sctp_ecn_enable(0);
-void SctpTransport::GlobalInit() {
+	usrsctp_sysctl_set_sctp_init_rtx_max_default(5);
-	std::lock_guard lock(GlobalMutex);
+	usrsctp_sysctl_set_sctp_path_rtx_max_default(5);
-	if (InstancesCount++ == 0) {
+	usrsctp_sysctl_set_sctp_assoc_rtx_max_default(5);              // single path
-		usrsctp_init(0, &SctpTransport::WriteCallback, nullptr);
+	usrsctp_sysctl_set_sctp_rto_min_default(1 * 1000);             // ms
-		usrsctp_sysctl_set_sctp_ecn_enable(0);
+	usrsctp_sysctl_set_sctp_rto_max_default(10 * 1000);            // ms
-		usrsctp_sysctl_set_sctp_init_rtx_max_default(5);
+	usrsctp_sysctl_set_sctp_rto_initial_default(1 * 1000);         // ms
-		usrsctp_sysctl_set_sctp_path_rtx_max_default(5);
+	usrsctp_sysctl_set_sctp_init_rto_max_default(10 * 1000);       // ms
-		usrsctp_sysctl_set_sctp_assoc_rtx_max_default(5);              // single path
+	usrsctp_sysctl_set_sctp_heartbeat_interval_default(10 * 1000); // ms
-		usrsctp_sysctl_set_sctp_rto_min_default(1 * 1000);             // ms
-		usrsctp_sysctl_set_sctp_rto_max_default(10 * 1000);            // ms
-		usrsctp_sysctl_set_sctp_rto_initial_default(1 * 1000);         // ms
-		usrsctp_sysctl_set_sctp_init_rto_max_default(10 * 1000);       // ms
-		usrsctp_sysctl_set_sctp_heartbeat_interval_default(10 * 1000); // ms
-	}
 }
 
-void SctpTransport::GlobalCleanup() {
+void SctpTransport::Cleanup() { usrsctp_finish(); }
-	std::lock_guard lock(GlobalMutex);
-	if (--InstancesCount == 0) {
-		usrsctp_finish();
-	}
-}
 
 SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
                              message_callback recvCallback, amount_callback bufferedAmountCallback,
@@ -84,7 +73,6 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 	onRecv(recvCallback);
 
 	PLOG_DEBUG << "Initializing SCTP transport";
-	GlobalInit();
 
 	usrsctp_register_address(this);
 	mSock = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP, &SctpTransport::RecvCallback,
@@ -156,6 +144,14 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 		throw std::runtime_error("Could not set socket option SCTP_INITMSG, errno=" +
 		                         std::to_string(errno));
 
+	// Prevent fragmented interleave of messages (i.e. level 0), see RFC 6458 8.1.20.
+	// Unless the user has set the fragmentation interleave level to 0, notifications
+	// may also be interleaved with partially delivered messages.
+	int level = 0;
+	if (usrsctp_setsockopt(mSock, IPPROTO_SCTP, SCTP_FRAGMENT_INTERLEAVE, &level, sizeof(level)))
+		throw std::runtime_error("Could not disable SCTP fragmented interleave, errno=" +
+		                         std::to_string(errno));
+
 	// The default send and receive window size of usrsctp is 256KiB, which is too small for
 	// realistic RTTs, therefore we increase it to 1MiB for better performance.
 	// See https://bugzilla.mozilla.org/show_bug.cgi?id=1051685
@@ -175,21 +171,19 @@ SctpTransport::~SctpTransport() {
 
 	usrsctp_close(mSock);
 	usrsctp_deregister_address(this);
-
-	GlobalCleanup();
 }
 
 SctpTransport::State SctpTransport::state() const { return mState; }
 
-void SctpTransport::stop() {
+bool SctpTransport::stop() {
-	Transport::stop();
+	if (!Transport::stop())
-	onRecv(nullptr);
+		return false;
 
-	if (!mShutdown.exchange(true)) {
+	mSendQueue.stop();
-		mSendQueue.stop();
+	safeFlush();
-		flush();
+	shutdown();
-		shutdown();
+	onRecv(nullptr);
-	}
+	return true;
 }
 
 void SctpTransport::connect() {
@@ -277,13 +271,15 @@ void SctpTransport::incoming(message_ptr message) {
 		mWrittenCondition.wait(lock, [&]() { return mWrittenOnce || mState != State::Connected; });
 	}
 
-	if (message) {
+	if (!message) {
-		usrsctp_conninput(this, message->data(), message->size(), 0);
-	} else {
 		PLOG_INFO << "SCTP disconnected";
 		changeState(State::Disconnected);
 		recv(nullptr);
+		return;
 	}
+
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	usrsctp_conninput(this, message->data(), message->size(), 0);
 }
 
 void SctpTransport::changeState(State state) {
@@ -368,6 +364,8 @@ bool SctpTransport::trySendMessage(message_ptr message) {
 
 	if (ret >= 0) {
 		PLOG_VERBOSE << "SCTP sent size=" << message->size();
+		if (message->type == Message::Type::Binary || message->type == Message::Type::String)
+			mBytesSent += message->size();
 		return true;
 	} else if (errno == EWOULDBLOCK || errno == EAGAIN) {
 		PLOG_VERBOSE << "SCTP sending not possible";
@@ -381,25 +379,42 @@ bool SctpTransport::trySendMessage(message_ptr message) {
 void SctpTransport::updateBufferedAmount(uint16_t streamId, long delta) {
 	// Requires mSendMutex to be locked
 	auto it = mBufferedAmount.insert(std::make_pair(streamId, 0)).first;
-	size_t amount = it->second;
+	size_t amount = size_t(std::max(long(it->second) + delta, long(0)));
-	amount = size_t(std::max(long(amount) + delta, long(0)));
 	if (amount == 0)
 		mBufferedAmount.erase(it);
+	else
+		it->second = amount;
+
 	mBufferedAmountCallback(streamId, amount);
 }
 
+bool SctpTransport::safeFlush() {
+	try {
+		flush();
+		return true;
+
+	} catch (const std::exception &e) {
+		PLOG_ERROR << "SCTP flush: " << e.what();
+		return false;
+	}
+}
+
 int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, const byte *data,
                               size_t len, struct sctp_rcvinfo info, int flags) {
 	try {
+		PLOG_VERBOSE << "Handle recv, len=" << len;
 		if (!len)
 			return -1;
+
+		// This is valid because SCTP_FRAGMENT_INTERLEAVE is set to level 0
+		// so partial messages and notifications may not be interleaved.
 		if (flags & MSG_EOR) {
 			if (!mPartialRecv.empty()) {
 				mPartialRecv.insert(mPartialRecv.end(), data, data + len);
 				data = mPartialRecv.data();
 				len = mPartialRecv.size();
 			}
-			// Message is complete, process it
+			// Message/Notification is complete, process it
 			if (flags & MSG_NOTIFICATION)
 				processNotification(reinterpret_cast<const union sctp_notification *>(data), len);
 			else
@@ -407,7 +422,7 @@ int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, co
 
 			mPartialRecv.clear();
 		} else {
-			// Message is not complete
+			// Message/Notification is not complete
 			mPartialRecv.insert(mPartialRecv.end(), data, data + len);
 		}
 	} catch (const std::exception &e) {
@@ -418,24 +433,21 @@ int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, co
 }
 
 int SctpTransport::handleSend(size_t free) {
-	try {
+	PLOG_VERBOSE << "Handle send, free=" << free;
-		std::lock_guard lock(mSendMutex);
+	return safeFlush() ? 0 : -1;
-		trySendQueue();
-	} catch (const std::exception &e) {
-		PLOG_ERROR << "SCTP send: " << e.what();
-		return -1;
-	}
-	return 0; // success
 }
 
 int SctpTransport::handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_df) {
 	try {
+		PLOG_VERBOSE << "Handle write, len=" << len;
+
 		std::unique_lock lock(mWriteMutex);
 		if (!outgoing(make_message(data, data + len)))
 			return -1;
 		mWritten = true;
 		mWrittenOnce = true;
 		mWrittenCondition.notify_all();
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "SCTP write: " << e.what();
 		return -1;
@@ -444,6 +456,8 @@ int SctpTransport::handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_
 }
 
 void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, PayloadId ppid) {
+	PLOG_VERBOSE << "Process data, len=" << len;
+
 	// The usage of the PPIDs "WebRTC String Partial" and "WebRTC Binary Partial" is deprecated.
 	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13#section-6.6
 	// We handle them at reception for compatibility reasons but should never send them.
@@ -458,9 +472,11 @@ void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, Payl
 
 	case PPID_STRING:
 		if (mPartialStringData.empty()) {
+			mBytesReceived += len;
 			recv(make_message(data, data + len, Message::String, sid));
 		} else {
 			mPartialStringData.insert(mPartialStringData.end(), data, data + len);
+			mBytesReceived += mPartialStringData.size();
 			recv(make_message(mPartialStringData.begin(), mPartialStringData.end(), Message::String,
 			                  sid));
 			mPartialStringData.clear();
@@ -480,9 +496,11 @@ void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, Payl
 
 	case PPID_BINARY:
 		if (mPartialBinaryData.empty()) {
+			mBytesReceived += len;
 			recv(make_message(data, data + len, Message::Binary, sid));
 		} else {
 			mPartialBinaryData.insert(mPartialBinaryData.end(), data, data + len);
+			mBytesReceived += mPartialStringData.size();
 			recv(make_message(mPartialBinaryData.begin(), mPartialBinaryData.end(), Message::Binary,
 			                  sid));
 			mPartialBinaryData.clear();
@@ -504,10 +522,15 @@ void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, Payl
 }
 
 void SctpTransport::processNotification(const union sctp_notification *notify, size_t len) {
-	if (len != size_t(notify->sn_header.sn_length))
+	if (len != size_t(notify->sn_header.sn_length)) {
+		PLOG_WARNING << "Invalid notification length";
 		return;
+	}
 
-	switch (notify->sn_header.sn_type) {
+	auto type = notify->sn_header.sn_type;
+	PLOG_VERBOSE << "Process notification, type=" << type;
+
+	switch (type) {
 	case SCTP_ASSOC_CHANGE: {
 		const struct sctp_assoc_change &assoc_change = notify->sn_assoc_change;
 		if (assoc_change.sac_state == SCTP_COMM_UP) {
@@ -523,13 +546,16 @@ void SctpTransport::processNotification(const union sctp_notification *notify, s
 			}
 			mWrittenCondition.notify_all();
 		}
+		break;
 	}
+
 	case SCTP_SENDER_DRY_EVENT: {
 		// It not should be necessary since the send callback should have been called already,
 		// but to be sure, let's try to send now.
-		std::lock_guard lock(mSendMutex);
+		safeFlush();
-		trySendQueue();
+		break;
 	}
+
 	case SCTP_STREAM_RESET_EVENT: {
 		const struct sctp_stream_reset_event &reset_event = notify->sn_strreset_event;
 		const int count = (reset_event.strreset_length - sizeof(reset_event)) / sizeof(uint16_t);
@@ -558,6 +584,26 @@ void SctpTransport::processNotification(const union sctp_notification *notify, s
 	}
 }
 
+void SctpTransport::clearStats() {
+	mBytesReceived = 0;
+	mBytesSent = 0;
+}
+
+size_t SctpTransport::bytesSent() { return mBytesSent; }
+
+size_t SctpTransport::bytesReceived() { return mBytesReceived; }
+
+std::optional<std::chrono::milliseconds> SctpTransport::rtt() {
+	struct sctp_status status = {};
+	socklen_t len = sizeof(status);
+
+	if (usrsctp_getsockopt(this->mSock, IPPROTO_SCTP, SCTP_STATUS, &status, &len)) {
+		PLOG_WARNING << "Could not read SCTP_STATUS";
+		return std::nullopt;
+	}
+	return std::chrono::milliseconds(status.sstat_primary.spinfo_srtt);
+}
+
 int SctpTransport::RecvCallback(struct socket *sock, union sctp_sockstore addr, void *data,
                                 size_t len, struct sctp_rcvinfo recv_info, int flags, void *ptr) {
 	int ret = static_cast<SctpTransport *>(ptr)->handleRecv(

src/sctptransport.hpp

@@ -35,6 +35,9 @@ namespace rtc {
 
 class SctpTransport : public Transport {
 public:
+	static void Init();
+	static void Cleanup();
+
 	enum class State { Disconnected, Connecting, Connected, Failed };
 
 	using amount_callback = std::function<void(uint16_t streamId, size_t amount)>;
@@ -46,11 +49,17 @@ public:
 
 	State state() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if buffered
 	void flush();
 	void reset(unsigned int stream);
 
+	// Stats
+	void clearStats();
+	size_t bytesSent();
+	size_t bytesReceived();
+	std::optional<std::chrono::milliseconds> rtt();
+
 private:
 	// Order seems wrong but these are the actual values
 	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13#section-8
@@ -72,6 +81,7 @@ private:
 	bool trySendQueue();
 	bool trySendMessage(message_ptr message);
 	void updateBufferedAmount(uint16_t streamId, long delta);
+	bool safeFlush();
 
 	int handleRecv(struct socket *sock, union sctp_sockstore addr, const byte *data, size_t len,
 	               struct sctp_rcvinfo recv_info, int flags);
@@ -94,23 +104,18 @@ private:
 	bool mWritten = false;
 	bool mWrittenOnce = false;
 
-	std::atomic<bool> mShutdown = false;
-
 	state_callback mStateChangeCallback;
 	std::atomic<State> mState;
 
+	// Stats
+	std::atomic<size_t> mBytesSent = 0, mBytesReceived = 0;
+
 	binary mPartialRecv, mPartialStringData, mPartialBinaryData;
 
 	static int RecvCallback(struct socket *sock, union sctp_sockstore addr, void *data, size_t len,
 	                        struct sctp_rcvinfo recv_info, int flags, void *user_data);
 	static int SendCallback(struct socket *sock, uint32_t sb_free);
 	static int WriteCallback(void *sctp_ptr, void *data, size_t len, uint8_t tos, uint8_t set_df);
-
-	void GlobalInit();
-	void GlobalCleanup();
-
-	static std::mutex GlobalMutex;
-	static int InstancesCount;
 };
 
 } // namespace rtc

src/transport.hpp

@@ -38,9 +38,10 @@ public:
 	}
 	virtual ~Transport() { stop(); }
 
-	virtual void stop() {
+	virtual bool stop() {
 		if (mLower)
 			mLower->onRecv(nullptr);
+		return !mShutdown.exchange(true);
 	}
 
 	virtual bool send(message_ptr message) = 0;
@@ -61,6 +62,7 @@ protected:
 private:
 	std::shared_ptr<Transport> mLower;
 	synchronized_callback<message_ptr> mRecvCallback;
+	std::atomic<bool> mShutdown = false;
 };
 
 } // namespace rtc

test/capi.cpp

@@ -131,21 +131,33 @@ static void deletePeer(Peer *peer) {
 }
 
 int test_capi_main() {
+	int attempts;
+
 	rtcInitLogger(RTC_LOG_DEBUG);
 
-	rtcConfiguration config;
-	memset(&config, 0, sizeof(config));
-	// const char *iceServers[1] = {"stun:stun.l.google.com:19302"};
-	// config.iceServers = iceServers;
-	// config.iceServersCount = 1;
-
 	// Create peer 1
-	peer1 = createPeer(&config);
+	rtcConfiguration config1;
+	memset(&config1, 0, sizeof(config1));
+	// STUN server example
+	// const char *iceServers[1] = {"stun:stun.l.google.com:19302"};
+	// config1.iceServers = iceServers;
+	// config1.iceServersCount = 1;
+
+	peer1 = createPeer(&config1);
 	if (!peer1)
 		goto error;
 
 	// Create peer 2
-	peer2 = createPeer(&config);
+	rtcConfiguration config2;
+	memset(&config2, 0, sizeof(config2));
+	// STUN server example
+	// config2.iceServers = iceServers;
+	// config2.iceServersCount = 1;
+	// Port range example
+	config2.portRangeBegin = 5000;
+	config2.portRangeEnd = 6000;
+
+	peer2 = createPeer(&config2);
 	if (!peer2)
 		goto error;
 
@@ -155,7 +167,19 @@ int test_capi_main() {
 	rtcSetClosedCallback(peer1->dc, closedCallback);
 	rtcSetMessageCallback(peer1->dc, messageCallback);
 
-	sleep(3);
+	attempts = 10;
+	while (!peer2->connected && !peer1->connected && attempts--)
+		sleep(1);
+
+	if (peer1->state != RTC_CONNECTED || peer2->state != RTC_CONNECTED) {
+		fprintf(stderr, "PeerConnection is not connected\n");
+		goto error;
+	}
+
+	if (!peer1->connected || !peer2->connected) {
+		fprintf(stderr, "DataChannel is not connected\n");
+		goto error;
+	}
 
 	char buffer[256];
 	if (rtcGetLocalAddress(peer1->pc, buffer, 256) >= 0)
@@ -167,13 +191,13 @@ int test_capi_main() {
 	if (rtcGetRemoteAddress(peer2->pc, buffer, 256) >= 0)
 		printf("Remote address 2: %s\n", buffer);
 
-	if (peer1->connected && peer2->connected) {
+	deletePeer(peer1);
-		deletePeer(peer1);
+	sleep(1);
-		deletePeer(peer2);
+	deletePeer(peer2);
-		sleep(1);
+	sleep(1);
-		printf("Success\n");
+
-		return 0;
+	printf("Success\n");
-	}
+	return 0;
 
 error:
 	deletePeer(peer1);

test/connectivity.cpp

@@ -31,12 +31,20 @@ template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
 void test_connectivity() {
 	InitLogger(LogLevel::Debug);
 
-	Configuration config;
+	Configuration config1;
-	// config.iceServers.emplace_back("stun:stun.l.google.com:19302");
+	// STUN server example
+	// config1.iceServers.emplace_back("stun:stun.l.google.com:19302");
 
-	auto pc1 = std::make_shared<PeerConnection>(config);
+	auto pc1 = std::make_shared<PeerConnection>(config1);
 
-	auto pc2 = std::make_shared<PeerConnection>(config);
+	Configuration config2;
+	// STUN server example
+	// config2.iceServers.emplace_back("stun:stun.l.google.com:19302");
+	// Port range example
+	config2.portRangeBegin = 5000;
+	config2.portRangeEnd = 6000;
+
+	auto pc2 = std::make_shared<PeerConnection>(config2);
 
 	pc1->onLocalDescription([wpc2 = make_weak_ptr(pc2)](const Description &sdp) {
 		auto pc2 = wpc2.lock();
@@ -106,7 +114,16 @@ void test_connectivity() {
 		}
 	});
 
-	this_thread::sleep_for(3s);
+	int attempts = 10;
+	while ((!dc2 || !dc2->isOpen() || !dc1->isOpen()) && attempts--)
+		this_thread::sleep_for(1s);
+
+	if (pc1->state() != PeerConnection::State::Connected &&
+	    pc2->state() != PeerConnection::State::Connected)
+		throw runtime_error("PeerConnection is not connected");
+
+	if (!dc1->isOpen() || !dc2->isOpen())
+		throw runtime_error("DataChannel is not open");
 
 	if (auto addr = pc1->localAddress())
 		cout << "Local address 1:  " << *addr << endl;
@@ -117,12 +134,10 @@ void test_connectivity() {
 	if (auto addr = pc2->remoteAddress())
 		cout << "Remote address 2: " << *addr << endl;
 
-	if (!dc1->isOpen() || !dc2->isOpen())
+	// Delay close of peer 2 to check closing works properly
-		throw runtime_error("DataChannel is not open");
-
 	pc1->close();
+	this_thread::sleep_for(1s);
 	pc2->close();
-
 	this_thread::sleep_for(1s);
 
 	cout << "Success" << endl;

test/p2p/answerer.cpp

@@ -76,7 +76,8 @@ int main(int argc, char **argv) {
 		     << "* 0: Exit /"
 		     << " 1: Enter remote description /"
 		     << " 2: Enter remote candidate /"
-		     << " 3: Send message *" << endl
+		     << " 3: Send message /"
+		     << " 4: Print Connection Info *" << endl
 		     << "[Command]: ";
 
 		int command = -1;
@@ -120,6 +121,30 @@ int main(int argc, char **argv) {
 			dc->send(message);
 			break;
 		}
+		case 4: {
+			// Connection Info
+			if (!dc || !dc->isOpen()) {
+				cout << "** Channel is not Open ** ";
+				break;
+			}
+			CandidateInfo local, remote;
+			std::optional<std::chrono::milliseconds> rtt = pc->rtt();
+			if (pc->getSelectedCandidatePair(&local, &remote)) {
+				cout << "Local: " << local.address << ":" << local.port << " " << local.type << " "
+				     << local.transportType << endl;
+				cout << "Remote: " << remote.address << ":" << remote.port << " " << remote.type
+				     << " " << remote.transportType << endl;
+				cout << "Bytes Sent:" << pc->bytesSent()
+				     << " / Bytes Received:" << pc->bytesReceived() << " / Round-Trip Time:";
+				if (rtt.has_value())
+					cout << rtt.value().count();
+				else
+					cout << "null";
+				cout << " ms";
+			} else
+				cout << "Could not get Candidate Pair Info" << endl;
+			break;
+		}
 		default: {
 			cout << "** Invalid Command ** ";
 			break;

test/p2p/offerer.cpp

@@ -77,7 +77,8 @@ int main(int argc, char **argv) {
 		     << "* 0: Exit /"
 		     << " 1: Enter remote description /"
 		     << " 2: Enter remote candidate /"
-		     << " 3: Send message *" << endl
+		     << " 3: Send message /"
+		     << " 4: Print Connection Info *" << endl
 		     << "[Command]: ";
 
 		int command = -1;
@@ -120,6 +121,30 @@ int main(int argc, char **argv) {
 			dc->send(message);
 			break;
 		}
+		case 4: {
+			// Connection Info
+			if (!dc || !dc->isOpen()) {
+				cout << "** Channel is not Open ** ";
+				break;
+			}
+			CandidateInfo local, remote;
+			std::optional<std::chrono::milliseconds> rtt = pc->rtt();
+			if (pc->getSelectedCandidatePair(&local, &remote)) {
+				cout << "Local: " << local.address << ":" << local.port << " " << local.type << " "
+				     << local.transportType << endl;
+				cout << "Remote: " << remote.address << ":" << remote.port << " " << remote.type
+				     << " " << remote.transportType << endl;
+				cout << "Bytes Sent:" << pc->bytesSent()
+				     << " / Bytes Received:" << pc->bytesReceived() << " / Round-Trip Time:";
+				if (rtt.has_value())
+					cout << rtt.value().count();
+				else
+					cout << "null";
+				cout << " ms";
+			} else
+				cout << "Could not get Candidate Pair Info" << endl;
+			break;
+		}
 		default: {
 			cout << "** Invalid Command ** ";
 			break;