Bumped version to 0.5.0

Asynchronous certificate generation

.github/workflows/build-gnutls.yml

@@ -1,11 +1,14 @@
-name: Build and test
+name: Build and test with GnuTLS
-on: [push]
+on:
-
+  push:
+    branches:
+    - master
+  pull_request:
+    branches:
+    - master
 jobs:
   build:
-
     runs-on: ubuntu-latest
-    
     steps:
     - uses: actions/checkout@v2
     - name: install packages

.github/workflows/build-nice.yml

@@ -0,0 +1,24 @@
+name: Build and test with libnice
+on:
+  push:
+    branches:
+    - master
+  pull_request:
+    branches:
+    - master
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: install packages
+      run: sudo apt update && sudo apt install libgnutls28-dev libnice-dev
+    - name: submodules
+      run: git submodule update --init --recursive
+    - name: cmake
+      run: cmake -B build -DUSE_JUICE=0 -DUSE_GNUTLS=1
+    - name: make
+      run: (cd build; make)
+    - name: test
+      run: ./build/tests
+

.github/workflows/build-openssl.yml

@@ -0,0 +1,24 @@
+name: Build and test with OpenSSL
+on:
+  push:
+    branches:
+    - master
+  pull_request:
+    branches:
+    - master
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: install packages
+      run: sudo apt update && sudo apt install libssl-dev
+    - name: submodules
+      run: git submodule update --init --recursive
+    - name: cmake
+      run: cmake -B build -DUSE_JUICE=1 -DUSE_GNUTLS=0
+    - name: make
+      run: (cd build; make)
+    - name: test
+      run: ./build/tests
+

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.7)
 project (libdatachannel
 	DESCRIPTION "WebRTC DataChannels Library"
-	VERSION 0.4.7
+	VERSION 0.5.0
 	LANGUAGES CXX)
 
 option(USE_GNUTLS "Use GnuTLS instead of OpenSSL" OFF)
@@ -17,8 +17,9 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 set(CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/cmake/Modules)
 
 if(WIN32)
-	if (MSYS OR MINGW)
+	add_definitions(-DWIN32_LEAN_AND_MEAN)
-		add_definitions(-DSCTP_STDINT_INCLUDE=<stdint.h>)
+	if (MSVC)
+		add_definitions(-DNOMINMAX)
 	endif()
 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
@@ -50,10 +71,19 @@ set(TESTS_ANSWERER_SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/test/p2p/answerer.cpp
 )
 
-set(THREADS_PREFER_PTHREAD_FLAG ON)
+set(CMAKE_THREAD_PREFER_PTHREAD TRUE)
+set(THREADS_PREFER_PTHREAD_FLAG TRUE)
 find_package(Threads REQUIRED)
 
 add_subdirectory(deps/usrsctp EXCLUDE_FROM_ALL)
+if (MSYS OR MINGW)
+	target_compile_definitions(usrsctp PUBLIC -DSCTP_STDINT_INCLUDE=<stdint.h>)
+	target_compile_definitions(usrsctp-static PUBLIC -DSCTP_STDINT_INCLUDE=<stdint.h>)
+endif()
+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)
 
@@ -63,10 +93,11 @@ set_target_properties(datachannel PROPERTIES
 	CXX_STANDARD 17)
 
 target_include_directories(datachannel PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
+target_include_directories(datachannel PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/deps/plog/include)
 target_include_directories(datachannel PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include/rtc)
 target_include_directories(datachannel PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
-target_include_directories(datachannel PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/deps/plog/include)
+target_link_libraries(datachannel PUBLIC Threads::Threads)
-target_link_libraries(datachannel Threads::Threads Usrsctp::UsrsctpStatic)
+target_link_libraries(datachannel PRIVATE Usrsctp::UsrsctpStatic)
 
 add_library(datachannel-static STATIC EXCLUDE_FROM_ALL ${LIBDATACHANNEL_SOURCES})
 set_target_properties(datachannel-static PROPERTIES
@@ -74,14 +105,15 @@ set_target_properties(datachannel-static PROPERTIES
 	CXX_STANDARD 17)
 
 target_include_directories(datachannel-static PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
+target_include_directories(datachannel-static PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/deps/plog/include)
 target_include_directories(datachannel-static PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include/rtc)
 target_include_directories(datachannel-static PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
-target_include_directories(datachannel-static PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/deps/plog/include)
+target_link_libraries(datachannel-static PUBLIC Threads::Threads)
-target_link_libraries(datachannel-static Threads::Threads Usrsctp::UsrsctpStatic)
+target_link_libraries(datachannel-static PRIVATE Usrsctp::UsrsctpStatic)
 
 if(WIN32)
-	target_link_libraries(datachannel "wsock32" "ws2_32") # winsock2
+	target_link_libraries(datachannel PRIVATE wsock32 ws2_32) # winsock2
-	target_link_libraries(datachannel-static "wsock32" "ws2_32") # winsock2
+	target_link_libraries(datachannel-static PRIVATE wsock32 ws2_32) # winsock2
 endif()
 
 if (USE_GNUTLS)
@@ -95,34 +127,37 @@ if (USE_GNUTLS)
 			IMPORTED_LOCATION "${GNUTLS_LIBRARIES}")
 	endif()
 	target_compile_definitions(datachannel PRIVATE USE_GNUTLS=1)
-	target_link_libraries(datachannel GnuTLS::GnuTLS)
+	target_link_libraries(datachannel PRIVATE GnuTLS::GnuTLS)
 	target_compile_definitions(datachannel-static PRIVATE USE_GNUTLS=1)
-	target_link_libraries(datachannel-static GnuTLS::GnuTLS)
+	target_link_libraries(datachannel-static PRIVATE GnuTLS::GnuTLS)
 else()
 	find_package(OpenSSL REQUIRED)
 	target_compile_definitions(datachannel PRIVATE USE_GNUTLS=0)
-	target_link_libraries(datachannel OpenSSL::SSL)
+	target_link_libraries(datachannel PRIVATE OpenSSL::SSL)
 	target_compile_definitions(datachannel-static PRIVATE USE_GNUTLS=0)
-	target_link_libraries(datachannel-static OpenSSL::SSL)
+	target_link_libraries(datachannel-static PRIVATE OpenSSL::SSL)
 endif()
 
 if (USE_JUICE)
 	add_subdirectory(deps/libjuice EXCLUDE_FROM_ALL)
 	target_compile_definitions(datachannel PRIVATE USE_JUICE=1)
-	target_link_libraries(datachannel LibJuice::LibJuiceStatic)
+	target_link_libraries(datachannel PRIVATE LibJuice::LibJuiceStatic)
 	target_compile_definitions(datachannel-static PRIVATE USE_JUICE=1)
-	target_link_libraries(datachannel-static LibJuice::LibJuiceStatic)
+	target_link_libraries(datachannel-static PRIVATE LibJuice::LibJuiceStatic)
 else()
 	find_package(LibNice REQUIRED)
 	target_compile_definitions(datachannel PRIVATE USE_JUICE=0)
-	target_link_libraries(datachannel LibNice::LibNice)
+	target_link_libraries(datachannel PRIVATE LibNice::LibNice)
 	target_compile_definitions(datachannel-static PRIVATE USE_JUICE=0)
-	target_link_libraries(datachannel-static LibNice::LibNice)
+	target_link_libraries(datachannel-static PRIVATE LibNice::LibNice)
 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

Jamfile

@@ -1,3 +1,5 @@
+import feature : feature ;
+
 project libdatachannel ;
 path-constant CWD : . ;
 
@@ -5,21 +7,26 @@ lib libdatachannel
 	: # sources
 	[ glob ./src/*.cpp ]
 	: # requirements
+	<cxxstd>17
 	<include>./include/rtc
-	<define>USE_GNUTLS=0
 	<define>USE_JUICE=1
-	<cxxflags>"`pkg-config --cflags openssl`"
 	<library>/libdatachannel//usrsctp
 	<library>/libdatachannel//juice
+	<library>/libdatachannel//plog
 	: # default build
 	<link>static
 	: # usage requirements
 	<include>./include
 	<library>/libdatachannel//plog
 	<cxxflags>-pthread
-	<linkflags>"`pkg-config --libs openssl`"
 	;
 
+feature crypto : openssl gnutls : composite propagated ;
+feature.compose <crypto>openssl
+    : <define>USE_GNUTLS=0 ;
+feature.compose <crypto>gnutls
+    : <define>USE_GNUTLS=1 ;
+
 alias plog
     : # no sources
     : # no build requirements
@@ -57,9 +64,21 @@ actions make_libusrsctp
 }
 
 make libjuice.a : : @make_libjuice ;
+
+rule make_libjuice ( targets * : sources * : properties * )
+{
+	if <crypto>gnutls in $(properties)
+	{
+		MAKEOPTS on $(targets) = "USE_NETTLE=1" ;
+	}
+	else
+	{
+		MAKEOPTS on $(targets) = "USE_NETTLE=0" ;
+	}
+}
 actions make_libjuice
 {
-	(cd $(CWD)/deps/libjuice && make USE_NETTLE=0)
+	(cd $(CWD)/deps/libjuice && make $(MAKEOPTS))
     cp $(CWD)/deps/libjuice/libjuice.a $(<)
 }
 

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 .
 

cmake/Modules/FindLibNice.cmake

@@ -10,7 +10,7 @@ if (NOT TARGET LibNice::LibNice)
             HINTS ${PC_LIBNICE_LIBDIR} ${PC_LIBNICE_LIBRARY_DIRS})
 
     include(FindPackageHandleStandardArgs)
-    find_package_handle_standard_args(Libnice DEFAULT_MSG
+    find_package_handle_standard_args(LibNice DEFAULT_MSG
             LIBNICE_LIBRARY LIBNICE_INCLUDE_DIR)
     mark_as_advanced(LIBNICE_INCLUDE_DIR LIBNICE_LIBRARY)
 

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
@@ -69,7 +68,7 @@ public:
 
 	synchronized_callback &operator=(std::function<void(P...)> func) {
 		std::lock_guard lock(mutex);
-		callback = func;
+		callback = std::move(func);
 		return *this;
 	}
 
@@ -79,7 +78,10 @@ public:
 			callback(args...);
 	}
 
-	operator bool() const { return callback ? true : false; }
+	operator bool() const {
+		std::lock_guard lock(mutex);
+		return callback ? true : false;
+	}
 
 private:
 	std::function<void(P...)> callback;

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/message.hpp

@@ -28,9 +28,9 @@
 namespace rtc {
 
 struct Message : binary {
-	enum Type { Binary, String, Control };
+	enum Type { Binary, String, Control, Reset };
 
-	Message(size_t size) : binary(size), type(Binary) {}
+	Message(size_t size, Type type_ = Binary) : binary(size), type(type_) {}
 
 	template <typename Iterator>
 	Message(Iterator begin_, Iterator end_, Type type_ = Binary)
@@ -46,7 +46,7 @@ using mutable_message_ptr = std::shared_ptr<Message>;
 using message_callback = std::function<void(message_ptr message)>;
 
 constexpr auto message_size_func = [](const message_ptr &m) -> size_t {
-	return m->type != Message::Control ? m->size() : 0;
+	return m->type == Message::Binary || m->type == Message::String ? m->size() : 0;
 };
 
 template <typename Iterator>
@@ -59,6 +59,15 @@ message_ptr make_message(Iterator begin, Iterator end, Message::Type type = Mess
 	return message;
 }
 
+inline message_ptr make_message(size_t size, Message::Type type = Message::Binary,
+                                unsigned int stream = 0,
+                                std::shared_ptr<Reliability> reliability = nullptr) {
+	auto message = std::make_shared<Message>(size, type);
+	message->stream = stream;
+	message->reliability = reliability;
+	return message;
+}
+
 } // namespace rtc
 
 #endif

include/rtc/peerconnection.hpp

@@ -24,12 +24,14 @@
 #include "datachannel.hpp"
 #include "description.hpp"
 #include "include.hpp"
+#include "init.hpp"
 #include "message.hpp"
 #include "reliability.hpp"
 #include "rtc.hpp"
 
 #include <atomic>
 #include <functional>
+#include <future>
 #include <list>
 #include <mutex>
 #include <shared_mutex>
@@ -43,6 +45,9 @@ class IceTransport;
 class DtlsTransport;
 class SctpTransport;
 
+using certificate_ptr = std::shared_ptr<Certificate>;
+using future_certificate_ptr = std::shared_future<certificate_ptr>;
+
 class PeerConnection : public std::enable_shared_from_this<PeerConnection> {
 public:
 	enum class State : int {
@@ -51,14 +56,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 +91,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,11 +124,13 @@ 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;
+	const future_certificate_ptr mCertificate;
 
 	std::optional<Description> mLocalDescription, mRemoteDescription;
 	mutable std::recursive_mutex mLocalDescriptionMutex, mRemoteDescriptionMutex;
@@ -121,7 +138,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);
@@ -111,6 +114,9 @@ int rtcGetAvailableAmount(int dc); // total size available to receive
 int rtcSetAvailableCallback(int dc, availableCallbackFunc cb);
 int rtcReceiveMessage(int dc, char *buffer, int *size);
 
+// Cleanup
+void rtcCleanup();
+
 #ifdef __cplusplus
 } // extern "C"
 #endif

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

@@ -141,14 +141,9 @@ string make_fingerprint(gnutls_x509_crt_t crt) {
 	return oss.str();
 }
 
-shared_ptr<Certificate> make_certificate(const string &commonName) {
+namespace {
-	static std::unordered_map<string, shared_ptr<Certificate>> cache;
-	static std::mutex cacheMutex;
-
-	std::lock_guard lock(cacheMutex);
-	if (auto it = cache.find(commonName); it != cache.end())
-		return it->second;
 
+certificate_ptr make_certificate_impl(string commonName) {
 	std::unique_ptr<gnutls_x509_crt_t, decltype(&delete_crt)> crt(create_crt(), delete_crt);
 	std::unique_ptr<gnutls_x509_privkey_t, decltype(&delete_privkey)> privkey(create_privkey(),
 	                                                                          delete_privkey);
@@ -174,11 +169,11 @@ shared_ptr<Certificate> make_certificate(const string &commonName) {
 	check_gnutls(gnutls_x509_crt_sign2(*crt, *crt, *privkey, GNUTLS_DIG_SHA256, 0),
 	             "Unable to auto-sign certificate");
 
-	auto certificate = std::make_shared<Certificate>(*crt, *privkey);
+	return std::make_shared<Certificate>(*crt, *privkey);
-	cache.emplace(std::make_pair(commonName, certificate));
-	return certificate;
 }
 
+} // namespace
+
 } // namespace rtc
 
 #else
@@ -236,22 +231,9 @@ string make_fingerprint(X509 *x509) {
 	return oss.str();
 }
 
+namespace {
 
-shared_ptr<Certificate> make_certificate(const string &commonName) {
+certificate_ptr make_certificate_impl(string commonName) {
-	static std::unordered_map<string, shared_ptr<Certificate>> cache;
-	static std::mutex cacheMutex;
-
-	std::lock_guard lock(cacheMutex);
-	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);
 
@@ -288,12 +270,54 @@ shared_ptr<Certificate> make_certificate(const string &commonName) {
 	if (!X509_sign(x509.get(), pkey.get(), EVP_sha256()))
 		throw std::runtime_error("Unable to auto-sign certificate");
 
-	auto certificate = std::make_shared<Certificate>(x509, pkey);
+	return std::make_shared<Certificate>(x509, pkey);
-	cache.emplace(std::make_pair(commonName, certificate));
-	return certificate;
 }
 
+} // namespace
+
 } // namespace rtc
 
 #endif
 
+// Common for GnuTLS and OpenSSL
+
+namespace rtc {
+
+namespace {
+
+// Helper function roughly equivalent to std::async with policy std::launch::async
+// since std::async might be unreliable on some platforms (e.g. Mingw32 on Windows)
+template <class F, class... Args>
+std::future<std::result_of_t<std::decay_t<F>(std::decay_t<Args>...)>> thread_call(F &&f,
+                                                                                  Args &&... args) {
+	using R = std::result_of_t<std::decay_t<F>(std::decay_t<Args>...)>;
+	std::packaged_task<R()> task(std::bind(f, std::forward<Args>(args)...));
+	std::future<R> future = task.get_future();
+	std::thread t(std::move(task));
+	t.detach();
+	return future;
+}
+
+static std::unordered_map<string, future_certificate_ptr> CertificateCache;
+static std::mutex CertificateCacheMutex;
+
+} // namespace
+
+future_certificate_ptr make_certificate(string commonName) {
+	std::lock_guard lock(CertificateCacheMutex);
+
+	if (auto it = CertificateCache.find(commonName); it != CertificateCache.end())
+		return it->second;
+
+	auto future = thread_call(make_certificate_impl, commonName);
+	auto shared = future.share();
+	CertificateCache.emplace(std::move(commonName), shared);
+	return shared;
+}
+
+void CleanupCertificateCache() {
+	std::lock_guard lock(CertificateCacheMutex);
+	CertificateCache.clear();
+}
+
+} // namespace rtc

src/certificate.hpp

@@ -21,6 +21,7 @@
 
 #include "include.hpp"
 
+#include <future>
 #include <tuple>
 
 #if USE_GNUTLS
@@ -62,7 +63,12 @@ string make_fingerprint(gnutls_x509_crt_t crt);
 string make_fingerprint(X509 *x509);
 #endif
 
-std::shared_ptr<Certificate> make_certificate(const string &commonName);
+using certificate_ptr = std::shared_ptr<Certificate>;
+using future_certificate_ptr = std::shared_future<certificate_ptr>;
+
+future_certificate_ptr make_certificate(string commonName); // cached
+
+void CleanupCertificateCache();
 
 } // namespace rtc
 

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,16 +93,20 @@ string DataChannel::protocol() const { return mProtocol; }
 Reliability DataChannel::reliability() const { return *mReliability; }
 
 void DataChannel::close() {
-	if (mIsOpen.exchange(false) && mSctpTransport)
-		mSctpTransport->reset(mStream);
 	mIsClosed = true;
+	if (mIsOpen.exchange(false))
+		if (auto transport = mSctpTransport.lock())
+			transport->close(mStream);
+
 	mSctpTransport.reset();
+	resetCallbacks();
 }
 
 void DataChannel::remoteClose() {
-	mIsOpen = false;
 	if (!mIsClosed.exchange(true))
 		triggerClosed();
+
+	mIsOpen = false;
 	mSctpTransport.reset();
 }
 
@@ -135,6 +140,9 @@ std::optional<std::variant<binary, string>> DataChannel::receive() {
 			    string(reinterpret_cast<const char *>(message->data()), message->size()));
 		case Message::Binary:
 			return std::make_optional(std::move(*message));
+		default:
+			// Ignore
+			break;
 		}
 	}
 
@@ -147,17 +155,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 +193,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 +249,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 +289,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,17 +55,22 @@ static bool check_gnutls(int ret, const string &message = "GnuTLS error") {
 
 namespace rtc {
 
-DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certificate> certificate,
+void DtlsTransport::Init() {
-                             verifier_callback verifierCallback,
+	// Nothing to do
-                             state_callback stateChangeCallback)
+}
+
+void DtlsTransport::Cleanup() {
+	// Nothing to do
+}
+
+DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, certificate_ptr certificate,
+                             verifier_callback verifierCallback, state_callback stateChangeCallback)
     : Transport(lower), mCertificate(certificate), mState(State::Disconnected),
       mVerifierCallback(std::move(verifierCallback)),
       mStateChangeCallback(std::move(stateChangeCallback)) {
 
 	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 +83,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);
@@ -92,6 +99,7 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 	gnutls_transport_set_pull_timeout_function(mSession, TimeoutCallback);
 
 	mRecvThread = std::thread(&DtlsTransport::runRecvLoop, this);
+	registerIncoming();
 }
 
 DtlsTransport::~DtlsTransport() {
@@ -102,15 +110,14 @@ 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();
+	mRecvThread.join();
-		gnutls_bye(mSession, GNUTLS_SHUT_RDWR);
+	return true;
-		mRecvThread.join();
-	}
 }
 
 bool DtlsTransport::send(message_ptr message) {
@@ -131,10 +138,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 +158,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 +182,7 @@ void DtlsTransport::runRecvLoop() {
 
 	// Receive loop
 	try {
+		PLOG_INFO << "DTLS handshake done";
 		changeState(State::Connected);
 
 		const size_t bufferSize = maxMtu;
@@ -203,6 +215,8 @@ void DtlsTransport::runRecvLoop() {
 		PLOG_ERROR << "DTLS recv: " << e.what();
 	}
 
+	gnutls_bye(mSession, GNUTLS_SHUT_RDWR);
+
 	PLOG_INFO << "DTLS disconnected";
 	changeState(State::Disconnected);
 	recv(nullptr);
@@ -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);
@@ -396,6 +410,7 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 	SSL_set_tmp_ecdh(mSsl, ecdh.get());
 
 	mRecvThread = std::thread(&DtlsTransport::runRecvLoop, this);
+	registerIncoming();
 }
 
 DtlsTransport::~DtlsTransport() {
@@ -405,16 +420,15 @@ 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);
-
+	return true;
-		SSL_shutdown(mSsl);
-	}
 }
 
 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,18 +43,21 @@ 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)>;
 	using state_callback = std::function<void(State state)>;
 
-	DtlsTransport(std::shared_ptr<IceTransport> lower, std::shared_ptr<Certificate> certificate,
+	DtlsTransport(std::shared_ptr<IceTransport> lower, certificate_ptr certificate,
 	              verifier_callback verifierCallback, state_callback stateChangeCallback);
 	~DtlsTransport();
 
 	State state() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if dropped
 
 private:
@@ -62,7 +65,7 @@ private:
 	void changeState(State state);
 	void runRecvLoop();
 
-	const std::shared_ptr<Certificate> mCertificate;
+	const certificate_ptr mCertificate;
 
 	Queue<message_ptr> mIncomingQueue;
 	std::atomic<State> mState;
@@ -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,8 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 IceTransport::~IceTransport() { stop(); }
 
-void IceTransport::stop() {
+bool IceTransport::stop() {
-	// Nothing to do
+	return Transport::stop();
 }
 
 Description::Role IceTransport::role() const { return mRole; }
@@ -162,12 +168,6 @@ bool IceTransport::send(message_ptr message) {
 	return outgoing(message);
 }
 
-void IceTransport::incoming(message_ptr message) { recv(message); }
-
-void IceTransport::incoming(const byte *data, int size) {
-	incoming(make_message(data, data + size));
-}
-
 bool IceTransport::outgoing(message_ptr message) {
 	return juice_send(mAgent.get(), reinterpret_cast<const char *>(message->data()),
 	                  message->size()) >= 0;
@@ -224,7 +224,9 @@ void IceTransport::RecvCallback(juice_agent_t *agent, const char *data, size_t s
                                 void *user_ptr) {
 	auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
 	try {
-		iceTransport->incoming(reinterpret_cast<const byte *>(data), size);
+		PLOG_VERBOSE << "Incoming size=" << size;
+		auto b = reinterpret_cast<const byte *>(data);
+		iceTransport->incoming(make_message(b, b + size));
 	} catch (const std::exception &e) {
 		PLOG_WARNING << e.what();
 	}
@@ -309,6 +311,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 +437,22 @@ 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";
+	nice_agent_attach_recv(mNiceAgent.get(), mStreamId, 1, g_main_loop_get_context(mMainLoop.get()),
+	                       NULL, NULL);
+	nice_agent_remove_stream(mNiceAgent.get(), mStreamId);
+	g_main_loop_quit(mMainLoop.get());
+	mMainLoopThread.join();
+	return true;
 }
 
 Description::Role IceTransport::role() const { return mRole; }
@@ -516,12 +536,6 @@ bool IceTransport::send(message_ptr message) {
 	return outgoing(message);
 }
 
-void IceTransport::incoming(message_ptr message) { recv(message); }
-
-void IceTransport::incoming(const byte *data, int size) {
-	incoming(make_message(data, data + size));
-}
-
 bool IceTransport::outgoing(message_ptr message) {
 	return nice_agent_send(mNiceAgent.get(), mStreamId, 1, message->size(),
 	                       reinterpret_cast<const char *>(message->data())) >= 0;
@@ -609,7 +623,9 @@ void IceTransport::RecvCallback(NiceAgent *agent, guint streamId, guint componen
                                 gchar *buf, gpointer userData) {
 	auto iceTransport = static_cast<rtc::IceTransport *>(userData);
 	try {
-		iceTransport->incoming(reinterpret_cast<byte *>(buf), len);
+		PLOG_VERBOSE << "Incoming size=" << len;
+		auto b = reinterpret_cast<byte *>(buf);
+		iceTransport->incoming(make_message(b, b + len));
 	} catch (const std::exception &e) {
 		PLOG_WARNING << e.what();
 	}
@@ -645,6 +661,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_PEER_REFLEXIVE:
+		return CandidateType::PeerReflexive;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_RELAYED:
+		return CandidateType::Relayed;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_SERVER_REFLEXIVE:
+		return CandidateType::ServerReflexive;
+	default:
+		return CandidateType::Host;
+	}
+}
+
+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;
+	default:
+		return CandidateTransportType::Udp;
+	}
+}
+
 } // namespace rtc
 
 #endif

src/icetransport.hpp

@@ -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,12 +81,10 @@ 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:
-	void incoming(message_ptr message) override;
-	void incoming(const byte *data, int size);
 	bool outgoing(message_ptr message) override;
 
 	void changeState(State state);
@@ -133,6 +133,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,88 @@
+/**
+ * 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 "certificate.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() {
+	CleanupCertificateCache();
+	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,34 @@ 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) {}
+    : mConfig(config), mCertificate(make_certificate("libdatachannel")), mState(State::New),
+      mGatheringState(GatheringState::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 +102,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 +204,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 +231,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 +248,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 +266,17 @@ 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 certificate = mCertificate.get();
 		auto lower = std::atomic_load(&mIceTransport);
 		auto transport = std::make_shared<DtlsTransport>(
-		    lower, mCertificate, std::bind(&PeerConnection::checkFingerprint, this, _1),
+		    lower, certificate, 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 +292,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 +310,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 +340,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 +356,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,12 +419,12 @@ 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 {
-			// Invalid, close the DataChannel by resetting the stream
+			// Invalid, close the DataChannel
-			sctpTransport->reset(message->stream);
+			sctpTransport->close(message->stream);
 			return;
 		}
 	}
@@ -397,7 +444,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 +458,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;
 	}
 }
 
@@ -456,9 +514,11 @@ void PeerConnection::processLocalDescription(Description description) {
 	if (auto remote = remoteDescription())
 		remoteSctpPort = remote->sctpPort();
 
+	auto certificate = mCertificate.get(); // wait for certificate if not ready
+
 	std::lock_guard lock(mLocalDescriptionMutex);
 	mLocalDescription.emplace(std::move(description));
-	mLocalDescription->setFingerprint(mCertificate->fingerprint());
+	mLocalDescription->setFingerprint(certificate->fingerprint());
 	mLocalDescription->setSctpPort(remoteSctpPort.value_or(DEFAULT_SCTP_PORT));
 	mLocalDescription->setMaxMessageSize(LOCAL_MAX_MESSAGE_SIZE);
 
@@ -483,21 +543,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 +635,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 +661,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));
 }
 
@@ -442,3 +450,5 @@ int rtcReceiveMessage(int dc, char *buffer, int *size) {
 		    *message);
 	});
 }
+
+void rtcCleanup() { rtc::Cleanup(); }

src/sctptransport.cpp

@@ -21,6 +21,7 @@
 #include <chrono>
 #include <exception>
 #include <iostream>
+#include <thread>
 #include <vector>
 
 #ifdef USE_JUICE
@@ -49,30 +50,22 @@ 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() {
-	std::lock_guard lock(GlobalMutex);
+	while (usrsctp_finish() != 0)
-	if (--InstancesCount == 0) {
+		std::this_thread::sleep_for(100ms);
-		usrsctp_finish();
-	}
 }
 
 SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
@@ -84,7 +77,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 +148,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
@@ -167,32 +167,36 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 		throw std::runtime_error("Could not set SCTP send buffer size, errno=" +
 		                         std::to_string(errno));
 
+	registerIncoming();
 	connect();
 }
 
 SctpTransport::~SctpTransport() {
 	stop();
 
-	usrsctp_close(mSock);
+	if (mSock)
-	usrsctp_deregister_address(this);
+		usrsctp_close(mSock);
 
-	GlobalCleanup();
+	usrsctp_deregister_address(this);
 }
 
 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() {
+	if (!mSock)
+		return;
+
 	PLOG_DEBUG << "SCTP connect";
 	changeState(State::Connecting);
 
@@ -216,12 +220,19 @@ void SctpTransport::connect() {
 }
 
 void SctpTransport::shutdown() {
+	if (!mSock)
+		return;
+
 	PLOG_DEBUG << "SCTP shutdown";
 
-	if (usrsctp_shutdown(mSock, SHUT_RDWR)) {
+	if (usrsctp_shutdown(mSock, SHUT_RDWR) != 0 && errno != ENOTCONN) {
 		PLOG_WARNING << "SCTP shutdown failed, errno=" << errno;
 	}
 
+	// close() abort the connection when linger is disabled, call it now
+	usrsctp_close(mSock);
+	mSock = nullptr;
+
 	PLOG_INFO << "SCTP disconnected";
 	changeState(State::Disconnected);
 	mWrittenCondition.notify_all();
@@ -243,31 +254,15 @@ bool SctpTransport::send(message_ptr message) {
 	return false;
 }
 
+void SctpTransport::close(unsigned int stream) {
+	send(make_message(0, Message::Reset, uint16_t(stream)));
+}
+
 void SctpTransport::flush() {
 	std::lock_guard lock(mSendMutex);
 	trySendQueue();
 }
 
-void SctpTransport::reset(unsigned int stream) {
-	PLOG_DEBUG << "SCTP resetting stream " << stream;
-
-	std::unique_lock lock(mWriteMutex);
-	mWritten = false;
-	using srs_t = struct sctp_reset_streams;
-	const size_t len = sizeof(srs_t) + sizeof(uint16_t);
-	byte buffer[len] = {};
-	srs_t &srs = *reinterpret_cast<srs_t *>(buffer);
-	srs.srs_flags = SCTP_STREAM_RESET_OUTGOING;
-	srs.srs_number_streams = 1;
-	srs.srs_stream_list[0] = uint16_t(stream);
-	if (usrsctp_setsockopt(mSock, IPPROTO_SCTP, SCTP_RESET_STREAMS, &srs, len) == 0) {
-		mWrittenCondition.wait_for(lock, 1000ms,
-		                           [&]() { return mWritten || mState != State::Connected; });
-	} else {
-		PLOG_WARNING << "SCTP reset stream " << stream << " failed, errno=" << errno;
-	}
-}
-
 void SctpTransport::incoming(message_ptr message) {
 	// There could be a race condition here where we receive the remote INIT before the local one is
 	// sent, which would result in the connection being aborted. Therefore, we need to wait for data
@@ -277,13 +272,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) {
@@ -305,16 +302,9 @@ bool SctpTransport::trySendQueue() {
 
 bool SctpTransport::trySendMessage(message_ptr message) {
 	// Requires mSendMutex to be locked
-	if (mState != State::Connected)
+	if (!mSock || mState != State::Connected)
 		return false;
 
-	PLOG_VERBOSE << "SCTP try send size=" << message->size();
-
-	// TODO: Implement SCTP ndata specification draft when supported everywhere
-	// See https://tools.ietf.org/html/draft-ietf-tsvwg-sctp-ndata-08
-
-	const Reliability reliability = message->reliability ? *message->reliability : Reliability();
-
 	uint32_t ppid;
 	switch (message->type) {
 	case Message::String:
@@ -323,11 +313,24 @@ bool SctpTransport::trySendMessage(message_ptr message) {
 	case Message::Binary:
 		ppid = !message->empty() ? PPID_BINARY : PPID_BINARY_EMPTY;
 		break;
-	default:
+	case Message::Control:
 		ppid = PPID_CONTROL;
 		break;
+	case Message::Reset:
+		sendReset(message->stream);
+		return true;
+	default:
+		// Ignore
+		return true;
 	}
 
+	PLOG_VERBOSE << "SCTP try send size=" << message->size();
+
+	// TODO: Implement SCTP ndata specification draft when supported everywhere
+	// See https://tools.ietf.org/html/draft-ietf-tsvwg-sctp-ndata-08
+
+	const Reliability reliability = message->reliability ? *message->reliability : Reliability();
+
 	struct sctp_sendv_spa spa = {};
 
 	// set sndinfo
@@ -368,6 +371,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 +386,69 @@ 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);
 }
 
+void SctpTransport::sendReset(uint16_t streamId) {
+	// Requires mSendMutex to be locked
+	if (!mSock || state() != State::Connected)
+		return;
+
+	PLOG_DEBUG << "SCTP resetting stream " << streamId;
+
+	using srs_t = struct sctp_reset_streams;
+	const size_t len = sizeof(srs_t) + sizeof(uint16_t);
+	byte buffer[len] = {};
+	srs_t &srs = *reinterpret_cast<srs_t *>(buffer);
+	srs.srs_flags = SCTP_STREAM_RESET_OUTGOING;
+	srs.srs_number_streams = 1;
+	srs.srs_stream_list[0] = streamId;
+
+	mWritten = false;
+	if (usrsctp_setsockopt(mSock, IPPROTO_SCTP, SCTP_RESET_STREAMS, &srs, len) == 0) {
+		std::unique_lock lock(mWriteMutex); // locking before setsockopt might deadlock usrsctp...
+		mWrittenCondition.wait_for(lock, 1000ms,
+		                           [&]() { return mWritten || mState != State::Connected; });
+	} else if (errno == EINVAL) {
+		PLOG_VERBOSE << "SCTP stream " << streamId << " already reset";
+	} else {
+		PLOG_WARNING << "SCTP reset stream " << streamId << " failed, errno=" << errno;
+	}
+}
+
+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 +456,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 +467,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 +490,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 +506,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 +530,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 +556,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 +580,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);
@@ -538,7 +598,7 @@ void SctpTransport::processNotification(const union sctp_notification *notify, s
 		if (flags & SCTP_STREAM_RESET_OUTGOING_SSN) {
 			for (int i = 0; i < count; ++i) {
 				uint16_t streamId = reset_event.strreset_stream_list[i];
-				reset(streamId);
+				close(streamId);
 			}
 		}
 		if (flags & SCTP_STREAM_RESET_INCOMING_SSN) {
@@ -558,6 +618,28 @@ 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<milliseconds> SctpTransport::rtt() {
+	if (!mSock || state() != State::Connected)
+		return nullopt;
+
+	struct sctp_status status = {};
+	socklen_t len = sizeof(status);
+	if (usrsctp_getsockopt(mSock, IPPROTO_SCTP, SCTP_STATUS, &status, &len)) {
+		PLOG_WARNING << "Could not read SCTP_STATUS";
+		return nullopt;
+	}
+	return 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,10 +49,16 @@ public:
 
 	State state() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if buffered
+	void close(unsigned int stream);
 	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
@@ -72,6 +81,8 @@ private:
 	bool trySendQueue();
 	bool trySendMessage(message_ptr message);
 	void updateBufferedAmount(uint16_t streamId, long delta);
+	void sendReset(uint16_t streamId);
+	bool safeFlush();
 
 	int handleRecv(struct socket *sock, union sctp_sockstore addr, const byte *data, size_t len,
 	               struct sctp_rcvinfo recv_info, int flags);
@@ -89,28 +100,23 @@ private:
 	std::map<uint16_t, size_t> mBufferedAmount;
 	amount_callback mBufferedAmountCallback;
 
-	std::recursive_mutex mWriteMutex;
+	std::mutex mWriteMutex;
-	std::condition_variable_any mWrittenCondition;
+	std::condition_variable mWrittenCondition;
-	bool mWritten = false;
+	std::atomic<bool> mWritten = false; // written outside lock
 	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

@@ -32,25 +32,30 @@ using namespace std::placeholders;
 
 class Transport {
 public:
-	Transport(std::shared_ptr<Transport> lower = nullptr) : mLower(std::move(lower)) {
+	Transport(std::shared_ptr<Transport> lower = nullptr) : mLower(std::move(lower)) {}
+	virtual ~Transport() {
+		stop();
+		if (mLower)
+			mLower->onRecv(nullptr); // doing it on stop could cause a deadlock
+	}
+
+	virtual bool stop() {
+		return !mShutdown.exchange(true);
+	}
+
+	void registerIncoming() {
 		if (mLower)
 			mLower->onRecv(std::bind(&Transport::incoming, this, _1));
 	}
-	virtual ~Transport() { stop(); }
-
-	virtual void stop() {
-		if (mLower)
-			mLower->onRecv(nullptr);
-	}
-
-	virtual bool send(message_ptr message) = 0;
 
 	void onRecv(message_callback callback) { mRecvCallback = std::move(callback); }
 
+	virtual bool send(message_ptr message) { return outgoing(message); }
+
 protected:
 	void recv(message_ptr message) { mRecvCallback(message); }
 
-	virtual void incoming(message_ptr message) = 0;
+	virtual void incoming(message_ptr message) { recv(message); }
 	virtual bool outgoing(message_ptr message) {
 		if (mLower)
 			return mLower->send(message);
@@ -61,6 +66,7 @@ protected:
 private:
 	std::shared_ptr<Transport> mLower;
 	synchronized_callback<message_ptr> mRecvCallback;
+	std::atomic<bool> mShutdown = false;
 };
 
 } // namespace rtc

test/capi.cpp

@@ -23,9 +23,12 @@
 #include <cstdlib>
 #include <cstring>
 
+#ifdef _WIN32
+#include <windows.h>
+static void sleep(unsigned int secs) { Sleep(secs * 1000); }
+#else
 #include <unistd.h> // for sleep
-
+#endif
-using namespace std;
 
 typedef struct {
 	rtcState state;
@@ -126,25 +129,38 @@ static void deletePeer(Peer *peer) {
 			rtcDeleteDataChannel(peer->dc);
 		if (peer->pc)
 			rtcDeletePeerConnection(peer->pc);
+		free(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;
 
@@ -154,7 +170,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)
@@ -166,13 +194,17 @@ 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;
+	// You may call rtcCleanup() when finished to free static resources
-	}
+	rtcCleanup();
+	sleep(1);
+
+	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,14 @@ 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);
 
+	// You may call rtc::Cleanup() when finished to free static resources
+	rtc::Cleanup();
 	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;

test/p2p_c_version/CMakeLists.txt

@@ -0,0 +1,12 @@
+cmake_minimum_required(VERSION 3.5.1)
+project(offerer C)
+
+
+set(CMAKE_C_STANDARD 11)
+set(CMAKE_C_FLAGS "-Wall -g -O2")
+
+add_executable(offerer offerer.c)
+target_link_libraries(offerer  datachannel)
+
+add_executable(answerer answerer.c)
+target_link_libraries(answerer  datachannel)

test/p2p_c_version/answerer.c

@@ -0,0 +1,326 @@
+/**
+ * Copyright (c) 2020 Paul-Louis Ageneau
+ * Copyright (c) 2020 Stevedan Ogochukwu Omodolor
+ *
+ * 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 <rtc/rtc.h>
+
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h> // for sleep
+#include <ctype.h>
+
+typedef struct {
+        rtcState state;
+        rtcGatheringState gatheringState;
+        int pc;
+        int dc;
+        bool connected;
+} Peer;
+
+Peer *peer = NULL;
+
+static void dataChannelCallback(int dc, void *ptr);
+
+static void descriptionCallback(const char *sdp, const char *type, void *ptr);
+
+static void candidateCallback(const char *cand, const char *mid, void *ptr);
+
+static void stateChangeCallback(rtcState state, void *ptr);
+
+static void gatheringStateCallback(rtcGatheringState state, void *ptr);
+
+static void closedCallback(void *ptr);
+
+static void messageCallback(const char *message, int size, void *ptr);
+
+static void deletePeer(Peer *peer);
+
+int all_space(const char *str);
+char* state_print(rtcState state);
+char* rtcGatheringState_print(rtcState state);
+
+int main(int argc, char **argv) {
+        rtcInitLogger(RTC_LOG_DEBUG);
+
+        // Create peer
+        rtcConfiguration config;
+        memset(&config, 0, sizeof(config));
+
+        Peer *peer = (Peer *)malloc(sizeof(Peer));
+        if (!peer) {
+
+                printf("Error allocating memory for peer\n");
+                deletePeer(peer);
+
+        }
+        memset(peer, 0, sizeof(Peer));
+
+        printf("Peer created\n");
+
+        // Create peer connection
+        peer->pc = rtcCreatePeerConnection(&config);
+
+        rtcSetUserPointer(peer->pc, peer);
+        rtcSetLocalDescriptionCallback(peer->pc, descriptionCallback);
+        rtcSetLocalCandidateCallback(peer->pc, candidateCallback);
+        rtcSetStateChangeCallback(peer->pc, stateChangeCallback);
+        rtcSetGatheringStateChangeCallback(peer->pc, gatheringStateCallback);
+
+        rtcSetUserPointer(peer->dc, NULL);
+        rtcSetDataChannelCallback(peer->pc, dataChannelCallback);
+
+
+
+
+        bool exit = false;
+
+        while (!exit) {
+
+                printf("\n");
+                printf("***************************************************************************************\n");
+
+                // << endl
+                printf("* 0: Exit /"
+                       " 1: Enter remote description /"
+                       " 2: Enter remote candidate /"
+                       " 3: Send message /"
+                       " 4: Print Connection Info *\n"
+                       "[Command]: ");
+
+                int command = -1;
+                int c;
+                // int check_scan
+                if (scanf("%d", &command)) {
+
+                }else {
+                        break;
+                }
+                while ((c = getchar()) != '\n' && c != EOF) { }
+
+                fflush(stdin);
+                switch (command) {
+                case 0: {
+                        exit = true;
+                        break;
+                }
+                case 1: {
+                        // Parse Description
+                        printf("[Description]: ");
+
+
+                        char *line = NULL;
+                        size_t len = 0;
+                        size_t read = 0;
+                        char *sdp = (char*) malloc(sizeof(char));
+                        while ((read = getline(&line, &len, stdin)) != -1 && !all_space(line)) {
+                                sdp = (char*) realloc (sdp,(strlen(sdp)+1) +strlen(line)+1);
+                                strcat(sdp, line);
+
+                        }
+                        printf("%s\n",sdp);
+                        rtcSetRemoteDescription(peer->pc, sdp, "offer");
+                        free(sdp);
+                        free(line);
+                        break;
+
+                }
+                case 2: {
+                        // Parse Candidate
+                        printf("[Candidate]: ");
+                        char* candidate = NULL;
+                        size_t candidate_size = 0;
+
+                        if(getline(&candidate, &candidate_size, stdin)) {
+                                rtcAddRemoteCandidate(peer->pc, candidate, "0");
+                                free(candidate);
+
+                        }else {
+                                printf("Error reading line\n");
+                                break;
+                        }
+
+                        break;
+                }
+                case 3: {
+                        // Send Message
+                        if(!peer->connected) {
+                                printf("** Channel is not Open **");
+                                break;
+                        }
+                        printf("[Message]: ");
+                        char* message = NULL;
+                        size_t message_size = 0;
+
+                        if(getline(&message, &message_size, stdin)) {
+                                rtcSendMessage(peer->dc, message, -1);
+                                free(message);
+                        }else {
+                                printf("Error reading line\n");
+                                break;
+                        }
+                        break;
+                }
+                case 4: {
+                        // Connection Info
+                        if(!peer->connected) {
+                                printf("** Channel is not Open **");
+                                break;
+                        }
+                        char buffer[256];
+                        if (rtcGetLocalAddress(peer->pc, buffer, 256) >= 0)
+                                printf("Local address 1:  %s\n", buffer);
+                        if (rtcGetRemoteAddress(peer->pc, buffer, 256) >= 0)
+                                printf("Remote address 1: %s\n", buffer);
+
+                        else
+                                printf("Could not get Candidate Pair Info\n");
+                        break;
+                }
+                default: {
+                        printf("** Invalid Command **");
+                        break;
+                }
+                }
+        }
+
+
+        deletePeer(peer);
+        return 0;
+}
+
+
+static void descriptionCallback(const char *sdp, const char *type, void *ptr) {
+        // Peer *peer = (Peer *)ptr;
+        printf("Description %s:\n%s\n", "answerer", sdp);
+}
+
+static void candidateCallback(const char *cand, const char *mid, void *ptr) {
+        // Peer *peer = (Peer *)ptr;
+        printf("Candidate %s: %s\n", "answerer", cand);
+
+}
+
+static void stateChangeCallback(rtcState state, void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->state = state;
+        printf("State %s: %s\n", "answerer", state_print(state));
+}
+
+static void gatheringStateCallback(rtcGatheringState state, void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->gatheringState = state;
+        printf("Gathering state %s: %s\n", "answerer", rtcGatheringState_print(state));
+}
+
+
+
+static void closedCallback(void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->connected = false;
+
+
+}
+static void messageCallback(const char *message, int size, void *ptr) {
+        // Peer *peer = (Peer *)ptr;
+        if (size < 0) { // negative size indicates a null-terminated string
+                printf("Message %s: %s\n", "answerer", message);
+        } else {
+                printf("Message %s: [binary of size %d]\n", "answerer", size);
+        }
+}
+static void deletePeer(Peer *peer) {
+        if (peer) {
+                if (peer->dc)
+                        rtcDeleteDataChannel(peer->dc);
+                if (peer->pc)
+                        rtcDeletePeerConnection(peer->pc);
+                free(peer);
+        }
+}
+
+static void dataChannelCallback(int dc, void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->dc = dc;
+        peer->connected = true;
+        rtcSetClosedCallback(dc, closedCallback);
+        rtcSetMessageCallback(dc, messageCallback);
+        char buffer[256];
+        if (rtcGetDataChannelLabel(dc, buffer, 256) >= 0)
+                printf("DataChannel %s: Received with label \"%s\"\n", "answerer", buffer);
+
+
+}
+int all_space(const char *str) {
+        while (*str) {
+                if (!isspace(*str++)) {
+                        return 0;
+                }
+        }
+        return 1;
+}
+
+char* state_print(rtcState state) {
+        char *str = NULL;
+        switch (state) {
+        case RTC_NEW:
+                str = "RTC_NEW";
+                break;
+        case RTC_CONNECTING:
+                str = "RTC_CONNECTING";
+                break;
+        case RTC_CONNECTED:
+                str = "RTC_CONNECTED";
+                break;
+        case RTC_DISCONNECTED:
+                str = "RTC_DISCONNECTED";
+                break;
+        case RTC_FAILED:
+                str = "RTC_FAILED";
+                break;
+        case RTC_CLOSED:
+                str = "RTC_CLOSED";
+                break;
+        default:
+                break;
+        }
+
+        return str;
+
+}
+
+char* rtcGatheringState_print(rtcState state) {
+        char* str = NULL;
+        switch (state) {
+        case RTC_GATHERING_NEW:
+                str = "RTC_GATHERING_NEW";
+                break;
+        case RTC_GATHERING_INPROGRESS:
+                str = "RTC_GATHERING_INPROGRESS";
+                break;
+        case RTC_GATHERING_COMPLETE:
+                str = "RTC_GATHERING_COMPLETE";
+                break;
+        default:
+                break;
+        }
+
+        return str;
+
+}

test/p2p_c_version/offerer.c

@@ -0,0 +1,334 @@
+/**
+ * Copyright (c) 2020 Paul-Louis Ageneau
+ * Copyright (c) 2020 Stevedan Ogochukwu Omodolor
+ *
+ * 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 <rtc/rtc.h>
+
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h> // for sleep
+#include <ctype.h>
+
+char* state_print(rtcState state);
+char* rtcGatheringState_print(rtcState state);
+
+typedef struct {
+        rtcState state;
+        rtcGatheringState gatheringState;
+        int pc;
+        int dc;
+        bool connected;
+} Peer;
+
+Peer *peer = NULL;
+
+static void descriptionCallback(const char *sdp, const char *type, void *ptr);
+
+static void candidateCallback(const char *cand, const char *mid, void *ptr);
+
+static void stateChangeCallback(rtcState state, void *ptr);
+
+static void gatheringStateCallback(rtcGatheringState state, void *ptr);
+
+static void openCallback(void *ptr);
+
+static void closedCallback(void *ptr);
+
+static void messageCallback(const char *message, int size, void *ptr);
+
+static void deletePeer(Peer *peer);
+
+int all_space(const char *str);
+
+int main(int argc, char **argv){
+        rtcInitLogger(RTC_LOG_DEBUG);
+
+        // Create peer
+        rtcConfiguration config;
+        memset(&config, 0, sizeof(config));
+
+        Peer *peer = (Peer *)malloc(sizeof(Peer));
+        if (!peer) {
+
+                printf("Error allocating memory for peer\n");
+                deletePeer(peer);
+
+        }
+        memset(peer, 0, sizeof(Peer));
+
+        printf("Peer created\n");
+
+        // Create peer connection
+        peer->pc = rtcCreatePeerConnection(&config);
+        rtcSetUserPointer(peer->pc, peer);
+        rtcSetLocalDescriptionCallback(peer->pc, descriptionCallback);
+        rtcSetLocalCandidateCallback(peer->pc, candidateCallback);
+        rtcSetStateChangeCallback(peer->pc, stateChangeCallback);
+        rtcSetGatheringStateChangeCallback(peer->pc, gatheringStateCallback);
+
+        // Since this is the offere, we will create a datachannel
+        peer->dc = rtcCreateDataChannel(peer->pc, "test");
+
+        rtcSetOpenCallback(peer->dc, openCallback);
+
+
+        rtcSetClosedCallback(peer->dc, closedCallback);
+
+        rtcSetMessageCallback(peer->dc, messageCallback);
+
+
+        sleep(1);
+
+        bool exit = false;
+
+        while (!exit) {
+
+                printf("\n");
+                printf("***************************************************************************************\n");
+
+                // << endl
+                printf("* 0: Exit /"
+                       " 1: Enter remote description /"
+                       " 2: Enter remote candidate /"
+                       " 3: Send message /"
+                       " 4: Print Connection Info *\n"
+                       "[Command]: ");
+
+                int command = -1;
+                int c;
+                if (scanf("%d", &command)) {
+
+                }else {
+                        break;
+                }
+                while ((c = getchar()) != '\n' && c != EOF) { }
+                fflush(stdin);
+
+                switch (command) {
+                case 0: {
+                        exit = true;
+                        break;
+                }
+                case 1: {
+                        // Parse Description
+                        printf("[Description]: ");
+
+
+                        char *line = NULL;
+                        size_t len = 0;
+                        size_t read = 0;
+                        char *sdp = (char*) malloc(sizeof(char));
+                        while ((read = getline(&line, &len, stdin)) != -1 && !all_space(line)) {
+                                sdp = (char*) realloc (sdp,(strlen(sdp)+1) +strlen(line)+1);
+                                strcat(sdp, line);
+
+                        }
+                        printf("%s\n",sdp);
+                        rtcSetRemoteDescription(peer->pc, sdp, "answer");
+                        free(sdp);
+                        free(line);
+                        break;
+
+                }
+                case 2: {
+                        // Parse Candidate
+                        printf("[Candidate]: ");
+                        char* candidate = NULL;
+                        size_t candidate_size = 0;
+                        if(getline(&candidate, &candidate_size, stdin)) {
+                                rtcAddRemoteCandidate(peer->pc, candidate, "0");
+                                free(candidate);
+
+                        }else {
+                                printf("Error reading line\n");
+                                break;
+                        }
+
+
+                        break;
+                }
+                case 3: {
+                        // Send Message
+                        if(!peer->connected) {
+                                printf("** Channel is not Open **");
+                                break;
+                        }
+                        printf("[Message]: ");
+                        char* message = NULL;
+                        size_t message_size = 0;
+                        if(getline(&message, &message_size, stdin)) {
+                                rtcSendMessage(peer->dc, message, -1);
+                                free(message);
+                        }else {
+                                printf("Error reading line\n");
+                                break;
+                        }
+
+                        break;
+                }
+                case 4: {
+                        // Connection Info
+                        if(!peer->connected) {
+                                printf("** Channel is not Open **");
+                                break;
+                        }
+                        char buffer[256];
+                        if (rtcGetLocalAddress(peer->pc, buffer, 256) >= 0)
+                                printf("Local address 1:  %s\n", buffer);
+                        if (rtcGetRemoteAddress(peer->pc, buffer, 256) >= 0)
+                                printf("Remote address 1: %s\n", buffer);
+
+                        else
+                                printf("Could not get Candidate Pair Info\n");
+                        break;
+                }
+                default: {
+                        printf("** Invalid Command **");
+                        break;
+                }
+                }
+        }
+
+        deletePeer(peer);
+        return 0;
+}
+
+
+
+
+
+
+static void descriptionCallback(const char *sdp, const char *type, void *ptr) {
+        // Peer *peer = (Peer *)ptr;
+        printf("Description %s:\n%s\n", "offerer", sdp);
+}
+
+static void candidateCallback(const char *cand, const char *mid, void *ptr) {
+        // Peer *peer = (Peer *)ptr;
+        printf("Candidate %s: %s\n", "offerer", cand);
+
+}
+
+static void stateChangeCallback(rtcState state, void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->state = state;
+        printf("State %s: %s\n", "offerer", state_print(state));
+}
+
+static void gatheringStateCallback(rtcGatheringState state, void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->gatheringState = state;
+        printf("Gathering state %s: %s\n", "offerer", rtcGatheringState_print(state));
+}
+
+
+static void openCallback(void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->connected = true;
+        char buffer[256];
+        if (rtcGetDataChannelLabel(peer->dc, buffer, 256) >= 0)
+                printf("DataChannel %s: Received with label \"%s\"\n","offerer", buffer);
+
+
+}
+
+static void closedCallback(void *ptr) {
+        Peer *peer = (Peer *)ptr;
+        peer->connected = false;
+
+
+}
+
+static void messageCallback(const char *message, int size, void *ptr) {
+        // Peer *peer = (Peer *)ptr;
+        if (size < 0) { // negative size indicates a null-terminated string
+                printf("Message %s: %s\n", "offerer", message);
+        } else {
+                printf("Message %s: [binary of size %d]\n", "offerer", size);
+        }
+}
+static void deletePeer(Peer *peer) {
+        if (peer) {
+                if (peer->dc)
+                        rtcDeleteDataChannel(peer->dc);
+                if (peer->pc)
+                        rtcDeletePeerConnection(peer->pc);
+                free(peer);
+        }
+}
+
+
+int all_space(const char *str) {
+        while (*str) {
+                if (!isspace(*str++)) {
+                        return 0;
+                }
+        }
+        return 1;
+}
+
+char* state_print(rtcState state) {
+        char *str = NULL;
+        switch (state) {
+        case RTC_NEW:
+                str = "RTC_NEW";
+                break;
+        case RTC_CONNECTING:
+                str = "RTC_CONNECTING";
+                break;
+        case RTC_CONNECTED:
+                str = "RTC_CONNECTED";
+                break;
+        case RTC_DISCONNECTED:
+                str = "RTC_DISCONNECTED";
+                break;
+        case RTC_FAILED:
+                str = "RTC_FAILED";
+                break;
+        case RTC_CLOSED:
+                str = "RTC_CLOSED";
+                break;
+        default:
+                break;
+        }
+
+        return str;
+
+}
+
+char* rtcGatheringState_print(rtcState state) {
+        char* str = NULL;
+        switch (state) {
+        case RTC_GATHERING_NEW:
+                str = "RTC_GATHERING_NEW";
+                break;
+        case RTC_GATHERING_INPROGRESS:
+                str = "RTC_GATHERING_INPROGRESS";
+                break;
+        case RTC_GATHERING_COMPLETE:
+                str = "RTC_GATHERING_COMPLETE";
+                break;
+        default:
+                break;
+        }
+
+        return str;
+
+}