Bumped version to 0.6.5

Better workaround for usrsctp send after closing/unregistering

.github/workflows/build-gnutls.yml

@@ -1,4 +1,4 @@
-name: Build and test with GnuTLS
+name: Build with GnuTLS
 on:
   push:
     branches:
@@ -7,7 +7,7 @@ on:
     branches:
     - master
 jobs:
-  build-ubuntu:
+  build-linux:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2
@@ -31,9 +31,6 @@ jobs:
       run: git submodule update --init --recursive
     - name: cmake
       run: cmake -B build -DUSE_JUICE=1 -DUSE_GNUTLS=1
-      env:
-# hack to bypass EPERM issue on sendto()
-        CFLAGS: -DJUICE_ENABLE_ADDRS_LOCALHOST
     - name: make
       run: (cd build; make -j2)
     - name: test

.github/workflows/build-nice.yml

@@ -1,4 +1,4 @@
-name: Build and test with libnice
+name: Build with libnice
 on:
   push:
     branches:
@@ -7,7 +7,7 @@ on:
     branches:
     - master
 jobs:
-  build-ubuntu:
+  build-linux:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2

.github/workflows/build-openssl.yml

@@ -1,4 +1,4 @@
-name: Build and test with OpenSSL
+name: Build with OpenSSL
 on:
   push:
     branches:
@@ -7,7 +7,7 @@ on:
     branches:
     - master
 jobs:
-  build-ubuntu:
+  build-linux:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2
@@ -34,9 +34,25 @@ jobs:
       env:
         OPENSSL_ROOT_DIR: /usr/local/opt/openssl
         OPENSSL_LIBRARIES: /usr/local/opt/openssl/lib
-# hack to bypass EPERM issue on sendto()
-        CFLAGS: -DJUICE_ENABLE_ADDRS_LOCALHOST
     - name: make
       run: (cd build; make -j2)
     - name: test
       run: ./build/tests
+  build-windows:
+    runs-on: windows-latest
+    steps:
+    - uses: actions/checkout@v2
+    - uses: ilammy/msvc-dev-cmd@v1
+    - name: install packages
+      run: choco install openssl
+    - name: submodules
+      run: git submodule update --init --recursive
+    - name: cmake
+      run: cmake -B build -G "NMake Makefiles" -DUSE_JUICE=1 -DUSE_GNUTLS=0
+    - name: nmake
+      run: |
+        cd build
+        nmake
+    - name: test
+      run: build/tests.exe
+

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "deps/plog"]
 	path = deps/plog
-	url = https://github.com/SergiusTheBest/plog
+	url = https://github.com/paullouisageneau/plog
 [submodule "usrsctp"]
 	path = deps/usrsctp
 	url = https://github.com/sctplab/usrsctp.git

CMakeLists.txt

@@ -1,12 +1,14 @@
 cmake_minimum_required(VERSION 3.7)
 project(libdatachannel
 	DESCRIPTION "WebRTC Data Channels Library"
-	VERSION 0.6.1
+	VERSION 0.6.5
 	LANGUAGES CXX)
 
 option(USE_GNUTLS "Use GnuTLS instead of OpenSSL" OFF)
 option(USE_JUICE "Use libjuice instead of libnice" OFF)
-option(RTC_ENABLE_WEBSOCKET "Build WebSocket support" ON)
+option(USE_SRTP "Enable SRTP for media support" OFF)
+option(NO_WEBSOCKET "Disable WebSocket support" OFF)
+option(NO_EXAMPLES "Disable examples" OFF)
 
 if(USE_GNUTLS)
 	option(USE_NETTLE "Use Nettle instead of OpenSSL in libjuice" ON)
@@ -21,6 +23,8 @@ if(WIN32)
 	add_definitions(-DWIN32_LEAN_AND_MEAN)
 	if (MSVC)
 		add_definitions(-DNOMINMAX)
+		add_definitions(-D_CRT_SECURE_NO_WARNINGS)
+		add_definitions(-D_SILENCE_CXX17_CODECVT_HEADER_DEPRECATION_WARNING)
 	endif()
 endif()
 
@@ -40,6 +44,8 @@ set(LIBDATACHANNEL_SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/src/rtc.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/sctptransport.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/tls.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/threadpool.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/processor.cpp
 )
 
 set(LIBDATACHANNEL_WEBSOCKET_SOURCES
@@ -80,7 +86,6 @@ set(TESTS_SOURCES
 set(CMAKE_THREAD_PREFER_PTHREAD TRUE)
 set(THREADS_PREFER_PTHREAD_FLAG TRUE)
 find_package(Threads REQUIRED)
-find_package(SRTP)
 
 set(CMAKE_POLICY_DEFAULT_CMP0048 NEW)
 add_subdirectory(deps/plog)
@@ -97,7 +102,14 @@ endif()
 add_library(Usrsctp::Usrsctp ALIAS usrsctp)
 add_library(Usrsctp::UsrsctpStatic ALIAS usrsctp-static)
 
-if (RTC_ENABLE_WEBSOCKET)
+if (NO_WEBSOCKET)
+	add_library(datachannel SHARED
+		${LIBDATACHANNEL_SOURCES})
+	add_library(datachannel-static STATIC EXCLUDE_FROM_ALL
+		${LIBDATACHANNEL_SOURCES})
+	target_compile_definitions(datachannel PUBLIC RTC_ENABLE_WEBSOCKET=0)
+	target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_WEBSOCKET=0)
+else()
 	add_library(datachannel SHARED
 		${LIBDATACHANNEL_SOURCES}
 		${LIBDATACHANNEL_WEBSOCKET_SOURCES})
@@ -106,13 +118,6 @@ if (RTC_ENABLE_WEBSOCKET)
 		${LIBDATACHANNEL_WEBSOCKET_SOURCES})
 	target_compile_definitions(datachannel PUBLIC RTC_ENABLE_WEBSOCKET=1)
 	target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_WEBSOCKET=1)
-else()
-	add_library(datachannel SHARED
-		${LIBDATACHANNEL_SOURCES})
-	add_library(datachannel-static STATIC EXCLUDE_FROM_ALL
-		${LIBDATACHANNEL_SOURCES})
-	target_compile_definitions(datachannel PUBLIC RTC_ENABLE_WEBSOCKET=0)
-	target_compile_definitions(datachannel-static PUBLIC RTC_ENABLE_WEBSOCKET=0)
 endif()
 
 set_target_properties(datachannel PROPERTIES
@@ -139,7 +144,8 @@ if(WIN32)
 	target_link_libraries(datachannel-static PRIVATE wsock32 ws2_32) # winsock2
 endif()
 
-if(SRTP_FOUND)
+if(USE_SRTP)
+	find_package(SRTP REQUIRED)
 	if(NOT TARGET SRTP::SRTP)
 		add_library(SRTP::SRTP UNKNOWN IMPORTED)
 		set_target_properties(SRTP::SRTP PROPERTIES
@@ -205,7 +211,11 @@ set_target_properties(datachannel-tests PROPERTIES
 	CXX_STANDARD 17)
 set_target_properties(datachannel-tests PROPERTIES OUTPUT_NAME tests)
 target_include_directories(datachannel-tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
-target_link_libraries(datachannel-tests datachannel nlohmann_json::nlohmann_json)
+if(WIN32)
+	target_link_libraries(datachannel-tests datachannel-static) # DLL exports only the C API
+else()
+	target_link_libraries(datachannel-tests datachannel)
+endif()
 
 # Benchmark
 add_executable(datachannel-benchmark test/benchmark.cpp)
@@ -215,12 +225,18 @@ set_target_properties(datachannel-benchmark PROPERTIES
 set_target_properties(datachannel-benchmark PROPERTIES OUTPUT_NAME benchmark)
 target_compile_definitions(datachannel-benchmark PRIVATE BENCHMARK_MAIN=1)
 target_include_directories(datachannel-benchmark PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
+if(WIN32)
+	target_link_libraries(datachannel-benchmark datachannel-static) # DLL exports only the C API
+else()
 	target_link_libraries(datachannel-benchmark datachannel)
+endif()
 
 # Examples
+if(NOT NO_EXAMPLES)
 	set(JSON_BuildTests OFF CACHE INTERNAL "")
 	add_subdirectory(deps/json)
 	add_subdirectory(examples/client)
 	add_subdirectory(examples/copy-paste)
 	add_subdirectory(examples/copy-paste-capi)
+endif()
 

Jamfile

@@ -3,6 +3,12 @@ import feature : feature ;
 project libdatachannel ;
 path-constant CWD : . ;
 
+feature gnutls : off on : composite propagated ;
+feature.compose <gnutls>off
+    : <define>USE_GNUTLS=0 ;
+feature.compose <gnutls>on
+    : <define>USE_GNUTLS=1 ;
+
 lib libdatachannel
 	: # sources
 	[ glob ./src/*.cpp ]
@@ -12,25 +18,25 @@ lib libdatachannel
 	<define>USE_JUICE=1
 	<define>RTC_ENABLE_MEDIA=0
 	<define>RTC_ENABLE_WEBSOCKET=0
+	<toolset>msvc:<define>WIN32_LEAN_AND_MEAN
+	<toolset>msvc:<define>NOMINMAX
+	<toolset>msvc:<define>_CRT_SECURE_NO_WARNINGS
+	<toolset>msvc:<define>_SILENCE_CXX17_CODECVT_HEADER_DEPRECATION_WARNING
 	<library>/libdatachannel//usrsctp
 	<library>/libdatachannel//juice
 	<library>/libdatachannel//plog
+	<gnutls>on:<library>gnutls/<link>shared
+	<gnutls>off:<library>ssl
+	<gnutls>off:<library>crypto
 	: # default build
 	<link>static
 	: # usage requirements
 	<include>./include
 	<library>/libdatachannel//plog
-	<cxxflags>-pthread
-	<toolset>gcc:<cxxflags>"-Wno-pedantic -Wno-unused-parameter -Wno-unused-variable"
-	<toolset>clang:<cxxflags>"-Wno-pedantic -Wno-unused-parameter -Wno-unused-variable"
+	<toolset>gcc:<cxxflags>"-pthread -Wno-pedantic -Wno-unused-parameter -Wno-unused-variable"
+	<toolset>clang:<cxxflags>"-pthread -Wno-pedantic -Wno-unused-parameter -Wno-unused-variable"
 	;
 
-feature gnutls : off on : composite propagated ;
-feature.compose <gnutls>off
-    : <define>USE_GNUTLS=0 ;
-feature.compose <gnutls>on
-    : <define>USE_GNUTLS=1 ;
-
 alias plog
     : # no sources
     : # no build requirements
@@ -48,41 +54,207 @@ alias usrsctp
     <library>libusrsctp.a
     ;
 
+alias usrsctp
+    : # no sources
+    : <toolset>msvc
+    : # no default build
+    : # usage requirements
+    <include>./deps/usrsctp/usrsctplib
+    <library>usrsctp.lib
+    ;
+
 alias juice
     : # no sources
     : # no build requirements
     : # no default build
     : # usage requirements
     <include>./deps/libjuice/include
-	<library>libjuice.a
+    <library>libjuice-static.a
+    <gnutls>on:<library>nettle
+    ;
+
+alias juice
+    : # no sources
+    : <toolset>msvc
+    : # no default build
+    : # usage requirements
+    <include>./deps/libjuice/include
+    <library>juice-static.lib
     ;
 
 make libusrsctp.a : : @make_libusrsctp ;
+make usrsctp.lib : : @make_libusrsctp_msvc ;
+
+rule make_libusrsctp ( targets * : sources * : properties * )
+{
+	local VARIANT = [ feature.get-values <variant> : $(properties) ] ;
+	VARIANT on $(targets) = $(VARIANT) ;
+	if <gnutls>on in $(properties)
+	{ BUILD_DIR on $(targets) = "build-gnutls-$(VARIANT)" ; }
+	else
+	{ BUILD_DIR on $(targets) = "build-openssl-$(VARIANT)" ; }
+}
 actions make_libusrsctp
 {
-	(cd $(CWD)/deps/usrsctp && \
-		./bootstrap && \
-		./configure --enable-static --disable-debug CFLAGS="-fPIC -Wno-address-of-packed-member" && \
-		make)
-    cp $(CWD)/deps/usrsctp/usrsctplib/.libs/libusrsctp.a $(<)
+    (cd $(CWD)/deps/usrsctp && mkdir $(BUILD_DIR) && cd $(BUILD_DIR) && cmake -DCMAKE_BUILD_TYPE=$(VARIANT) -DCMAKE_C_FLAGS="-fPIC" .. && make -j2 usrsctp-static)
+    cp $(CWD)/deps/usrsctp/$(BUILD_DIR)/usrsctplib/libusrsctp.a $(<)
+}
+rule make_libusrsctp_msvc ( targets * : sources * : properties * )
+{
+	local VARIANT = [ feature.get-values <variant> : $(properties) ] ;
+	VARIANT on $(targets) = $(VARIANT) ;
+	if <gnutls>on in $(properties)
+	{ BUILD_DIR on $(targets) = "build-gnutls-$(VARIANT)" ; }
+	else
+	{ BUILD_DIR on $(targets) = "build-openssl-$(VARIANT)" ; }
+}
+actions make_libusrsctp_msvc
+{
+	SET OLDD=%CD%
+    cd $(CWD)/deps/usrsctp
+    mkdir $(BUILD_SIR)
+    cd $(BUILD_DIR)
+    cmake -G "Visual Studio 16 2019" ..
+    msbuild usrsctplib.sln /property:Configuration=$(VARIANT)
+    cd %OLDD%
+    cp $(CWD)/deps/usrsctp/build/usrsctplib/Release/usrsctp.lib $(<)
 }
 
-make libjuice.a : : @make_libjuice ;
+make libjuice-static.a : : @make_libjuice ;
+make juice-static.lib : : @make_libjuice_msvc ;
 
 rule make_libjuice ( targets * : sources * : properties * )
 {
-	if <crypto>gnutls in $(properties)
+	local VARIANT = [ feature.get-values <variant> : $(properties) ] ;
+	VARIANT on $(targets) = $(VARIANT) ;
+	if <gnutls>on in $(properties)
 	{
-		MAKEOPTS on $(targets) = "USE_NETTLE=1" ;
+		BUILD_DIR on $(targets) = "build-gnutls-$(VARIANT)" ;
+		CMAKEOPTS on $(targets) = "-DUSE_NETTLE=1" ;
 	}
 	else
 	{
-		MAKEOPTS on $(targets) = "USE_NETTLE=0" ;
+		local OPENSSL_INCLUDE = [ feature.get-values <openssl-include> : $(properties) ] ;
+
+		if <target-os>darwin in $(properties) && $(OPENSSL_INCLUDE) = ""
+		{
+			# on macOS, default to pick up openssl from the homebrew installation
+			# brew install openssl
+			OPENSSL_INCLUDE = /usr/local/opt/openssl/include ;
+		}
+
+		BUILD_DIR on $(targets) = "build-openssl-$(VARIANT)" ;
+		CMAKEOPTS on $(targets) = "-DUSE_NETTLE=0" ;
+		if $(OPENSSL_INCLUDE) != ""
+		{ CMAKEOPTS on $(targets) += " -DOPENSSL_ROOT_DIR=$(OPENSSL_INCLUDE)/.." ; }
 	}
 }
 actions make_libjuice
 {
-	(cd $(CWD)/deps/libjuice && make $(MAKEOPTS))
-    cp $(CWD)/deps/libjuice/libjuice.a $(<)
+    (cd $(CWD)/deps/libjuice && mkdir $(BUILD_DIR) && cd $(BUILD_DIR) && cmake -DCMAKE_C_FLAGS="-fPIC" -DCMAKE_BUILD_TYPE=$(VARIANT) $(CMAKEOPTS) .. && make -j2 juice-static)
+    cp $(CWD)/deps/libjuice/$(BUILD_DIR)/libjuice-static.a $(<)
+}
+rule make_libjuice_msvc ( targets * : sources * : properties * )
+{
+	local VARIANT = [ feature.get-values <variant> : $(properties) ] ;
+	VARIANT on $(targets) = $(VARIANT) ;
+	if <gnutls>on in $(properties)
+	{
+		BUILD_DIR on $(targets) += "build-gnutls-$(VARIANT)" ;
+		CMAKEOPTS on $(targets) = "-DUSE_NETTLE=1" ;
+	}
+	else
+	{
+		BUILD_DIR on $(targets) += "build-openssl-$(VARIANT)" ;
+		CMAKEOPTS on $(targets) = "-DUSE_NETTLE=0" ;
+	}
+}
+actions make_libjuice_msvc
+{
+	SET OLDD=%CD%
+    cd $(CWD)/deps/libjuice
+    mkdir $(BUILD_DIR)
+    cd $(BUILD_DIR)
+    cmake -G "Visual Studio 16 2019" $(CMAKEOPTS) ..
+    msbuild libjuice.sln /property:Configuration=$(VARIANT)
+    cd %OLDD%
+    cp $(CWD)/deps/libjuice/$(BUILD_DIR)/Release/juice-static.lib $(<)
 }
 
+# the search path to pick up the openssl libraries from. This is the <search>
+# property of those libraries
+rule openssl-lib-path ( properties * )
+{
+    local OPENSSL_LIB = [ feature.get-values <openssl-lib> : $(properties) ] ;
+
+    if <target-os>darwin in $(properties) && $(OPENSSL_LIB) = ""
+    {
+        # on macOS, default to pick up openssl from the homebrew installation
+        # brew install openssl
+        OPENSSL_LIB = /usr/local/opt/openssl/lib ;
+    }
+    else if <target-os>windows in $(properties) && $(OPENSSL_LIB) = ""
+    {
+        # on windows, assume openssl is installed to c:\OpenSSL-Win32
+        if <address-model>64 in $(properties)
+        { OPENSSL_LIB = c:\\OpenSSL-Win64\\lib ; }
+        else
+        { OPENSSL_LIB = c:\\OpenSSL-Win32\\lib ; }
+    }
+
+    local result ;
+    result += <search>$(OPENSSL_LIB) ;
+    return $(result) ;
+}
+
+# the include path to pick up openssl headers from. This is the
+# usage-requirement for the openssl-related libraries
+rule openssl-include-path ( properties * )
+{
+    local OPENSSL_INCLUDE = [ feature.get-values <openssl-include> : $(properties) ] ;
+
+    if <target-os>darwin in $(properties) && $(OPENSSL_INCLUDE) = ""
+    {
+        # on macOS, default to pick up openssl from the homebrew installation
+        # brew install openssl
+        OPENSSL_INCLUDE = /usr/local/opt/openssl/include ;
+    }
+    else if <target-os>windows in $(properties) && $(OPENSSL_INCLUDE) = ""
+    {
+        # on windows, assume openssl is installed to c:\OpenSSL-Win32
+        if <address-model>64 in $(properties)
+        { OPENSSL_INCLUDE = c:\\OpenSSL-Win64\\include ; }
+        else
+        { OPENSSL_INCLUDE = c:\\OpenSSL-Win32\\include ; }
+    }
+
+    local result ;
+    result += <include>$(OPENSSL_INCLUDE) ;
+    return $(result) ;
+}
+
+# libraries for OpenSSL on Windows
+lib advapi32 : : <name>advapi32 ;
+lib user32 : : <name>user32 ;
+lib shell32 : : <name>shell32 ;
+lib gdi32 : : <name>gdi32 ;
+lib bcrypt : : <name>bcrypt ;
+lib z : : <link>shared <name>z ;
+alias ssl-deps : advapi32 user32 shell32 gdi32 ;
+
+# OpenSSL on Windows
+lib crypto : ssl-deps : <toolset>msvc <openssl-version>1.1 <name>libcrypto
+    <conditional>@openssl-lib-path : : <conditional>@openssl-include-path ;
+lib ssl : ssl-deps : <toolset>msvc <openssl-version>1.1 <name>libssl <use>crypto
+    <conditional>@openssl-lib-path : : <conditional>@openssl-include-path ;
+
+# OpenSSL on other platforms
+lib crypto : : <name>crypto <use>z <conditional>@openssl-lib-path : :
+    <conditional>@openssl-include-path ;
+lib ssl : : <name>ssl <use>crypto <conditional>@openssl-lib-path : :
+    <conditional>@openssl-include-path ;
+
+# GnuTLS
+lib gnutls : : <link>shared <name>gnutls ;
+lib nettle : : <link>shared <name>nettle ;
+

Makefile

@@ -38,22 +38,22 @@ else
         LIBS+=glib-2.0 gobject-2.0 nice
 endif
 
-RTC_ENABLE_MEDIA ?= 0
-ifneq ($(RTC_ENABLE_MEDIA), 0)
+USE_SRTP ?= 0
+ifneq ($(USE_SRTP), 0)
         CPPFLAGS+=-DRTC_ENABLE_MEDIA=1
         LIBS+=srtp
 else
         CPPFLAGS+=-DRTC_ENABLE_MEDIA=0
 endif
 
-RTC_ENABLE_WEBSOCKET ?= 1
-ifneq ($(RTC_ENABLE_WEBSOCKET), 0)
+
+NO_WEBSOCKET ?= 0
+ifeq ($(NO_WEBSOCKET), 0)
         CPPFLAGS+=-DRTC_ENABLE_WEBSOCKET=1
 else
         CPPFLAGS+=-DRTC_ENABLE_WEBSOCKET=0
 endif
 
-
 INCLUDES+=$(shell pkg-config --cflags $(LIBS))
 LDLIBS+=$(LOCALLIBS) $(shell pkg-config --libs $(LIBS))
 

README.md

@@ -44,31 +44,55 @@ Features:
 
 ## Dependencies
 
+Dependencies:
 - GnuTLS: https://www.gnutls.org/ or OpenSSL: https://www.openssl.org/
 
-Optional:
+Optional dependencies:
 - libnice: https://nice.freedesktop.org/ (substituable with libjuice)
-- libSRTP: https://github.com/cisco/libsrtp
+- libSRTP: https://github.com/cisco/libsrtp (only necessary for media transport)
 
 Submodules:
 - libjuice: https://github.com/paullouisageneau/libjuice
 - usrsctp: https://github.com/sctplab/usrsctp
 
 ## Building
-### Building with CMake (preferred)
+
+### Clone repository and submodules
 
 ```bash
+$ git clone https://github.com/paullouisageneau/libdatachannel.git
+$ cd libdatachannel
 $ git submodule update --init --recursive
-$ mkdir build
-$ cd build
-$ cmake -DUSE_JUICE=1 -DUSE_GNUTLS=1 ..
-$ make
 ```
 
-### Building directly with Make
+### Building with CMake
+
+The CMake library targets `libdatachannel` and `libdatachannel-static` respectively correspond to the shared and static libraries. On Windows, the DLL resulting from the shared library build only exposes the C API, use the static library for the C++ API. The default target will build tests and examples.
+
+#### POSIX-compliant operating systems (including Linux and Apple macOS)
+```bash
+$ cmake -B build -DUSE_JUICE=1 -DUSE_GNUTLS=1
+$ cd build
+$ make -j2
+```
+
+#### Microsoft Windows with MinGW cross-compilation
+```bash
+$ cmake -B build -DUSE_JUICE=1 -DCMAKE_TOOLCHAIN_FILE=/usr/share/mingw/toolchain-x86_64-w64-mingw32.cmake # replace with your toolchain file
+$ cd build
+$ make -j2
+```
+
+#### Microsoft Windows with Microsoft Visual C++
+```bash
+$ cmake -B build -G "NMake Makefiles" -DUSE_JUICE=1
+$ cd build
+$ nmake
+```
+
+### Building directly with Make (Linux only)
 
 ```bash
-$ git submodule update --init --recursive
 $ make USE_JUICE=1 USE_GNUTLS=1
 ```
 

examples/client/CMakeLists.txt

@@ -7,5 +7,11 @@ add_executable(datachannel-client main.cpp)
 set_target_properties(datachannel-client PROPERTIES
 	CXX_STANDARD 17
 	OUTPUT_NAME client)
+
+if(WIN32)
+	target_link_libraries(datachannel-client datachannel-static) # DLL exports only the C API
+else()
+	target_link_libraries(datachannel-client datachannel)
+endif()
 target_link_libraries(datachannel-client datachannel nlohmann_json)
 

examples/client/main.cpp

@@ -212,7 +212,7 @@ string randomId(size_t length) {
 	    "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz");
 	string id(length, '0');
 	default_random_engine rng(random_device{}());
-	uniform_int_distribution<int> dist(0, characters.size() - 1);
+	uniform_int_distribution<int> dist(0, int(characters.size() - 1));
 	generate(id.begin(), id.end(), [&]() { return characters.at(dist(rng)); });
 	return id;
 }

examples/copy-paste-capi/CMakeLists.txt

@@ -2,7 +2,6 @@ cmake_minimum_required(VERSION 3.5.1)
 project(offerer C)
 
 set(CMAKE_C_STANDARD 11)
-set(CMAKE_C_FLAGS "-Wall -g -O2")
 
 add_executable(datachannel-copy-paste-capi-offerer offerer.c)
 set_target_properties(datachannel-copy-paste-capi-offerer PROPERTIES
@@ -13,3 +12,4 @@ add_executable(datachannel-copy-paste-capi-answerer answerer.c)
 set_target_properties(datachannel-copy-paste-capi-answerer PROPERTIES
 	OUTPUT_NAME answerer)
 target_link_libraries(datachannel-copy-paste-capi-answerer datachannel)
+

examples/copy-paste-capi/answerer.c

@@ -41,30 +41,22 @@ typedef struct {
         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);
+char *rtcGatheringState_print(rtcGatheringState state);
+
+int all_space(const char *str);
 
 int main(int argc, char **argv) {
-        rtcInitLogger(RTC_LOG_DEBUG);
+	rtcInitLogger(RTC_LOG_DEBUG, NULL);
 
 	// Create peer
 	rtcConfiguration config;
@@ -259,16 +251,6 @@ static void dataChannelCallback(int dc, void *ptr) {
                 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) {
@@ -297,7 +279,7 @@ char* state_print(rtcState state) {
 	return str;
 }
 
-char* rtcGatheringState_print(rtcState state) {
+char *rtcGatheringState_print(rtcGatheringState state) {
 	char *str = NULL;
 	switch (state) {
 	case RTC_GATHERING_NEW:
@@ -315,3 +297,13 @@ char* rtcGatheringState_print(rtcState state) {
 
 	return str;
 }
+
+int all_space(const char *str) {
+	while (*str) {
+		if (!isspace(*str++)) {
+			return 0;
+		}
+	}
+
+	return 1;
+}

examples/copy-paste-capi/getline.h

@@ -1,10 +1,11 @@
 // Simple POSIX getline() implementation
 // This code is public domain
 
-#include "malloc.h"
-#include "stdio.h"
+#include <malloc.h>
+#include <stddef.h>
+#include <stdio.h>
 
-ssize_t getline(char **lineptr, size_t *n, FILE *stream) {
+int getline(char **lineptr, size_t *n, FILE *stream) {
 	if (!lineptr || !stream || !n)
 		return -1;
 
@@ -20,7 +21,7 @@ ssize_t getline(char **lineptr, size_t *n, FILE *stream) {
 		*n = 128;
 	}
 
-	size_t pos = 0;
+	int pos = 0;
 	while(c != EOF) {
         if (pos + 1 >= *n) {
             size_t new_size = *n + (*n >> 2);
@@ -45,4 +46,3 @@ ssize_t getline(char **lineptr, size_t *n, FILE *stream) {
     (*lineptr)[pos] = '\0';
     return pos;
 }
-

examples/copy-paste-capi/offerer.c

@@ -33,9 +33,6 @@ static void sleep(unsigned int secs) { Sleep(secs * 1000); }
 #include <unistd.h> // for sleep
 #endif
 
-char* state_print(rtcState state);
-char* rtcGatheringState_print(rtcState state);
-
 typedef struct {
         rtcState state;
         rtcGatheringState gatheringState;
@@ -44,28 +41,22 @@ typedef struct {
         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);
 
+char *state_print(rtcState state);
+char *rtcGatheringState_print(rtcGatheringState state);
+
 int all_space(const char *str);
 
 int main(int argc, char **argv){
-        rtcInitLogger(RTC_LOG_DEBUG);
+	rtcInitLogger(RTC_LOG_DEBUG, NULL);
 
 	// Create peer
 	rtcConfiguration config;
@@ -263,16 +254,6 @@ static void deletePeer(Peer *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) {
@@ -299,10 +280,9 @@ char* state_print(rtcState state) {
 	}
 
 	return str;
-
 }
 
-char* rtcGatheringState_print(rtcState state) {
+char *rtcGatheringState_print(rtcGatheringState state) {
 	char *str = NULL;
 	switch (state) {
 	case RTC_GATHERING_NEW:
@@ -320,3 +300,13 @@ char* rtcGatheringState_print(rtcState state) {
 
 	return str;
 }
+
+int all_space(const char *str) {
+	while (*str) {
+		if (!isspace(*str++)) {
+			return 0;
+		}
+	}
+
+	return 1;
+}

examples/copy-paste/CMakeLists.txt

@@ -4,11 +4,19 @@ add_executable(datachannel-copy-paste-offerer offerer.cpp)
 set_target_properties(datachannel-copy-paste-offerer PROPERTIES
 	CXX_STANDARD 17
 	OUTPUT_NAME offerer)
+if(WIN32)
+	target_link_libraries(datachannel-copy-paste-offerer datachannel-static) # DLL exports only the C API
+else()
 	target_link_libraries(datachannel-copy-paste-offerer datachannel)
+endif()
 
 add_executable(datachannel-copy-paste-answerer answerer.cpp)
 set_target_properties(datachannel-copy-paste-answerer PROPERTIES
 	CXX_STANDARD 17
 	OUTPUT_NAME answerer)
+if(WIN32)
+	target_link_libraries(datachannel-copy-paste-answerer datachannel-static) # DLL exports only the C API
+else()
 	target_link_libraries(datachannel-copy-paste-answerer datachannel)
+endif()
 

include/rtc/datachannel.hpp

@@ -108,8 +108,8 @@ template <typename Iterator> bool DataChannel::sendBuffer(Iterator first, Iterat
 	auto message = std::make_shared<Message>(size);
 	auto pos = message->begin();
 	for (Iterator it = first; it != last; ++it) {
-		auto [bytes, size] = to_bytes(*it);
-		pos = std::copy(bytes, bytes + size, pos);
+		auto [bytes, len] = to_bytes(*it);
+		pos = std::copy(bytes, bytes + len, pos);
 	}
 	return outgoing(message);
 }

include/rtc/description.hpp

@@ -49,6 +49,7 @@ public:
 	bool ended() const;
 
 	void hintType(Type type);
+	void setDataMid(string mid);
 	void setFingerprint(string fingerprint);
 	void setSctpPort(uint16_t port);
 	void setMaxMessageSize(size_t size);
@@ -86,7 +87,7 @@ private:
 		string mid;
 		std::vector<string> attributes;
 	};
-	std::map<string, Media> mMedia; // by mid
+	std::map<int, Media> mMedia; // by m-line index
 
 	// Candidates
 	std::vector<Candidate> mCandidates;

include/rtc/include.hpp

@@ -64,6 +64,8 @@ const uint16_t DEFAULT_SCTP_PORT = 5000; // SCTP port to use by default
 const size_t DEFAULT_MAX_MESSAGE_SIZE = 65536;    // Remote max message size if not specified in SDP
 const size_t LOCAL_MAX_MESSAGE_SIZE = 256 * 1024; // Local max message size
 
+const int THREADPOOL_SIZE = 4; // Number of threads in the global thread pool
+
 // overloaded helper
 template <class... Ts> struct overloaded : Ts... { using Ts::operator()...; };
 template <class... Ts> overloaded(Ts...)->overloaded<Ts...>;

include/rtc/init.hpp

@@ -25,8 +25,7 @@
 
 namespace rtc {
 
-class Init;
-using init_token = std::shared_ptr<Init>;
+using init_token = std::shared_ptr<void>;
 
 class Init {
 public:
@@ -39,9 +38,10 @@ public:
 private:
 	Init();
 
-	static std::weak_ptr<Init> Weak;
-	static init_token Global;
-	static std::mutex Mutex;
+	static std::weak_ptr<void> Weak;
+	static std::shared_ptr<void> *Global;
+	static bool Initialized;
+	static std::recursive_mutex Mutex;
 };
 
 inline void Preload() { Init::Preload(); }

include/rtc/peerconnection.hpp

@@ -41,6 +41,7 @@
 namespace rtc {
 
 class Certificate;
+class Processor;
 class IceTransport;
 class DtlsTransport;
 class SctpTransport;
@@ -101,7 +102,7 @@ public:
 	// Media
 	bool hasMedia() const;
 	void sendMedia(const binary &packet);
-	void send(const byte *packet, size_t size);
+	void sendMedia(const byte *packet, size_t size);
 
 	void onMedia(std::function<void(const binary &packet)> callback);
 
@@ -139,13 +140,13 @@ private:
 
 	void outgoingMedia(message_ptr message);
 
+	const init_token mInitToken = Init::Token();
 	const Configuration mConfig;
 	const future_certificate_ptr mCertificate;
-
-	init_token mInitToken = Init::Token();
+	const std::unique_ptr<Processor> mProcessor;
 
 	std::optional<Description> mLocalDescription, mRemoteDescription;
-	mutable std::recursive_mutex mLocalDescriptionMutex, mRemoteDescriptionMutex;
+	mutable std::mutex mLocalDescriptionMutex, mRemoteDescriptionMutex;
 
 	std::shared_ptr<IceTransport> mIceTransport;
 	std::shared_ptr<DtlsTransport> mDtlsTransport;

include/rtc/rtc.h

@@ -23,9 +23,11 @@
 extern "C" {
 #endif
 
-#include <stdint.h>
-
-// libdatachannel C API
+#ifdef _WIN32
+#define RTC_EXPORT __declspec(dllexport)
+#else
+#define RTC_EXPORT
+#endif
 
 #ifndef RTC_ENABLE_MEDIA
 #define RTC_ENABLE_MEDIA 1
@@ -35,6 +37,10 @@ extern "C" {
 #define RTC_ENABLE_WEBSOCKET 1
 #endif
 
+#include <stdint.h>
+
+// libdatachannel C API
+
 typedef enum {
 	RTC_NEW = 0,
 	RTC_CONNECTING = 1,
@@ -71,71 +77,72 @@ typedef struct {
 	uint16_t portRangeEnd;
 } rtcConfiguration;
 
-typedef void (*dataChannelCallbackFunc)(int dc, void *ptr);
-typedef void (*descriptionCallbackFunc)(const char *sdp, const char *type, void *ptr);
-typedef void (*candidateCallbackFunc)(const char *cand, const char *mid, void *ptr);
-typedef void (*stateChangeCallbackFunc)(rtcState state, void *ptr);
-typedef void (*gatheringStateCallbackFunc)(rtcGatheringState state, void *ptr);
-typedef void (*openCallbackFunc)(void *ptr);
-typedef void (*closedCallbackFunc)(void *ptr);
-typedef void (*errorCallbackFunc)(const char *error, void *ptr);
-typedef void (*messageCallbackFunc)(const char *message, int size, void *ptr);
-typedef void (*bufferedAmountLowCallbackFunc)(void *ptr);
-typedef void (*availableCallbackFunc)(void *ptr);
+typedef void (*rtcLogCallbackFunc)(rtcLogLevel level, const char *message);
+typedef void (*rtcDataChannelCallbackFunc)(int dc, void *ptr);
+typedef void (*rtcDescriptionCallbackFunc)(const char *sdp, const char *type, void *ptr);
+typedef void (*rtcCandidateCallbackFunc)(const char *cand, const char *mid, void *ptr);
+typedef void (*rtcStateChangeCallbackFunc)(rtcState state, void *ptr);
+typedef void (*rtcGatheringStateCallbackFunc)(rtcGatheringState state, void *ptr);
+typedef void (*rtcOpenCallbackFunc)(void *ptr);
+typedef void (*rtcClosedCallbackFunc)(void *ptr);
+typedef void (*rtcErrorCallbackFunc)(const char *error, void *ptr);
+typedef void (*rtcMessageCallbackFunc)(const char *message, int size, void *ptr);
+typedef void (*rtcBufferedAmountLowCallbackFunc)(void *ptr);
+typedef void (*rtcAvailableCallbackFunc)(void *ptr);
 
 // Log
-void rtcInitLogger(rtcLogLevel level);
+RTC_EXPORT void rtcInitLogger(rtcLogLevel level, rtcLogCallbackFunc cb); // NULL cb to log to stdout
 
 // User pointer
-void rtcSetUserPointer(int id, void *ptr);
+RTC_EXPORT void rtcSetUserPointer(int id, void *ptr);
 
 // PeerConnection
-int rtcCreatePeerConnection(const rtcConfiguration *config); // returns pc id
-int rtcDeletePeerConnection(int pc);
+RTC_EXPORT int rtcCreatePeerConnection(const rtcConfiguration *config); // returns pc id
+RTC_EXPORT int rtcDeletePeerConnection(int pc);
 
-int rtcSetDataChannelCallback(int pc, dataChannelCallbackFunc cb);
-int rtcSetLocalDescriptionCallback(int pc, descriptionCallbackFunc cb);
-int rtcSetLocalCandidateCallback(int pc, candidateCallbackFunc cb);
-int rtcSetStateChangeCallback(int pc, stateChangeCallbackFunc cb);
-int rtcSetGatheringStateChangeCallback(int pc, gatheringStateCallbackFunc cb);
+RTC_EXPORT int rtcSetDataChannelCallback(int pc, rtcDataChannelCallbackFunc cb);
+RTC_EXPORT int rtcSetLocalDescriptionCallback(int pc, rtcDescriptionCallbackFunc cb);
+RTC_EXPORT int rtcSetLocalCandidateCallback(int pc, rtcCandidateCallbackFunc cb);
+RTC_EXPORT int rtcSetStateChangeCallback(int pc, rtcStateChangeCallbackFunc cb);
+RTC_EXPORT int rtcSetGatheringStateChangeCallback(int pc, rtcGatheringStateCallbackFunc cb);
 
-int rtcSetRemoteDescription(int pc, const char *sdp, const char *type);
-int rtcAddRemoteCandidate(int pc, const char *cand, const char *mid);
+RTC_EXPORT int rtcSetRemoteDescription(int pc, const char *sdp, const char *type);
+RTC_EXPORT int rtcAddRemoteCandidate(int pc, const char *cand, const char *mid);
 
-int rtcGetLocalAddress(int pc, char *buffer, int size);
-int rtcGetRemoteAddress(int pc, char *buffer, int size);
+RTC_EXPORT int rtcGetLocalAddress(int pc, char *buffer, int size);
+RTC_EXPORT int rtcGetRemoteAddress(int pc, char *buffer, int size);
 
 // DataChannel
-int rtcCreateDataChannel(int pc, const char *label); // returns dc id
-int rtcDeleteDataChannel(int dc);
+RTC_EXPORT int rtcCreateDataChannel(int pc, const char *label); // returns dc id
+RTC_EXPORT int rtcDeleteDataChannel(int dc);
 
-int rtcGetDataChannelLabel(int dc, char *buffer, int size);
+RTC_EXPORT int rtcGetDataChannelLabel(int dc, char *buffer, int size);
 
 // WebSocket
 #if RTC_ENABLE_WEBSOCKET
-int rtcCreateWebSocket(const char *url); // returns ws id
-int rtcDeleteWebsocket(int ws);
+RTC_EXPORT int rtcCreateWebSocket(const char *url); // returns ws id
+RTC_EXPORT int rtcDeleteWebsocket(int ws);
 #endif
 
 // DataChannel and WebSocket common API
-int rtcSetOpenCallback(int id, openCallbackFunc cb);
-int rtcSetClosedCallback(int id, closedCallbackFunc cb);
-int rtcSetErrorCallback(int id, errorCallbackFunc cb);
-int rtcSetMessageCallback(int id, messageCallbackFunc cb);
-int rtcSendMessage(int id, const char *data, int size);
+RTC_EXPORT int rtcSetOpenCallback(int id, rtcOpenCallbackFunc cb);
+RTC_EXPORT int rtcSetClosedCallback(int id, rtcClosedCallbackFunc cb);
+RTC_EXPORT int rtcSetErrorCallback(int id, rtcErrorCallbackFunc cb);
+RTC_EXPORT int rtcSetMessageCallback(int id, rtcMessageCallbackFunc cb);
+RTC_EXPORT int rtcSendMessage(int id, const char *data, int size);
 
-int rtcGetBufferedAmount(int id); // total size buffered to send
-int rtcSetBufferedAmountLowThreshold(int id, int amount);
-int rtcSetBufferedAmountLowCallback(int id, bufferedAmountLowCallbackFunc cb);
+RTC_EXPORT int rtcGetBufferedAmount(int id); // total size buffered to send
+RTC_EXPORT int rtcSetBufferedAmountLowThreshold(int id, int amount);
+RTC_EXPORT int rtcSetBufferedAmountLowCallback(int id, rtcBufferedAmountLowCallbackFunc cb);
 
 // DataChannel and WebSocket common extended API
-int rtcGetAvailableAmount(int id); // total size available to receive
-int rtcSetAvailableCallback(int id, availableCallbackFunc cb);
-int rtcReceiveMessage(int id, char *buffer, int *size);
+RTC_EXPORT int rtcGetAvailableAmount(int id); // total size available to receive
+RTC_EXPORT int rtcSetAvailableCallback(int id, rtcAvailableCallbackFunc cb);
+RTC_EXPORT int rtcReceiveMessage(int id, char *buffer, int *size);
 
 // Optional preload and cleanup
-void rtcPreload();
-void rtcCleanup();
+RTC_EXPORT void rtcPreload(void);
+RTC_EXPORT void rtcCleanup(void);
 
 #ifdef __cplusplus
 } // extern "C"

src/base64.cpp

@@ -42,7 +42,7 @@ string to_base64(const binary &data) {
 		i += 3;
 	}
 
-	int left = data.size() - i;
+	int left = int(data.size() - i);
 	if (left) {
 		auto d0 = to_integer<uint8_t>(data[i]);
 		out += tab[d0 >> 2];

src/candidate.cpp

@@ -28,6 +28,7 @@
 #else
 #include <netdb.h>
 #include <sys/socket.h>
+#include <netinet/in.h>
 #endif
 
 #include <sys/types.h>
@@ -98,8 +99,8 @@ bool Candidate::resolve(ResolveMode mode) {
 					// Rewrite the candidate
 					char nodebuffer[MAX_NUMERICNODE_LEN];
 					char servbuffer[MAX_NUMERICSERV_LEN];
-					if (getnameinfo(p->ai_addr, p->ai_addrlen, nodebuffer, MAX_NUMERICNODE_LEN,
-					                servbuffer, MAX_NUMERICSERV_LEN,
+					if (getnameinfo(p->ai_addr, socklen_t(p->ai_addrlen), nodebuffer,
+					                MAX_NUMERICNODE_LEN, servbuffer, MAX_NUMERICSERV_LEN,
 					                NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
 						const char sp{' '};
 						std::ostringstream oss;
@@ -107,8 +108,9 @@ bool Candidate::resolve(ResolveMode mode) {
 						oss << sp << nodebuffer << sp << servbuffer << sp << "typ" << sp << type;
 						oss << left;
 						mCandidate = oss.str();
+						mIsResolved = true;
 						PLOG_VERBOSE << "Resolved candidate: " << mCandidate;
-						return mIsResolved = true;
+						break;
 					}
 				}
 		}
@@ -116,7 +118,7 @@ bool Candidate::resolve(ResolveMode mode) {
 		freeaddrinfo(result);
 	}
 
-	return false;
+	return mIsResolved;
 }
 
 bool Candidate::isResolved() const { return mIsResolved; }

src/certificate.cpp

@@ -17,6 +17,7 @@
  */
 
 #include "certificate.hpp"
+#include "threadpool.hpp"
 
 #include <cassert>
 #include <chrono>
@@ -132,14 +133,14 @@ namespace rtc {
 
 Certificate::Certificate(string crt_pem, string key_pem) {
 	BIO *bio = BIO_new(BIO_s_mem());
-	BIO_write(bio, crt_pem.data(), crt_pem.size());
+	BIO_write(bio, crt_pem.data(), int(crt_pem.size()));
 	mX509 = shared_ptr<X509>(PEM_read_bio_X509(bio, nullptr, 0, 0), X509_free);
 	BIO_free(bio);
 	if (!mX509)
 		throw std::invalid_argument("Unable to import certificate PEM");
 
 	bio = BIO_new(BIO_s_mem());
-	BIO_write(bio, key_pem.data(), key_pem.size());
+	BIO_write(bio, key_pem.data(), int(key_pem.size()));
 	mPKey = shared_ptr<EVP_PKEY>(PEM_read_bio_PrivateKey(bio, nullptr, 0, 0), EVP_PKEY_free);
 	BIO_free(bio);
 	if (!mPKey)
@@ -201,8 +202,8 @@ certificate_ptr make_certificate_impl(string commonName) {
 	auto *commonNameBytes =
 	    reinterpret_cast<unsigned char *>(const_cast<char *>(commonName.c_str()));
 
-	if (!X509_gmtime_adj(X509_get_notBefore(x509.get()), 3600 * -1) ||
-	    !X509_gmtime_adj(X509_get_notAfter(x509.get()), 3600 * 24 * 365) ||
+	if (!X509_gmtime_adj(X509_getm_notBefore(x509.get()), 3600 * -1) ||
+	    !X509_gmtime_adj(X509_getm_notAfter(x509.get()), 3600 * 24 * 365) ||
 	    !X509_set_version(x509.get(), 1) || !X509_set_pubkey(x509.get(), pkey.get()) ||
 	    !BN_pseudo_rand(serial_number.get(), serialSize, 0, 0) ||
 	    !BN_to_ASN1_INTEGER(serial_number.get(), X509_get_serialNumber(x509.get())) ||
@@ -230,19 +231,6 @@ 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::invoke_result_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;
 
@@ -254,7 +242,7 @@ future_certificate_ptr make_certificate(string commonName) {
 	if (auto it = CertificateCache.find(commonName); it != CertificateCache.end())
 		return it->second;
 
-	auto future = thread_call(make_certificate_impl, commonName);
+	auto future = ThreadPool::Instance().enqueue(make_certificate_impl, commonName);
 	auto shared = future.share();
 	CertificateCache.emplace(std::move(commonName), shared);
 	return shared;

src/datachannel.cpp

@@ -96,7 +96,7 @@ void DataChannel::close() {
 	mIsClosed = true;
 	if (mIsOpen.exchange(false))
 		if (auto transport = mSctpTransport.lock())
-			transport->close(mStream);
+			transport->closeStream(mStream);
 
 	mSctpTransport.reset();
 	resetCallbacks();
@@ -186,8 +186,8 @@ void DataChannel::open(shared_ptr<SctpTransport> transport) {
 	open.channelType = mReliability->type;
 	open.priority = htons(0);
 	open.reliabilityParameter = htonl(reliabilityParameter);
-	open.labelLength = htons(mLabel.size());
-	open.protocolLength = htons(mProtocol.size());
+	open.labelLength = htons(uint16_t(mLabel.size()));
+	open.protocolLength = htons(uint16_t(mProtocol.size()));
 
 	auto end = reinterpret_cast<char *>(buffer.data() + sizeof(OpenMessage));
 	std::copy(mLabel.begin(), mLabel.end(), end);

src/description.cpp

@@ -26,6 +26,7 @@
 
 using std::size_t;
 using std::string;
+using std::chrono::system_clock;
 
 namespace {
 
@@ -54,7 +55,7 @@ Description::Description(const string &sdp, Type type, Role role)
 	mData.mid = "data";
 	hintType(type);
 
-	auto seed = std::chrono::system_clock::now().time_since_epoch().count();
+	auto seed = static_cast<unsigned int>(system_clock::now().time_since_epoch().count());
 	std::default_random_engine generator(seed);
 	std::uniform_int_distribution<uint32_t> uniform;
 	mSessionId = std::to_string(uniform(generator));
@@ -62,6 +63,7 @@ Description::Description(const string &sdp, Type type, Role role)
 	std::istringstream ss(sdp);
 	std::optional<Media> currentMedia;
 
+	int mlineIndex = 0;
 	bool finished;
 	do {
 		string line;
@@ -75,7 +77,9 @@ Description::Description(const string &sdp, Type type, Role role)
 					if (currentMedia->type == "application")
 						mData.mid = currentMedia->mid;
 					else
-						mMedia.emplace(currentMedia->mid, std::move(*currentMedia));
+						mMedia.emplace(mlineIndex, std::move(*currentMedia));
+
+					++mlineIndex;
 
 				} else if (line.find(" ICE/SDP") != string::npos) {
 					PLOG_WARNING << "SDP \"m=\" line has no corresponding mid, ignoring";
@@ -112,7 +116,8 @@ Description::Description(const string &sdp, Type type, Role role)
 				if (match_prefix(value, "sha-256 ")) {
 					mFingerprint = value.substr(8);
 					std::transform(mFingerprint->begin(), mFingerprint->end(),
-					               mFingerprint->begin(), [](char c) { return std::toupper(c); });
+					               mFingerprint->begin(),
+					               [](char c) { return char(std::toupper(c)); });
 				} else {
 					PLOG_WARNING << "Unknown SDP fingerprint type: " << value;
 				}
@@ -161,6 +166,8 @@ void Description::hintType(Type type) {
 	}
 }
 
+void Description::setDataMid(string mid) { mData.mid = mid; }
+
 void Description::setFingerprint(string fingerprint) {
 	mFingerprint.emplace(std::move(fingerprint));
 }
@@ -185,11 +192,8 @@ std::vector<Candidate> Description::extractCandidates() {
 bool Description::hasMedia() const { return !mMedia.empty(); }
 
 void Description::addMedia(const Description &source) {
-	for (auto [mid, media] : source.mMedia)
-		if (mid != mData.mid)
-			mMedia.emplace(mid, media);
-		else
-			PLOG_WARNING << "Media mid \"" << mid << "\" is the same as data mid, ignoring";
+	for (auto p : source.mMedia)
+		mMedia.emplace(p);
 }
 
 Description::operator string() const { return generateSdp("\r\n"); }
@@ -210,13 +214,40 @@ string Description::generateSdp(const string &eol) const {
 	// see Negotiating Media Multiplexing Using the Session Description Protocol
 	// https://tools.ietf.org/html/draft-ietf-mmusic-sdp-bundle-negotiation-54
 	sdp << "a=group:BUNDLE";
-	for (const auto &[mid, _] : mMedia)
-		sdp << " " << mid;
-	sdp << " " << mData.mid << eol;
+	for (int i = 0; i < int(mMedia.size() + 1); ++i)
+		if (auto it = mMedia.find(i); it != mMedia.end())
+			sdp << ' ' << it->second.mid;
+		else
+			sdp << ' ' << mData.mid;
+	sdp << eol;
 
+	sdp << "a=msid-semantic: WMS" << eol;
+
+	// Non-data media
+	if (!mMedia.empty()) {
+		// Lip-sync
+		sdp << "a=group:LS";
+		for (const auto &p : mMedia)
+			sdp << " " << p.second.mid;
+		sdp << eol;
+	}
+
+	// Descriptions and attributes
+	for (int i = 0; i < int(mMedia.size() + 1); ++i) {
+		if (auto it = mMedia.find(i); it != mMedia.end()) {
+			// Non-data media
+			const auto &media = it->second;
+			sdp << "m=" << media.type << ' ' << 0 << ' ' << media.description << eol;
+			sdp << "c=IN IP4 0.0.0.0" << eol;
+			sdp << "a=bundle-only" << eol;
+			sdp << "a=mid:" << media.mid << eol;
+			for (const auto &attr : media.attributes)
+				sdp << "a=" << attr << eol;
+
+		} else {
 			// Data
-	const string dataDescription = "UDP/DTLS/SCTP webrtc-datachannel";
-	sdp << "m=application" << ' ' << (!mMedia.empty() ? 0 : 9) << ' ' << dataDescription << eol;
+			const string description = "UDP/DTLS/SCTP webrtc-datachannel";
+			sdp << "m=application" << ' ' << (!mMedia.empty() ? 0 : 9) << ' ' << description << eol;
 			sdp << "c=IN IP4 0.0.0.0" << eol;
 			if (!mMedia.empty())
 				sdp << "a=bundle-only" << eol;
@@ -225,32 +256,18 @@ string Description::generateSdp(const string &eol) const {
 				sdp << "a=sctp-port:" << *mData.sctpPort << eol;
 			if (mData.maxMessageSize)
 				sdp << "a=max-message-size:" << *mData.maxMessageSize << eol;
-
-	// Non-data media
-	if (!mMedia.empty()) {
-		// Lip-sync
-		sdp << "a=group:LS";
-		for (const auto &[mid, _] : mMedia)
-			sdp << " " << mid;
-		sdp << eol;
-
-		// Descriptions and attributes
-		for (const auto &[_, media] : mMedia) {
-			sdp << "m=" << media.type << ' ' << 0 << ' ' << media.description << eol;
-			sdp << "c=IN IP4 0.0.0.0" << eol;
-			sdp << "a=bundle-only" << eol;
-			sdp << "a=mid:" << media.mid << eol;
-			for (const auto &attr : media.attributes)
-				sdp << "a=" << attr << eol;
 		}
 	}
 
 	// Common
+	if (!mEnded)
 		sdp << "a=ice-options:trickle" << eol;
+
 	sdp << "a=ice-ufrag:" << mIceUfrag << eol;
 	sdp << "a=ice-pwd:" << mIcePwd << eol;
 	sdp << "a=setup:" << roleToString(mRole) << eol;
-	sdp << "a=dtls-id:1" << eol;
+	sdp << "a=tls-id:1" << eol;
+
 	if (mFingerprint)
 		sdp << "a=fingerprint:sha-256 " << *mFingerprint << eol;
 

src/dtlssrtptransport.cpp

@@ -43,37 +43,65 @@ DtlsSrtpTransport::DtlsSrtpTransport(std::shared_ptr<IceTransport> lower,
                     std::move(stateChangeCallback)),
       mSrtpRecvCallback(std::move(srtpRecvCallback)) { // distinct from Transport recv callback
 
-	PLOG_DEBUG << "Initializing SRTP transport";
+	PLOG_DEBUG << "Initializing DTLS-SRTP transport";
 
-#if USE_GNUTLS
-	PLOG_DEBUG << "Initializing DTLS-SRTP transport (GnuTLS)";
-	gnutls::check(gnutls_srtp_set_profile(mSession, GNUTLS_SRTP_AES128_CM_HMAC_SHA1_80),
-	              "Failed to set SRTP profile");
-#else
-	PLOG_DEBUG << "Initializing DTLS-SRTP transport (OpenSSL)";
-	openssl::check(SSL_set_tlsext_use_srtp(mSsl, "SRTP_AES128_CM_SHA1_80"),
-	               "Failed to set SRTP profile");
-#endif
+	if (srtp_err_status_t err = srtp_create(&mSrtpIn, nullptr)) {
+		throw std::runtime_error("SRTP create failed, status=" + to_string(static_cast<int>(err)));
+	}
+	if (srtp_err_status_t err = srtp_create(&mSrtpOut, nullptr)) {
+		srtp_dealloc(mSrtpIn);
+		throw std::runtime_error("SRTP create failed, status=" + to_string(static_cast<int>(err)));
+	}
 }
 
 DtlsSrtpTransport::~DtlsSrtpTransport() {
 	stop();
 
-	if (mCreated)
-		srtp_dealloc(mSrtp);
+	srtp_dealloc(mSrtpIn);
+	srtp_dealloc(mSrtpOut);
 }
 
-bool DtlsSrtpTransport::send(message_ptr message) {
+bool DtlsSrtpTransport::sendMedia(message_ptr message) {
 	if (!message)
 		return false;
 
+	if (!mInitDone) {
+		PLOG_WARNING << "SRTP media sent before keys are derived";
+		return false;
+	}
+
 	int size = message->size();
 	PLOG_VERBOSE << "Send size=" << size;
 
-	// srtp_protect() assumes that it can write SRTP_MAX_TRAILER_LEN (for the authentication tag)
-	// into the location in memory immediately following the RTP packet.
+	// The RTP header has a minimum size of 12 bytes
+	if (size < 12)
+		throw std::runtime_error("RTP/RTCP packet too short");
+
+	// srtp_protect() and srtp_protect_rtcp() assume that they can write SRTP_MAX_TRAILER_LEN (for
+	// the authentication tag) into the location in memory immediately following the RTP packet.
 	message->resize(size + SRTP_MAX_TRAILER_LEN);
-	if (srtp_err_status_t err = srtp_protect(mSrtp, message->data(), &size)) {
+
+	uint8_t value2 = to_integer<uint8_t>(*(message->begin() + 1)) & 0x7F;
+	PLOG_VERBOSE << "Demultiplexing SRTCP and SRTP with RTP payload type, value="
+	             << unsigned(value2);
+
+	// RFC 5761 Multiplexing RTP and RTCP 4. Distinguishable RTP and RTCP Packets
+	// It is RECOMMENDED to follow the guidelines in the RTP/AVP profile for the choice of RTP
+	// payload type values, with the additional restriction that payload type values in the
+	// range 64-95 MUST NOT be used. Specifically, dynamic RTP payload types SHOULD be chosen in
+	// the range 96-127 where possible. Values below 64 MAY be used if that is insufficient
+	// [...]
+	if (value2 >= 64 && value2 <= 95) { // Range 64-95 (inclusive) MUST be RTCP
+		if (srtp_err_status_t err = srtp_protect_rtcp(mSrtpOut, message->data(), &size)) {
+			if (err == srtp_err_status_replay_fail)
+				throw std::runtime_error("SRTCP packet is a replay");
+			else
+				throw std::runtime_error("SRTCP protect error, status=" +
+				                         to_string(static_cast<int>(err)));
+		}
+		PLOG_VERBOSE << "Protected SRTCP packet, size=" << size;
+	} else {
+		if (srtp_err_status_t err = srtp_protect(mSrtpOut, message->data(), &size)) {
 			if (err == srtp_err_status_replay_fail)
 				throw std::runtime_error("SRTP packet is a replay");
 			else
@@ -81,12 +109,20 @@ bool DtlsSrtpTransport::send(message_ptr message) {
 				                         to_string(static_cast<int>(err)));
 		}
 		PLOG_VERBOSE << "Protected SRTP packet, size=" << size;
+	}
+
 	message->resize(size);
 	outgoing(message);
 	return true;
 }
 
 void DtlsSrtpTransport::incoming(message_ptr message) {
+	if (!mInitDone) {
+		// Bypas
+		DtlsTransport::incoming(message);
+		return;
+	}
+
 	int size = message->size();
 	if (size == 0)
 		return;
@@ -95,15 +131,39 @@ void DtlsSrtpTransport::incoming(message_ptr message) {
 	// The process for demultiplexing a packet is as follows. The receiver looks at the first byte
 	// of the packet. [...] If the value is in between 128 and 191 (inclusive), then the packet is
 	// RTP (or RTCP [...]). If the value is between 20 and 63 (inclusive), the packet is DTLS.
-	uint8_t value = to_integer<uint8_t>(*message->begin());
+	uint8_t value1 = to_integer<uint8_t>(*message->begin());
+	PLOG_VERBOSE << "Demultiplexing DTLS and SRTP/SRTCP with first byte, value="
+	             << unsigned(value1);
 
-	if (value >= 128 && value <= 192) {
+	if (value1 >= 20 && value1 <= 63) {
 		PLOG_VERBOSE << "Incoming DTLS packet, size=" << size;
 		DtlsTransport::incoming(message);
-	} else if (value >= 20 && value <= 64) {
-		PLOG_VERBOSE << "Incoming SRTP packet, size=" << size;
 
-		if (srtp_err_status_t err = srtp_unprotect(mSrtp, message->data(), &size)) {
+	} else if (value1 >= 128 && value1 <= 191) {
+		// The RTP header has a minimum size of 12 bytes
+		if (size < 12) {
+			PLOG_WARNING << "Incoming SRTP/SRTCP packet too short, size=" << size;
+			return;
+		}
+
+		uint8_t value2 = to_integer<uint8_t>(*(message->begin() + 1)) & 0x7F;
+		PLOG_VERBOSE << "Demultiplexing SRTCP and SRTP with RTP payload type, value="
+		             << unsigned(value2);
+
+		// See RFC 5761 reference above
+		if (value2 >= 64 && value2 <= 95) { // Range 64-95 (inclusive) MUST be RTCP
+			PLOG_VERBOSE << "Incoming SRTCP packet, size=" << size;
+			if (srtp_err_status_t err = srtp_unprotect_rtcp(mSrtpIn, message->data(), &size)) {
+				if (err == srtp_err_status_replay_fail)
+					PLOG_WARNING << "Incoming SRTCP packet is a replay";
+				else
+					PLOG_WARNING << "SRTCP unprotect error, status=" << err;
+				return;
+			}
+			PLOG_VERBOSE << "Unprotected SRTCP packet, size=" << size;
+		} else {
+			PLOG_VERBOSE << "Incoming SRTP packet, size=" << size;
+			if (srtp_err_status_t err = srtp_unprotect(mSrtpIn, message->data(), &size)) {
 				if (err == srtp_err_status_replay_fail)
 					PLOG_WARNING << "Incoming SRTP packet is a replay";
 				else
@@ -111,33 +171,40 @@ void DtlsSrtpTransport::incoming(message_ptr message) {
 				return;
 			}
 			PLOG_VERBOSE << "Unprotected SRTP packet, size=" << size;
+		}
+
 		message->resize(size);
 		mSrtpRecvCallback(message);
 
 	} else {
-		PLOG_WARNING << "Unknown packet type, value=" << value << ", size=" << size;
+		PLOG_WARNING << "Unknown packet type, value=" << unsigned(value1) << ", size=" << size;
 	}
 }
 
+void DtlsSrtpTransport::postCreation() {
+#if USE_GNUTLS
+	PLOG_DEBUG << "Setting SRTP profile (GnuTLS)";
+	gnutls::check(gnutls_srtp_set_profile(mSession, GNUTLS_SRTP_AES128_CM_HMAC_SHA1_80),
+	              "Failed to set SRTP profile");
+#else
+	PLOG_DEBUG << "Setting SRTP profile (OpenSSL)";
+	// returns 0 on success, 1 on error
+	if (SSL_set_tlsext_use_srtp(mSsl, "SRTP_AES128_CM_SHA1_80"), "Failed to set SRTP profile")
+		throw std::runtime_error("Failed to set SRTP profile: " + openssl::error_string(ERR_get_error()));
+#endif
+}
+
 void DtlsSrtpTransport::postHandshake() {
-	if (mCreated)
+	if (mInitDone)
 		return;
 
-	srtp_policy_t inbound = {};
-	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtp);
-	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtcp);
-	inbound.ssrc.type = ssrc_any_inbound;
-
-	srtp_policy_t outbound = {};
-	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&outbound.rtp);
-	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&outbound.rtcp);
-	outbound.ssrc.type = ssrc_any_outbound;
-
 	const size_t materialLen = SRTP_AES_ICM_128_KEY_LEN_WSALT * 2;
 	unsigned char material[materialLen];
 	const unsigned char *clientKey, *clientSalt, *serverKey, *serverSalt;
 
 #if USE_GNUTLS
+	PLOG_INFO << "Deriving SRTP keying material (GnuTLS)";
+
 	gnutls_datum_t clientKeyDatum, clientSaltDatum, serverKeyDatum, serverSaltDatum;
 	gnutls::check(gnutls_srtp_get_keys(mSession, material, materialLen, &clientKeyDatum,
 	                                   &clientSaltDatum, &serverKeyDatum, &serverSaltDatum),
@@ -160,18 +227,23 @@ void DtlsSrtpTransport::postHandshake() {
 	serverKey = reinterpret_cast<const unsigned char *>(serverKeyDatum.data);
 	serverSalt = reinterpret_cast<const unsigned char *>(serverSaltDatum.data);
 #else
-	// This provides the client write master key, the server write master key, the client write
-	// master salt and the server write master salt in that order.
+	PLOG_INFO << "Deriving SRTP keying material (OpenSSL)";
+
+	// The extractor provides the client write master key, the server write master key, the client
+	// write master salt and the server write master salt in that order.
 	const string label = "EXTRACTOR-dtls_srtp";
-	openssl::check(SSL_export_keying_material(mSsl, material, materialLen, label.c_str(),
-	                                          label.size(), nullptr, 0, 0),
-	               "Failed to derive SRTP keys");
+
+	// returns 1 on success, 0 or -1 on failure (OpenSSL API is a complete mess...)
+	if (SSL_export_keying_material(mSsl, material, materialLen, label.c_str(), label.size(),
+	                               nullptr, 0, 0) <= 0)
+		throw std::runtime_error("Failed to derive SRTP keys: " +
+		                         openssl::error_string(ERR_get_error()));
 
 	clientKey = material;
 	clientSalt = clientKey + SRTP_AES_128_KEY_LEN;
 
 	serverKey = material + SRTP_AES_ICM_128_KEY_LEN_WSALT;
-	serverSalt = serverSalt + SRTP_AES_128_KEY_LEN;
+	serverSalt = serverKey + SRTP_AES_128_KEY_LEN;
 #endif
 
 	unsigned char clientSessionKey[SRTP_AES_ICM_128_KEY_LEN_WSALT];
@@ -182,22 +254,31 @@ void DtlsSrtpTransport::postHandshake() {
 	std::memcpy(serverSessionKey, serverKey, SRTP_AES_128_KEY_LEN);
 	std::memcpy(serverSessionKey + SRTP_AES_128_KEY_LEN, serverSalt, SRTP_SALT_LEN);
 
-	if (mIsClient) {
-		inbound.key = serverSessionKey;
-		outbound.key = clientSessionKey;
-	} else {
-		inbound.key = clientSessionKey;
-		outbound.key = serverSessionKey;
-	}
+	srtp_policy_t inbound = {};
+	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtp);
+	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtcp);
+	inbound.ssrc.type = ssrc_any_inbound;
+	inbound.ssrc.value = 0;
+	inbound.key = mIsClient ? serverSessionKey : clientSessionKey;
+	inbound.next = nullptr;
 
-	srtp_policy_t *policies = &inbound;
-	inbound.next = &outbound;
+	if (srtp_err_status_t err = srtp_add_stream(mSrtpIn, &inbound))
+		throw std::runtime_error("SRTP add inbound stream failed, status=" +
+		                         to_string(static_cast<int>(err)));
+
+	srtp_policy_t outbound = {};
+	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&outbound.rtp);
+	srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&outbound.rtcp);
+	outbound.ssrc.type = ssrc_any_outbound;
+	outbound.ssrc.value = 0;
+	outbound.key = mIsClient ? clientSessionKey : serverSessionKey;
 	outbound.next = nullptr;
 
-	if (srtp_err_status_t err = srtp_create(&mSrtp, policies))
-		throw std::runtime_error("SRTP create failed, status=" + to_string(static_cast<int>(err)));
+	if (srtp_err_status_t err = srtp_add_stream(mSrtpOut, &outbound))
+		throw std::runtime_error("SRTP add outbound stream failed, status=" +
+		                         to_string(static_cast<int>(err)));
 
-	mCreated = true;
+	mInitDone = true;
 }
 
 } // namespace rtc

src/dtlssrtptransport.hpp

@@ -38,16 +38,17 @@ public:
 	                  state_callback stateChangeCallback);
 	~DtlsSrtpTransport();
 
-	bool send(message_ptr message) override;
+	bool sendMedia(message_ptr message);
 
 private:
 	void incoming(message_ptr message) override;
+	void postCreation() override;
 	void postHandshake() override;
 
 	message_callback mSrtpRecvCallback;
 
-	srtp_t mSrtp;
-	bool mCreated = false;
+	srtp_t mSrtpIn, mSrtpOut;
+	bool mInitDone = false;
 };
 
 } // namespace rtc

src/dtlstransport.cpp

@@ -24,6 +24,14 @@
 #include <exception>
 #include <iostream>
 
+#if !USE_GNUTLS
+#ifdef _WIN32
+#include <winsock2.h> // for timeval
+#else
+#include <sys/time.h> // for timeval
+#endif
+#endif
+
 using namespace std::chrono;
 
 using std::shared_ptr;
@@ -77,6 +85,8 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, certificate_ptr cer
 		gnutls_transport_set_pull_function(mSession, ReadCallback);
 		gnutls_transport_set_pull_timeout_function(mSession, TimeoutCallback);
 
+		postCreation();
+
 		mRecvThread = std::thread(&DtlsTransport::runRecvLoop, this);
 		registerIncoming();
 
@@ -129,6 +139,10 @@ void DtlsTransport::incoming(message_ptr message) {
 	mIncomingQueue.push(message);
 }
 
+void DtlsTransport::postCreation() {
+	// Dummy
+}
+
 void DtlsTransport::postHandshake() {
 	// Dummy
 }
@@ -300,7 +314,8 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 	PLOG_DEBUG << "Initializing DTLS transport (OpenSSL)";
 
 	try {
-		if (!(mCtx = SSL_CTX_new(DTLS_method())))
+		mCtx = SSL_CTX_new(DTLS_method());
+		if (!mCtx)
 			throw std::runtime_error("Failed to create SSL context");
 
 		openssl::check(SSL_CTX_set_cipher_list(mCtx, "ALL:!LOW:!EXP:!RC4:!MD5:@STRENGTH"),
@@ -324,7 +339,8 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 
 		openssl::check(SSL_CTX_check_private_key(mCtx), "SSL local private key check failed");
 
-		if (!(mSsl = SSL_new(mCtx)))
+		mSsl = SSL_new(mCtx);
+		if (!mSsl)
 			throw std::runtime_error("Failed to create SSL instance");
 
 		SSL_set_ex_data(mSsl, TransportExIndex, this);
@@ -334,7 +350,9 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 		else
 			SSL_set_accept_state(mSsl);
 
-		if (!(mInBio = BIO_new(BIO_s_mem())) || !(mOutBio = BIO_new(BioMethods)))
+		mInBio = BIO_new(BIO_s_mem());
+		mOutBio = BIO_new(BioMethods);
+		if (!mInBio || !mOutBio)
 			throw std::runtime_error("Failed to create BIO");
 
 		BIO_set_mem_eof_return(mInBio, BIO_EOF);
@@ -382,7 +400,7 @@ bool DtlsTransport::send(message_ptr message) {
 
 	PLOG_VERBOSE << "Send size=" << message->size();
 
-	int ret = SSL_write(mSsl, message->data(), message->size());
+	int ret = SSL_write(mSsl, message->data(), int(message->size()));
 	return openssl::check(mSsl, ret);
 }
 
@@ -396,6 +414,10 @@ void DtlsTransport::incoming(message_ptr message) {
 	mIncomingQueue.push(message);
 }
 
+void DtlsTransport::postCreation() {
+	// Dummy
+}
+
 void DtlsTransport::postHandshake() {
 	// Dummy
 }
@@ -416,11 +438,11 @@ void DtlsTransport::runRecvLoop() {
 			// Process pending messages
 			while (!mIncomingQueue.empty()) {
 				auto message = *mIncomingQueue.pop();
-				BIO_write(mInBio, message->data(), message->size());
+				BIO_write(mInBio, message->data(), int(message->size()));
 
 				if (state() == State::Connecting) {
 					// Continue the handshake
-					int ret = SSL_do_handshake(mSsl);
+					ret = SSL_do_handshake(mSsl);
 					if (!openssl::check(mSsl, ret, "Handshake failed"))
 						break;
 
@@ -434,7 +456,7 @@ void DtlsTransport::runRecvLoop() {
 						postHandshake();
 					}
 				} else {
-					int ret = SSL_read(mSsl, buffer, bufferSize);
+					ret = SSL_read(mSsl, buffer, bufferSize);
 					if (!openssl::check(mSsl, ret))
 						break;
 
@@ -447,7 +469,7 @@ void DtlsTransport::runRecvLoop() {
 			std::optional<milliseconds> duration;
 			if (state() == State::Connecting) {
 				// Warning: This function breaks the usual return value convention
-				int ret = DTLSv1_handle_timeout(mSsl);
+				ret = DTLSv1_handle_timeout(mSsl);
 				if (ret < 0) {
 					throw std::runtime_error("Handshake timeout"); // write BIO can't fail
 				} else if (ret > 0) {
@@ -486,7 +508,7 @@ void DtlsTransport::runRecvLoop() {
 	}
 }
 
-int DtlsTransport::CertificateCallback(int preverify_ok, X509_STORE_CTX *ctx) {
+int DtlsTransport::CertificateCallback(int /*preverify_ok*/, X509_STORE_CTX *ctx) {
 	SSL *ssl =
 	    static_cast<SSL *>(X509_STORE_CTX_get_ex_data(ctx, SSL_get_ex_data_X509_STORE_CTX_idx()));
 	DtlsTransport *t =
@@ -535,7 +557,7 @@ int DtlsTransport::BioMethodWrite(BIO *bio, const char *in, int inl) {
 	return inl; // can't fail
 }
 
-long DtlsTransport::BioMethodCtrl(BIO *bio, int cmd, long num, void *ptr) {
+long DtlsTransport::BioMethodCtrl(BIO * /*bio*/, int cmd, long /*num*/, void * /*ptr*/) {
 	switch (cmd) {
 	case BIO_CTRL_FLUSH:
 		return 1;

src/dtlstransport.hpp

@@ -52,6 +52,7 @@ public:
 
 protected:
 	virtual void incoming(message_ptr message) override;
+	virtual void postCreation();
 	virtual void postHandshake();
 	void runRecvLoop();
 

src/icetransport.cpp

@@ -40,6 +40,7 @@ using namespace std::chrono_literals;
 
 using std::shared_ptr;
 using std::weak_ptr;
+using std::chrono::system_clock;
 
 #if USE_JUICE
 
@@ -69,7 +70,7 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 	// Randomize servers order
 	std::vector<IceServer> servers = config.iceServers;
-	unsigned seed = std::chrono::system_clock::now().time_since_epoch().count();
+	auto seed = static_cast<unsigned int>(system_clock::now().time_since_epoch().count());
 	std::shuffle(servers.begin(), servers.end(), std::default_random_engine(seed));
 
 	// Pick a STUN server (TURN support is not implemented in libjuice yet)
@@ -82,7 +83,7 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 			mStunHostname = server.hostname;
 			mStunService = server.service;
 			jconfig.stun_server_host = mStunHostname.c_str();
-			jconfig.stun_server_port = std::stoul(mStunService);
+			jconfig.stun_server_port = uint16_t(std::stoul(mStunService));
 			break;
 		}
 	}
@@ -206,7 +207,7 @@ void IceTransport::processCandidate(const string &candidate) {
 
 void IceTransport::processGatheringDone() { changeGatheringState(GatheringState::Complete); }
 
-void IceTransport::StateChangeCallback(juice_agent_t *agent, juice_state_t state, void *user_ptr) {
+void IceTransport::StateChangeCallback(juice_agent_t *, juice_state_t state, void *user_ptr) {
 	auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
 	try {
 		iceTransport->processStateChange(static_cast<unsigned int>(state));
@@ -215,7 +216,7 @@ void IceTransport::StateChangeCallback(juice_agent_t *agent, juice_state_t state
 	}
 }
 
-void IceTransport::CandidateCallback(juice_agent_t *agent, const char *sdp, void *user_ptr) {
+void IceTransport::CandidateCallback(juice_agent_t *, const char *sdp, void *user_ptr) {
 	auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
 	try {
 		iceTransport->processCandidate(sdp);
@@ -224,7 +225,7 @@ void IceTransport::CandidateCallback(juice_agent_t *agent, const char *sdp, void
 	}
 }
 
-void IceTransport::GatheringDoneCallback(juice_agent_t *agent, void *user_ptr) {
+void IceTransport::GatheringDoneCallback(juice_agent_t *, void *user_ptr) {
 	auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
 	try {
 		iceTransport->processGatheringDone();
@@ -233,8 +234,7 @@ void IceTransport::GatheringDoneCallback(juice_agent_t *agent, void *user_ptr) {
 	}
 }
 
-void IceTransport::RecvCallback(juice_agent_t *agent, const char *data, size_t size,
-                                void *user_ptr) {
+void IceTransport::RecvCallback(juice_agent_t *, const char *data, size_t size, void *user_ptr) {
 	auto iceTransport = static_cast<rtc::IceTransport *>(user_ptr);
 	try {
 		PLOG_VERBOSE << "Incoming size=" << size;
@@ -337,7 +337,7 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 	// Randomize order
 	std::vector<IceServer> servers = config.iceServers;
-	unsigned seed = std::chrono::system_clock::now().time_since_epoch().count();
+	auto seed = static_cast<unsigned int>(system_clock::now().time_since_epoch().count());
 	std::shuffle(servers.begin(), servers.end(), std::default_random_engine(seed));
 
 	// Add one STUN server

src/init.cpp

@@ -21,6 +21,7 @@
 #include "certificate.hpp"
 #include "dtlstransport.hpp"
 #include "sctptransport.hpp"
+#include "threadpool.hpp"
 #include "tls.hpp"
 
 #if RTC_ENABLE_WEBSOCKET
@@ -39,36 +40,19 @@ using std::shared_ptr;
 
 namespace rtc {
 
-std::weak_ptr<Init> Init::Weak;
-init_token Init::Global;
-std::mutex Init::Mutex;
+namespace {
 
-init_token Init::Token() {
-	std::lock_guard lock(Mutex);
+void doInit() {
+	PLOG_DEBUG << "Global initialization";
 
-	if (!Global) {
-		if (auto token = Weak.lock())
-			Global = token;
-		else
-			Global = shared_ptr<Init>(new Init());
-	}
-	return Global;
-}
-
-void Init::Preload() {
-	Token();                   // pre-init
-	make_certificate().wait(); // preload certificate
-}
-
-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
 
+	ThreadPool::Instance().spawn(THREADPOOL_SIZE);
+
 #if USE_GNUTLS
 	// Nothing to do
 #else
@@ -85,8 +69,12 @@ Init::Init() {
 #endif
 }
 
-Init::~Init() {
+void doCleanup() {
+	PLOG_DEBUG << "Global cleanup";
+
+	ThreadPool::Instance().join();
 	CleanupCertificateCache();
+
 	SctpTransport::Cleanup();
 	DtlsTransport::Cleanup();
 #if RTC_ENABLE_WEBSOCKET
@@ -101,5 +89,57 @@ Init::~Init() {
 #endif
 }
 
+} // namespace
+
+std::weak_ptr<void> Init::Weak;
+std::shared_ptr<void> *Init::Global = nullptr;
+bool Init::Initialized = false;
+std::recursive_mutex Init::Mutex;
+
+init_token Init::Token() {
+	std::unique_lock lock(Mutex);
+	if (auto token = Weak.lock())
+		return token;
+
+	delete Global;
+	Global = new shared_ptr<void>(new Init());
+	Weak = *Global;
+	return *Global;
+}
+
+void Init::Preload() {
+	std::unique_lock lock(Mutex);
+	auto token = Token();
+	if (!Global)
+		Global = new shared_ptr<void>(token);
+
+	PLOG_DEBUG << "Preloading certificate";
+	make_certificate().wait();
+}
+
+void Init::Cleanup() {
+	std::unique_lock lock(Mutex);
+	delete Global;
+	Global = nullptr;
+}
+
+Init::Init() {
+	// Mutex is locked by Token() here
+	if (!std::exchange(Initialized, true))
+		doInit();
+}
+
+Init::~Init() {
+	std::thread t([]() {
+		// We need to lock Mutex ourselves
+		std::unique_lock lock(Mutex);
+		if (Global)
+			return;
+		if (std::exchange(Initialized, false))
+			doCleanup();
+	});
+	t.detach();
+}
+
 } // namespace rtc
 

src/log.cpp

@@ -19,6 +19,8 @@
 #include "log.hpp"
 
 #include "plog/Appenders/ColorConsoleAppender.h"
+#include "plog/Formatters/TxtFormatter.h"
+#include "plog/Init.h"
 #include "plog/Log.h"
 #include "plog/Logger.h"
 

src/peerconnection.cpp

@@ -18,9 +18,12 @@
 
 #include "peerconnection.hpp"
 #include "certificate.hpp"
+#include "include.hpp"
+#include "processor.hpp"
+#include "threadpool.hpp"
+
 #include "dtlstransport.hpp"
 #include "icetransport.hpp"
-#include "include.hpp"
 #include "sctptransport.hpp"
 
 #if RTC_ENABLE_MEDIA
@@ -39,8 +42,8 @@ using std::weak_ptr;
 PeerConnection::PeerConnection() : PeerConnection(Configuration()) {}
 
 PeerConnection::PeerConnection(const Configuration &config)
-    : mConfig(config), mCertificate(make_certificate()), mState(State::New),
-      mGatheringState(GatheringState::New) {
+    : mConfig(config), mCertificate(make_certificate()), mProcessor(std::make_unique<Processor>()),
+      mState(State::New), mGatheringState(GatheringState::New) {
 	PLOG_VERBOSE << "Creating PeerConnection";
 
 	if (config.portRangeEnd && config.portRangeBegin > config.portRangeEnd)
@@ -92,18 +95,20 @@ void PeerConnection::setLocalDescription(std::optional<Description> description)
 
 void PeerConnection::setRemoteDescription(Description description) {
 	description.hintType(localDescription() ? Description::Type::Answer : Description::Type::Offer);
-	auto remoteCandidates = description.extractCandidates();
-
-	std::lock_guard lock(mRemoteDescriptionMutex);
-	mRemoteDescription.emplace(std::move(description));
+	auto type = description.type();
+	auto remoteCandidates = description.extractCandidates(); // Candidates will be added at the end
 
 	auto iceTransport = std::atomic_load(&mIceTransport);
 	if (!iceTransport)
 		iceTransport = initIceTransport(Description::Role::ActPass);
+	iceTransport->setRemoteDescription(description);
 
-	iceTransport->setRemoteDescription(*mRemoteDescription);
+	{
+		std::lock_guard lock(mRemoteDescriptionMutex);
+		mRemoteDescription.emplace(std::move(description));
+	}
 
-	if (mRemoteDescription->type() == Description::Type::Offer) {
+	if (type == Description::Type::Offer) {
 		// This is an offer and we are the answerer.
 		Description localDescription = iceTransport->getLocalDescription(Description::Type::Answer);
 		localDescription.addMedia(description); // blindly accept media
@@ -143,6 +148,7 @@ void PeerConnection::addRemoteCandidate(Candidate candidate) {
 		iceTransport->addRemoteCandidate(candidate);
 	} else {
 		// OK, we might need a lookup, do it asynchronously
+		// We don't use the thread pool because we have no control on the timeout
 		weak_ptr<IceTransport> weakIceTransport{iceTransport};
 		std::thread t([weakIceTransport, candidate]() mutable {
 			if (candidate.resolve(Candidate::ResolveMode::Lookup))
@@ -225,7 +231,7 @@ void PeerConnection::sendMedia(const binary &packet) {
 	outgoingMedia(make_message(packet.begin(), packet.end(), Message::Binary));
 }
 
-void PeerConnection::send(const byte *packet, size_t size) {
+void PeerConnection::sendMedia(const byte *packet, size_t size) {
 	outgoingMedia(make_message(packet, packet + size, Message::Binary));
 }
 
@@ -242,7 +248,7 @@ void PeerConnection::outgoingMedia(message_ptr message) {
 	if (!transport)
 		throw std::runtime_error("PeerConnection is not open");
 
-	std::dynamic_pointer_cast<DtlsSrtpTransport>(transport)->send(message);
+	std::dynamic_pointer_cast<DtlsSrtpTransport>(transport)->sendMedia(message);
 #else
 	PLOG_WARNING << "Ignoring sent media (not compiled with SRTP support)";
 #endif
@@ -443,8 +449,7 @@ void PeerConnection::closeTransports() {
 	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, token = mInitToken]() mutable {
+	ThreadPool::Instance().enqueue([sctp, dtls, ice]() mutable {
 		if (sctp)
 			sctp->stop();
 		if (dtls)
@@ -456,8 +461,6 @@ void PeerConnection::closeTransports() {
 		dtls.reset();
 		ice.reset();
 	});
-		t.detach();
-	}
 }
 
 void PeerConnection::endLocalCandidates() {
@@ -481,7 +484,7 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		return;
 	}
 
-	auto channel = findDataChannel(message->stream);
+	auto channel = findDataChannel(uint16_t(message->stream));
 
 	auto iceTransport = std::atomic_load(&mIceTransport);
 	auto sctpTransport = std::atomic_load(&mSctpTransport);
@@ -500,7 +503,7 @@ void PeerConnection::forwardMessage(message_ptr message) {
 			mDataChannels.insert(std::make_pair(message->stream, channel));
 		} else {
 			// Invalid, close the DataChannel
-			sctpTransport->close(message->stream);
+			sctpTransport->closeStream(message->stream);
 			return;
 		}
 	}
@@ -592,18 +595,26 @@ void PeerConnection::remoteCloseDataChannels() {
 
 void PeerConnection::processLocalDescription(Description description) {
 	std::optional<uint16_t> remoteSctpPort;
-	if (auto remote = remoteDescription())
+	std::optional<string> remoteDataMid;
+	if (auto remote = remoteDescription()) {
+		remoteDataMid = remote->dataMid();
 	    remoteSctpPort = remote->sctpPort();
+	}
 
 	auto certificate = mCertificate.get(); // wait for certificate if not ready
 
+	{
 		std::lock_guard lock(mLocalDescriptionMutex);
 		mLocalDescription.emplace(std::move(description));
+		if (remoteDataMid)
+			mLocalDescription->setDataMid(*remoteDataMid);
+
 		mLocalDescription->setFingerprint(certificate->fingerprint());
 		mLocalDescription->setSctpPort(remoteSctpPort.value_or(DEFAULT_SCTP_PORT));
 		mLocalDescription->setMaxMessageSize(LOCAL_MAX_MESSAGE_SIZE);
+	}
 
-	mLocalDescriptionCallback(*mLocalDescription);
+	mProcessor->enqueue([this]() { mLocalDescriptionCallback(*mLocalDescription); });
 }
 
 void PeerConnection::processLocalCandidate(Candidate candidate) {
@@ -613,7 +624,8 @@ void PeerConnection::processLocalCandidate(Candidate candidate) {
 
 	mLocalDescription->addCandidate(candidate);
 
-	mLocalCandidateCallback(candidate);
+	mProcessor->enqueue(
+	    [this, candidate = std::move(candidate)]() { mLocalCandidateCallback(candidate); });
 }
 
 void PeerConnection::triggerDataChannel(weak_ptr<DataChannel> weakDataChannel) {
@@ -621,7 +633,8 @@ void PeerConnection::triggerDataChannel(weak_ptr<DataChannel> weakDataChannel) {
 	if (!dataChannel)
 		return;
 
-	mDataChannelCallback(dataChannel);
+	mProcessor->enqueue(
+	    [this, dataChannel = std::move(dataChannel)]() { mDataChannelCallback(dataChannel); });
 }
 
 bool PeerConnection::changeState(State state) {
@@ -635,13 +648,13 @@ bool PeerConnection::changeState(State state) {
 
 	} while (!mState.compare_exchange_weak(current, state));
 
-	mStateChangeCallback(state);
+	mProcessor->enqueue([this, state]() { mStateChangeCallback(state); });
 	return true;
 }
 
 bool PeerConnection::changeGatheringState(GatheringState state) {
 	if (mGatheringState.exchange(state) != state)
-		mGatheringStateChangeCallback(state);
+		mProcessor->enqueue([this, state] { mGatheringStateChangeCallback(state); });
 	return true;
 }
 
@@ -654,13 +667,14 @@ void PeerConnection::resetCallbacks() {
 	mGatheringStateChangeCallback = nullptr;
 }
 
-bool PeerConnection::getSelectedCandidatePair(CandidateInfo *local, CandidateInfo *remote) {
-#if not USE_JUICE
+bool PeerConnection::getSelectedCandidatePair([[maybe_unused]] CandidateInfo *local,
+                                              [[maybe_unused]] CandidateInfo *remote) {
+#if USE_JUICE
+	PLOG_WARNING << "getSelectedCandidatePair() is not implemented for libjuice";
+	return false;
+#else
 	auto iceTransport = std::atomic_load(&mIceTransport);
 	return iceTransport->getSelectedCandidatePair(local, remote);
-#else
-	PLOG_WARNING << "getSelectedCandidatePair is not implemented for libjuice";
-	return false;
 #endif
 }
 

src/processor.cpp

@@ -0,0 +1,44 @@
+/**
+ * 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 "processor.hpp"
+
+namespace rtc {
+
+Processor::~Processor() { join(); }
+
+void Processor::join() {
+	std::unique_lock lock(mMutex);
+	mCondition.wait(lock, [this]() { return !mPending && mTasks.empty(); });
+}
+
+void Processor::schedule() {
+	std::unique_lock lock(mMutex);
+	if (mTasks.empty()) {
+		// No more tasks
+		mPending = false;
+		mCondition.notify_all();
+		return;
+	}
+
+	ThreadPool::Instance().enqueue(std::move(mTasks.front()));
+	mTasks.pop();
+}
+
+} // namespace rtc
+

src/processor.hpp

@@ -0,0 +1,92 @@
+/**
+ * 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_PROCESSOR_H
+#define RTC_PROCESSOR_H
+
+#include "include.hpp"
+#include "init.hpp"
+#include "threadpool.hpp"
+
+#include <condition_variable>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <queue>
+
+namespace rtc {
+
+// Processed tasks in order by delegating them to the thread pool
+class Processor final {
+public:
+	Processor() = default;
+	~Processor();
+
+	Processor(const Processor &) = delete;
+	Processor &operator=(const Processor &) = delete;
+	Processor(Processor &&) = delete;
+	Processor &operator=(Processor &&) = delete;
+
+	void join();
+
+	template <class F, class... Args>
+	auto enqueue(F &&f, Args &&... args) -> invoke_future_t<F, Args...>;
+
+protected:
+	void schedule();
+
+	// Keep an init token
+	const init_token mInitToken = Init::Token();
+
+	std::queue<std::function<void()>> mTasks;
+	bool mPending = false; // true iff a task is pending in the thread pool
+
+	mutable std::mutex mMutex;
+	std::condition_variable mCondition;
+};
+
+template <class F, class... Args>
+auto Processor::enqueue(F &&f, Args &&... args) -> invoke_future_t<F, Args...> {
+	std::unique_lock lock(mMutex);
+	using R = std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>;
+	auto task = std::make_shared<std::packaged_task<R()>>(
+	    std::bind(std::forward<F>(f), std::forward<Args>(args)...));
+	std::future<R> result = task->get_future();
+
+	auto bundle = [this, task = std::move(task)]() {
+		try {
+			(*task)();
+		} catch (const std::exception &e) {
+			PLOG_WARNING << "Unhandled exception in task: " << e.what();
+		}
+		schedule(); // chain the next task
+	};
+
+	if (!mPending) {
+		ThreadPool::Instance().enqueue(std::move(bundle));
+		mPending = true;
+	} else {
+		mTasks.emplace(std::move(bundle));
+	}
+
+	return result;
+}
+
+} // namespace rtc
+
+#endif

src/rtc.cpp

@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2019 Paul-Louis Ageneau
+ * 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
@@ -18,14 +18,16 @@
 
 #include "include.hpp"
 
-#include "datachannel.hpp"
-#include "peerconnection.hpp"
+#include "rtc.h"
 
+#include "datachannel.hpp"
+#include "log.hpp"
+#include "peerconnection.hpp"
 #if RTC_ENABLE_WEBSOCKET
 #include "websocket.hpp"
 #endif
 
-#include <rtc.h>
+#include "plog/Formatters/FuncMessageFormatter.h"
 
 #include <exception>
 #include <mutex>
@@ -33,6 +35,11 @@
 #include <unordered_map>
 #include <utility>
 
+#ifdef _WIN32
+#include <codecvt>
+#include <locale>
+#endif
+
 using namespace rtc;
 using std::shared_ptr;
 using std::string;
@@ -48,18 +55,15 @@ std::unordered_map<int, void *> userPointerMap;
 std::mutex mutex;
 int lastId = 0;
 
-void *getUserPointer(int id) {
+std::optional<void *> getUserPointer(int id) {
 	std::lock_guard lock(mutex);
 	auto it = userPointerMap.find(id);
-	return it != userPointerMap.end() ? it->second : nullptr;
+	return it != userPointerMap.end() ? std::make_optional(it->second) : nullopt;
 }
 
 void setUserPointer(int i, void *ptr) {
 	std::lock_guard lock(mutex);
-	if (ptr)
-		userPointerMap.insert(std::make_pair(i, ptr));
-	else
-		userPointerMap.erase(i);
+	userPointerMap[i] = ptr;
 }
 
 shared_ptr<PeerConnection> getPeerConnection(int id) {
@@ -82,6 +86,7 @@ int emplacePeerConnection(shared_ptr<PeerConnection> ptr) {
 	std::lock_guard lock(mutex);
 	int pc = ++lastId;
 	peerConnectionMap.emplace(std::make_pair(pc, ptr));
+	userPointerMap.emplace(std::make_pair(pc, nullptr));
 	return pc;
 }
 
@@ -89,6 +94,7 @@ int emplaceDataChannel(shared_ptr<DataChannel> ptr) {
 	std::lock_guard lock(mutex);
 	int dc = ++lastId;
 	dataChannelMap.emplace(std::make_pair(dc, ptr));
+	userPointerMap.emplace(std::make_pair(dc, nullptr));
 	return dc;
 }
 
@@ -119,6 +125,7 @@ int emplaceWebSocket(shared_ptr<WebSocket> ptr) {
 	std::lock_guard lock(mutex);
 	int ws = ++lastId;
 	webSocketMap.emplace(std::make_pair(ws, ptr));
+	userPointerMap.emplace(std::make_pair(ws, nullptr));
 	return ws;
 }
 
@@ -143,7 +150,7 @@ shared_ptr<Channel> getChannel(int id) {
 
 template <typename F> int wrap(F func) {
 	try {
-		return func();
+		return int(func());
 
 	} catch (const std::invalid_argument &e) {
 		PLOG_ERROR << e.what();
@@ -160,9 +167,48 @@ template <typename F> int wrap(F func) {
 		return RTC_ERR_SUCCESS;                                                                    \
 	})
 
+class plog_appender : public plog::IAppender {
+public:
+	plog_appender(rtcLogCallbackFunc cb = nullptr) { set_callback(cb); }
+
+	void set_callback(rtcLogCallbackFunc cb) {
+		std::lock_guard lock(mutex);
+		callback = cb;
+	}
+
+	void write(const plog::Record &record) override {
+		plog::Severity severity = record.getSeverity();
+		auto formatted = plog::FuncMessageFormatter::format(record);
+		formatted.pop_back(); // remove newline
+#ifdef _WIN32
+		using convert_type = std::codecvt_utf8<wchar_t>;
+		std::wstring_convert<convert_type, wchar_t> converter;
+		std::string str = converter.to_bytes(formatted);
+#else
+		std::string str = formatted;
+#endif
+		std::lock_guard lock(mutex);
+		if (callback)
+			callback(static_cast<rtcLogLevel>(record.getSeverity()), str.c_str());
+		else
+			std::cout << plog::severityToString(severity) << " " << str << std::endl;
+	}
+
+private:
+	rtcLogCallbackFunc callback;
+};
+
 } // namespace
 
-void rtcInitLogger(rtcLogLevel level) { InitLogger(static_cast<LogLevel>(level)); }
+void rtcInitLogger(rtcLogLevel level, rtcLogCallbackFunc cb) {
+	static std::optional<plog_appender> appender;
+	if (appender)
+		appender->set_callback(cb);
+	else if (cb)
+		appender.emplace(plog_appender(cb));
+
+	InitLogger(static_cast<plog::Severity>(level), appender ? &appender.value() : nullptr);
+}
 
 void rtcSetUserPointer(int i, void *ptr) { setUserPointer(i, ptr); }
 
@@ -198,7 +244,8 @@ int rtcCreateDataChannel(int pc, const char *label) {
 	return WRAP({
 		auto peerConnection = getPeerConnection(pc);
 		int dc = emplaceDataChannel(peerConnection->createDataChannel(string(label)));
-		rtcSetUserPointer(dc, getUserPointer(pc));
+		if (auto ptr = getUserPointer(pc))
+			rtcSetUserPointer(dc, *ptr);
 		return dc;
 	});
 }
@@ -241,63 +288,68 @@ int rtcDeleteWebsocket(int ws) {
 }
 #endif
 
-int rtcSetDataChannelCallback(int pc, dataChannelCallbackFunc cb) {
+int rtcSetDataChannelCallback(int pc, rtcDataChannelCallbackFunc cb) {
 	return WRAP({
 		auto peerConnection = getPeerConnection(pc);
 		if (cb)
 			peerConnection->onDataChannel([pc, cb](std::shared_ptr<DataChannel> dataChannel) {
 				int dc = emplaceDataChannel(dataChannel);
-				void *ptr = getUserPointer(pc);
-				rtcSetUserPointer(dc, ptr);
-				cb(dc, ptr);
+				if (auto ptr = getUserPointer(pc)) {
+					rtcSetUserPointer(dc, *ptr);
+					cb(dc, *ptr);
+				}
 			});
 		else
 			peerConnection->onDataChannel(nullptr);
 	});
 }
 
-int rtcSetLocalDescriptionCallback(int pc, descriptionCallbackFunc cb) {
+int rtcSetLocalDescriptionCallback(int pc, rtcDescriptionCallbackFunc cb) {
 	return WRAP({
 		auto peerConnection = getPeerConnection(pc);
 		if (cb)
 			peerConnection->onLocalDescription([pc, cb](const Description &desc) {
-				cb(string(desc).c_str(), desc.typeString().c_str(), getUserPointer(pc));
+				if (auto ptr = getUserPointer(pc))
+					cb(string(desc).c_str(), desc.typeString().c_str(), *ptr);
 			});
 		else
 			peerConnection->onLocalDescription(nullptr);
 	});
 }
 
-int rtcSetLocalCandidateCallback(int pc, candidateCallbackFunc cb) {
+int rtcSetLocalCandidateCallback(int pc, rtcCandidateCallbackFunc cb) {
 	return WRAP({
 		auto peerConnection = getPeerConnection(pc);
 		if (cb)
 			peerConnection->onLocalCandidate([pc, cb](const Candidate &cand) {
-				cb(cand.candidate().c_str(), cand.mid().c_str(), getUserPointer(pc));
+				if (auto ptr = getUserPointer(pc))
+					cb(cand.candidate().c_str(), cand.mid().c_str(), *ptr);
 			});
 		else
 			peerConnection->onLocalCandidate(nullptr);
 	});
 }
 
-int rtcSetStateChangeCallback(int pc, stateChangeCallbackFunc cb) {
+int rtcSetStateChangeCallback(int pc, rtcStateChangeCallbackFunc cb) {
 	return WRAP({
 		auto peerConnection = getPeerConnection(pc);
 		if (cb)
 			peerConnection->onStateChange([pc, cb](PeerConnection::State state) {
-				cb(static_cast<rtcState>(state), getUserPointer(pc));
+				if (auto ptr = getUserPointer(pc))
+					cb(static_cast<rtcState>(state), *ptr);
 			});
 		else
 			peerConnection->onStateChange(nullptr);
 	});
 }
 
-int rtcSetGatheringStateChangeCallback(int pc, gatheringStateCallbackFunc cb) {
+int rtcSetGatheringStateChangeCallback(int pc, rtcGatheringStateCallbackFunc cb) {
 	return WRAP({
 		auto peerConnection = getPeerConnection(pc);
 		if (cb)
 			peerConnection->onGatheringStateChange([pc, cb](PeerConnection::GatheringState state) {
-				cb(static_cast<rtcGatheringState>(state), getUserPointer(pc));
+				if (auto ptr = getUserPointer(pc))
+					cb(static_cast<rtcGatheringState>(state), *ptr);
 			});
 		else
 			peerConnection->onGatheringStateChange(nullptr);
@@ -333,9 +385,13 @@ int rtcGetLocalAddress(int pc, char *buffer, int size) {
 		if (!buffer)
 			throw std::invalid_argument("Unexpected null pointer");
 
+		if (size <= 0)
+			return 0;
+
 		if (auto addr = peerConnection->localAddress()) {
-			size = std::min(size_t(size - 1), addr->size());
-			std::copy(addr->data(), addr->data() + size, buffer);
+			const char *data = addr->data();
+			size = std::min(size - 1, int(addr->size()));
+			std::copy(data, data + size, buffer);
 			buffer[size] = '\0';
 			return size + 1;
 		}
@@ -349,11 +405,15 @@ int rtcGetRemoteAddress(int pc, char *buffer, int size) {
 		if (!buffer)
 			throw std::invalid_argument("Unexpected null pointer");
 
+		if (size <= 0)
+			return 0;
+
 		if (auto addr = peerConnection->remoteAddress()) {
-			size = std::min(size_t(size - 1), addr->size());
-			std::copy(addr->data(), addr->data() + size, buffer);
+			const char *data = addr->data();
+			size = std::min(size - 1, int(addr->size()));
+			std::copy(data, data + size, buffer);
 			buffer[size] = '\0';
-			return size + 1;
+			return int(size + 1);
 		}
 	});
 }
@@ -365,58 +425,70 @@ int rtcGetDataChannelLabel(int dc, char *buffer, int size) {
 		if (!buffer)
 			throw std::invalid_argument("Unexpected null pointer");
 
-		if (size >= 0) {
-			string label = dataChannel->label();
-			size = std::min(size_t(size - 1), label.size());
-			std::copy(label.data(), label.data() + size, buffer);
-			buffer[size] = '\0';
-			return size + 1;
-		} else {
+		if (size <= 0)
 			return 0;
-		}
+
+		string label = dataChannel->label();
+		const char *data = label.data();
+		size = std::min(size - 1, int(label.size()));
+		std::copy(data, data + size, buffer);
+		buffer[size] = '\0';
+		return int(size + 1);
 	});
 }
 
-int rtcSetOpenCallback(int id, openCallbackFunc cb) {
+int rtcSetOpenCallback(int id, rtcOpenCallbackFunc cb) {
 	return WRAP({
 		auto channel = getChannel(id);
 		if (cb)
-			channel->onOpen([id, cb]() { cb(getUserPointer(id)); });
+			channel->onOpen([id, cb]() {
+				if (auto ptr = getUserPointer(id))
+					cb(*ptr);
+			});
 		else
 			channel->onOpen(nullptr);
 	});
 }
 
-int rtcSetClosedCallback(int id, closedCallbackFunc cb) {
+int rtcSetClosedCallback(int id, rtcClosedCallbackFunc cb) {
 	return WRAP({
 		auto channel = getChannel(id);
 		if (cb)
-			channel->onClosed([id, cb]() { cb(getUserPointer(id)); });
+			channel->onClosed([id, cb]() {
+				if (auto ptr = getUserPointer(id))
+					cb(*ptr);
+			});
 		else
 			channel->onClosed(nullptr);
 	});
 }
 
-int rtcSetErrorCallback(int id, errorCallbackFunc cb) {
+int rtcSetErrorCallback(int id, rtcErrorCallbackFunc cb) {
 	return WRAP({
 		auto channel = getChannel(id);
 		if (cb)
-			channel->onError(
-			    [id, cb](const string &error) { cb(error.c_str(), getUserPointer(id)); });
+			channel->onError([id, cb](const string &error) {
+				if (auto ptr = getUserPointer(id))
+					cb(error.c_str(), *ptr);
+			});
 		else
 			channel->onError(nullptr);
 	});
 }
 
-int rtcSetMessageCallback(int id, messageCallbackFunc cb) {
+int rtcSetMessageCallback(int id, rtcMessageCallbackFunc cb) {
 	return WRAP({
 		auto channel = getChannel(id);
 		if (cb)
 			channel->onMessage(
 			    [id, cb](const binary &b) {
-				    cb(reinterpret_cast<const char *>(b.data()), b.size(), getUserPointer(id));
+				    if (auto ptr = getUserPointer(id))
+					    cb(reinterpret_cast<const char *>(b.data()), int(b.size()), *ptr);
 			    },
-			    [id, cb](const string &s) { cb(s.c_str(), -1, getUserPointer(id)); });
+			    [id, cb](const string &s) {
+				    if (auto ptr = getUserPointer(id))
+					    cb(s.c_str(), -int(s.size() + 1), *ptr);
+			    });
 		else
 			channel->onMessage(nullptr);
 	});
@@ -435,7 +507,7 @@ int rtcSendMessage(int id, const char *data, int size) {
 			return size;
 		} else {
 			string str(data);
-			int len = str.size();
+			int len = int(str.size());
 			channel->send(std::move(str));
 			return len;
 		}
@@ -456,11 +528,14 @@ int rtcSetBufferedAmountLowThreshold(int id, int amount) {
 	});
 }
 
-int rtcSetBufferedAmountLowCallback(int id, bufferedAmountLowCallbackFunc cb) {
+int rtcSetBufferedAmountLowCallback(int id, rtcBufferedAmountLowCallbackFunc cb) {
 	return WRAP({
 		auto channel = getChannel(id);
 		if (cb)
-			channel->onBufferedAmountLow([id, cb]() { cb(getUserPointer(id)); });
+			channel->onBufferedAmountLow([id, cb]() {
+				if (auto ptr = getUserPointer(id))
+					cb(*ptr);
+			});
 		else
 			channel->onBufferedAmountLow(nullptr);
 	});
@@ -470,11 +545,14 @@ int rtcGetAvailableAmount(int id) {
 	return WRAP({ return int(getChannel(id)->availableAmount()); });
 }
 
-int rtcSetAvailableCallback(int id, availableCallbackFunc cb) {
+int rtcSetAvailableCallback(int id, rtcAvailableCallbackFunc cb) {
 	return WRAP({
 		auto channel = getChannel(id);
 		if (cb)
-			channel->onOpen([id, cb]() { cb(getUserPointer(id)); });
+			channel->onOpen([id, cb]() {
+				if (auto ptr = getUserPointer(id))
+					cb(*ptr);
+			});
 		else
 			channel->onOpen(nullptr);
 	});

src/sctptransport.cpp

@@ -50,6 +50,9 @@ using std::shared_ptr;
 
 namespace rtc {
 
+std::unordered_set<SctpTransport *> SctpTransport::Instances;
+std::shared_mutex SctpTransport::InstancesMutex;
+
 void SctpTransport::Init() {
 	usrsctp_init(0, &SctpTransport::WriteCallback, nullptr);
 	usrsctp_sysctl_set_sctp_ecn_enable(0);
@@ -92,6 +95,11 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 	PLOG_DEBUG << "Initializing SCTP transport";
 
 	usrsctp_register_address(this);
+	{
+		std::unique_lock lock(InstancesMutex);
+		Instances.insert(this);
+	}
+
 	mSock = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP, &SctpTransport::RecvCallback,
 	                       &SctpTransport::SendCallback, 0, this);
 	if (!mSock)
@@ -186,14 +194,21 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 
 SctpTransport::~SctpTransport() {
 	stop();
-
-	if (mSock)
-		usrsctp_close(mSock);
+	close();
 
 	usrsctp_deregister_address(this);
+	{
+		std::unique_lock lock(InstancesMutex);
+		Instances.erase(this);
+	}
 }
 
 bool SctpTransport::stop() {
+	// Transport::stop() will unregister incoming() from the lower layer, therefore we need to make
+	// sure the thread from lower layers is not blocked in incoming() by the WrittenOnce condition.
+	mWrittenOnce = true;
+	mWrittenCondition.notify_all();
+
 	if (!Transport::stop())
 		return false;
 
@@ -204,6 +219,13 @@ bool SctpTransport::stop() {
 	return true;
 }
 
+void SctpTransport::close() {
+	if (mSock) {
+		usrsctp_close(mSock);
+		mSock = nullptr;
+	}
+}
+
 void SctpTransport::connect() {
 	if (!mSock)
 		return;
@@ -240,9 +262,7 @@ void SctpTransport::shutdown() {
 		PLOG_WARNING << "SCTP shutdown failed, errno=" << errno;
 	}
 
-	// close() abort the connection when linger is disabled, call it now
-	usrsctp_close(mSock);
-	mSock = nullptr;
+	close();
 
 	PLOG_INFO << "SCTP disconnected";
 	changeState(State::Disconnected);
@@ -262,11 +282,11 @@ bool SctpTransport::send(message_ptr message) {
 		return true;
 
 	mSendQueue.push(message);
-	updateBufferedAmount(message->stream, message_size_func(message));
+	updateBufferedAmount(uint16_t(message->stream), long(message_size_func(message)));
 	return false;
 }
 
-void SctpTransport::close(unsigned int stream) {
+void SctpTransport::closeStream(unsigned int stream) {
 	send(make_message(0, Message::Reset, uint16_t(stream)));
 }
 
@@ -302,7 +322,7 @@ bool SctpTransport::trySendQueue() {
 		if (!trySendMessage(message))
 			return false;
 		mSendQueue.pop();
-		updateBufferedAmount(message->stream, -message_size_func(message));
+		updateBufferedAmount(uint16_t(message->stream), -long(message_size_func(message)));
 	}
 	return true;
 }
@@ -324,7 +344,7 @@ bool SctpTransport::trySendMessage(message_ptr message) {
 		ppid = PPID_CONTROL;
 		break;
 	case Message::Reset:
-		sendReset(message->stream);
+		sendReset(uint16_t(message->stream));
 		return true;
 	default:
 		// Ignore
@@ -405,7 +425,7 @@ void SctpTransport::updateBufferedAmount(uint16_t streamId, long delta) {
 	try {
 		mBufferedAmountCallback(streamId, amount);
 	} catch (const std::exception &e) {
-		PLOG_WARNING << "SCTP buffered amount callback: " << e.what();
+		PLOG_DEBUG << "SCTP buffered amount callback: " << e.what();
 	}
 	mSendMutex.lock();
 }
@@ -448,32 +468,46 @@ bool SctpTransport::safeFlush() {
 	}
 }
 
-int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, const byte *data,
-                              size_t len, struct sctp_rcvinfo info, int flags) {
+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;
+			return 0; // Ignore
 
-		// This is valid because SCTP_FRAGMENT_INTERLEAVE is set to level 0
-		// so partial messages and notifications may not be interleaved.
+		// SCTP_FRAGMENT_INTERLEAVE does not seem to work as expected for messages > 64KB,
+		// therefore partial notifications and messages need to be handled separately.
+		if (flags & MSG_NOTIFICATION) {
+			// SCTP event notification
 			if (flags & MSG_EOR) {
-			if (!mPartialRecv.empty()) {
-				mPartialRecv.insert(mPartialRecv.end(), data, data + len);
-				data = mPartialRecv.data();
-				len = mPartialRecv.size();
+				if (!mPartialNotification.empty()) {
+					mPartialNotification.insert(mPartialNotification.end(), data, data + len);
+					data = mPartialNotification.data();
+					len = mPartialNotification.size();
 				}
-			// Message/Notification is complete, process it
-			if (flags & MSG_NOTIFICATION)
+				// Notification is complete, process it
 				processNotification(reinterpret_cast<const union sctp_notification *>(data), len);
-			else
-				processData(data, len, info.rcv_sid, PayloadId(htonl(info.rcv_ppid)));
-
-			mPartialRecv.clear();
+				mPartialNotification.clear();
 			} else {
-			// Message/Notification is not complete
-			mPartialRecv.insert(mPartialRecv.end(), data, data + len);
+				mPartialNotification.insert(mPartialNotification.end(), data, data + len);
 			}
+
+		} else {
+			// SCTP message
+			if (flags & MSG_EOR) {
+				if (!mPartialMessage.empty()) {
+					mPartialMessage.insert(mPartialMessage.end(), data, data + len);
+					data = mPartialMessage.data();
+					len = mPartialMessage.size();
+				}
+				// Message is complete, process it
+				processData(data, len, info.rcv_sid, PayloadId(htonl(info.rcv_ppid)));
+				mPartialMessage.clear();
+			} else {
+				mPartialMessage.insert(mPartialMessage.end(), data, data + len);
+			}
+		}
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "SCTP recv: " << e.what();
 		return -1;
@@ -486,11 +520,11 @@ int SctpTransport::handleSend(size_t free) {
 	return safeFlush() ? 0 : -1;
 }
 
-int SctpTransport::handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_df) {
+int SctpTransport::handleWrite(byte *data, size_t len, uint8_t /*tos*/, uint8_t /*set_df*/) {
 	try {
+		std::unique_lock lock(mWriteMutex);
 		PLOG_VERBOSE << "Handle write, len=" << len;
 
-		std::unique_lock lock(mWriteMutex);
 		if (!outgoing(make_message(data, data + len)))
 			return -1;
 
@@ -614,7 +648,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];
-				close(streamId);
+				closeStream(streamId);
 			}
 		}
 		if (flags & SCTP_STREAM_RESET_INCOMING_SSN) {
@@ -672,7 +706,15 @@ int SctpTransport::SendCallback(struct socket *sock, uint32_t sb_free) {
 
 	auto sconn = reinterpret_cast<struct sockaddr_conn *>(&paddrinfo.spinfo_address);
 	void *ptr = sconn->sconn_addr;
-	return static_cast<SctpTransport *>(ptr)->handleSend(size_t(sb_free));
+	auto *transport = static_cast<SctpTransport *>(ptr);
+
+	// Workaround for sctplab/usrsctp#405: Send callback is invoked on already closed socket
+	// https://github.com/sctplab/usrsctp/issues/405
+	std::shared_lock lock(InstancesMutex);
+	if (Instances.find(transport) == Instances.end())
+		return -1;
+
+	return transport->handleSend(size_t(sb_free));
 }
 
 int SctpTransport::WriteCallback(void *ptr, void *data, size_t len, uint8_t tos, uint8_t set_df) {

src/sctptransport.hpp

@@ -28,6 +28,8 @@
 #include <functional>
 #include <map>
 #include <mutex>
+#include <shared_mutex>
+#include <unordered_set>
 
 #include "usrsctp.h"
 
@@ -46,7 +48,7 @@ public:
 
 	bool stop() override;
 	bool send(message_ptr message) override; // false if buffered
-	void close(unsigned int stream);
+	void closeStream(unsigned int stream);
 	void flush();
 
 	// Stats
@@ -70,6 +72,7 @@ private:
 
 	void connect();
 	void shutdown();
+	void close();
 	void incoming(message_ptr message) override;
 
 	bool trySendQueue();
@@ -99,7 +102,8 @@ private:
 	std::atomic<bool> mWritten = false; // written outside lock
 	std::atomic<bool> mWrittenOnce = false; // same
 
-	binary mPartialRecv, mPartialStringData, mPartialBinaryData;
+	binary mPartialMessage, mPartialNotification;
+	binary mPartialStringData, mPartialBinaryData;
 
 	// Stats
 	std::atomic<size_t> mBytesSent = 0, mBytesReceived = 0;
@@ -108,6 +112,9 @@ private:
 	                        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);
+
+	static std::unordered_set<SctpTransport *> Instances;
+	static std::shared_mutex InstancesMutex;
 };
 
 } // namespace rtc

src/tcptransport.cpp

@@ -21,6 +21,7 @@
 #if RTC_ENABLE_WEBSOCKET
 
 #include <exception>
+
 #ifndef _WIN32
 #include <fcntl.h>
 #include <unistd.h>
@@ -37,8 +38,8 @@ SelectInterrupter::SelectInterrupter() {
 		throw std::runtime_error("Failed to create pipe");
 	::fcntl(pipefd[0], F_SETFL, O_NONBLOCK);
 	::fcntl(pipefd[1], F_SETFL, O_NONBLOCK);
-	mPipeOut = pipefd[0]; // read
-	mPipeIn = pipefd[1];  // write
+	mPipeOut = pipefd[1]; // read
+	mPipeIn = pipefd[0];  // write
 #endif
 }
 
@@ -61,11 +62,8 @@ int SelectInterrupter::prepare(fd_set &readfds, fd_set &writefds) {
 	FD_SET(mDummySock, &readfds);
 	return SOCKET_TO_INT(mDummySock) + 1;
 #else
-	int ret;
-	do {
 	char dummy;
-		ret = ::read(mPipeIn, &dummy, 1);
-	} while (ret > 0);
+	::read(mPipeIn, &dummy, 1);
 	FD_SET(mPipeIn, &readfds);
 	return mPipeIn + 1;
 #endif
@@ -106,6 +104,7 @@ bool TcpTransport::stop() {
 }
 
 bool TcpTransport::send(message_ptr message) {
+	std::unique_lock lock(mSockMutex);
 	if (state() != State::Connected)
 		return false;
 
@@ -125,6 +124,7 @@ void TcpTransport::incoming(message_ptr message) {
 }
 
 bool TcpTransport::outgoing(message_ptr message) {
+	// mSockMutex must be locked
 	// If nothing is pending, try to send directly
 	// It's safe because if the queue is empty, the thread is not sending
 	if (mSendQueue.empty() && trySendMessage(message))
@@ -150,7 +150,7 @@ void TcpTransport::connect(const string &hostname, const string &service) {
 
 	for (auto p = result; p; p = p->ai_next) {
 		try {
-			connect(p->ai_addr, p->ai_addrlen);
+			connect(p->ai_addr, socklen_t(p->ai_addrlen));
 
 			PLOG_INFO << "Connected to " << hostname << ":" << service;
 			freeaddrinfo(result);
@@ -173,6 +173,7 @@ void TcpTransport::connect(const string &hostname, const string &service) {
 }
 
 void TcpTransport::connect(const sockaddr *addr, socklen_t addrlen) {
+	std::unique_lock lock(mSockMutex);
 	try {
 		char node[MAX_NUMERICNODE_LEN];
 		char serv[MAX_NUMERICSERV_LEN];
@@ -201,7 +202,7 @@ void TcpTransport::connect(const sockaddr *addr, socklen_t addrlen) {
 
 		// Initiate connection
 		int ret = ::connect(mSock, addr, addrlen);
-		if (ret < 0 && errno != EINPROGRESS) {
+		if (ret < 0 && sockerrno != SEINPROGRESS && sockerrno != SEWOULDBLOCK) {
 			std::ostringstream msg;
 			msg << "TCP connection to " << node << ":" << serv << " failed, errno=" << sockerrno;
 			throw std::runtime_error(msg.str());
@@ -247,15 +248,18 @@ void TcpTransport::connect(const sockaddr *addr, socklen_t addrlen) {
 }
 
 void TcpTransport::close() {
+	std::unique_lock lock(mSockMutex);
 	if (mSock != INVALID_SOCKET) {
 		PLOG_DEBUG << "Closing TCP socket";
 		::closesocket(mSock);
 		mSock = INVALID_SOCKET;
 	}
 	changeState(State::Disconnected);
+	interruptSelect();
 }
 
 bool TcpTransport::trySendQueue() {
+	// mSockMutex must be locked
 	while (auto next = mSendQueue.peek()) {
 		auto message = *next;
 		if (!trySendMessage(message)) {
@@ -268,17 +272,18 @@ bool TcpTransport::trySendQueue() {
 }
 
 bool TcpTransport::trySendMessage(message_ptr &message) {
+	// mSockMutex must be locked
 	auto data = reinterpret_cast<const char *>(message->data());
 	auto size = message->size();
 	while (size) {
-#if defined(__APPLE__) or defined(_WIN32)
+#if defined(__APPLE__) || defined(_WIN32)
 		int flags = 0;
 #else
 		int flags = MSG_NOSIGNAL;
 #endif
-		int len = ::send(mSock, data, size, flags);
+		int len = ::send(mSock, data, int(size), flags);
 		if (len < 0) {
-			if (errno == EAGAIN || errno == EWOULDBLOCK) {
+			if (sockerrno == SEAGAIN || sockerrno == SEWOULDBLOCK) {
 				message = make_message(message->end() - size, message->end());
 				return false;
 			} else {
@@ -313,13 +318,22 @@ void TcpTransport::runLoop() {
 		changeState(State::Connected);
 
 		while (true) {
+			std::unique_lock lock(mSockMutex);
+			if (mSock == INVALID_SOCKET)
+				break;
+
 			fd_set readfds, writefds;
 			int n = prepareSelect(readfds, writefds);
 
 			struct timeval tv;
 			tv.tv_sec = 10;
 			tv.tv_usec = 0;
+			lock.unlock();
 			int ret = ::select(n, &readfds, &writefds, NULL, &tv);
+			lock.lock();
+			if (mSock == INVALID_SOCKET)
+				break;
+
 			if (ret < 0) {
 				throw std::runtime_error("Failed to wait on socket");
 			} else if (ret == 0) {
@@ -335,7 +349,7 @@ void TcpTransport::runLoop() {
 				char buffer[bufferSize];
 				int len = ::recv(mSock, buffer, bufferSize, 0);
 				if (len < 0) {
-					if (errno == EAGAIN || errno == EWOULDBLOCK) {
+					if (sockerrno == SEAGAIN || sockerrno == SEWOULDBLOCK) {
 						continue;
 					} else {
 						throw std::runtime_error("Connection lost");

src/tcptransport.hpp

@@ -78,6 +78,7 @@ private:
 	string mHostname, mService;
 
 	socket_t mSock = INVALID_SOCKET;
+	std::mutex mSockMutex;
 	std::thread mThread;
 	SelectInterrupter mInterrupter;
 	Queue<message_ptr> mSendQueue;

src/threadpool.cpp

@@ -0,0 +1,81 @@
+/**
+ * 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 "threadpool.hpp"
+
+namespace rtc {
+
+ThreadPool &ThreadPool::Instance() {
+	static ThreadPool instance;
+	return instance;
+}
+
+ThreadPool::~ThreadPool() { join(); }
+
+int ThreadPool::count() const {
+	std::unique_lock lock(mWorkersMutex);
+	return mWorkers.size();
+}
+
+void ThreadPool::spawn(int count) {
+	std::unique_lock lock(mWorkersMutex);
+	mJoining = false;
+	while (count-- > 0)
+		mWorkers.emplace_back(std::bind(&ThreadPool::run, this));
+}
+
+void ThreadPool::join() {
+	std::unique_lock lock(mWorkersMutex);
+	mJoining = true;
+	mCondition.notify_all();
+
+	for (auto &w : mWorkers)
+		w.join();
+
+	mWorkers.clear();
+}
+
+void ThreadPool::run() {
+	while (runOne()) {
+	}
+}
+
+bool ThreadPool::runOne() {
+	if (auto task = dequeue()) {
+		try {
+			task();
+		} catch (const std::exception &e) {
+			PLOG_WARNING << "Unhandled exception in task: " << e.what();
+		}
+		return true;
+	}
+	return false;
+}
+
+std::function<void()> ThreadPool::dequeue() {
+	std::unique_lock lock(mMutex);
+	mCondition.wait(lock, [this]() { return !mTasks.empty() || mJoining; });
+	if (mTasks.empty())
+		return nullptr;
+	auto task = std::move(mTasks.front());
+	mTasks.pop();
+	return task;
+}
+
+} // namespace rtc
+

src/threadpool.hpp

@@ -0,0 +1,87 @@
+/**
+ * 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_THREADPOOL_H
+#define RTC_THREADPOOL_H
+
+#include "include.hpp"
+#include "init.hpp"
+
+#include <condition_variable>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+namespace rtc {
+
+template <class F, class... Args>
+using invoke_future_t = std::future<std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>>;
+
+class ThreadPool final {
+public:
+	static ThreadPool &Instance();
+
+	ThreadPool(const ThreadPool &) = delete;
+	ThreadPool &operator=(const ThreadPool &) = delete;
+	ThreadPool(ThreadPool &&) = delete;
+	ThreadPool &operator=(ThreadPool &&) = delete;
+
+	int count() const;
+	void spawn(int count = 1);
+	void join();
+	void run();
+	bool runOne();
+
+	template <class F, class... Args>
+	auto enqueue(F &&f, Args &&... args) -> invoke_future_t<F, Args...>;
+
+protected:
+	ThreadPool() = default;
+	~ThreadPool();
+
+	std::function<void()> dequeue(); // returns null function if joining
+
+	std::vector<std::thread> mWorkers;
+	std::queue<std::function<void()>> mTasks;
+	std::atomic<bool> mJoining = false;
+
+	mutable std::mutex mMutex, mWorkersMutex;
+	std::condition_variable mCondition;
+};
+
+template <class F, class... Args>
+auto ThreadPool::enqueue(F &&f, Args &&... args) -> invoke_future_t<F, Args...> {
+	std::unique_lock lock(mMutex);
+	using R = std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>;
+	auto task = std::make_shared<std::packaged_task<R()>>(
+	    std::bind(std::forward<F>(f), std::forward<Args>(args)...));
+	std::future<R> result = task->get_future();
+
+	mTasks.emplace([task = std::move(task), token = Init::Token()]() { return (*task)(); });
+	mCondition.notify_one();
+	return result;
+}
+
+} // namespace rtc
+
+#endif

src/tls.cpp

@@ -86,9 +86,8 @@ void init() {
 
 	std::lock_guard lock(mutex);
 	if (!done) {
-		OPENSSL_init_ssl(0, NULL);
-		SSL_load_error_strings();
-		ERR_load_crypto_strings();
+		OPENSSL_init_ssl(OPENSSL_INIT_LOAD_SSL_STRINGS | OPENSSL_INIT_LOAD_CRYPTO_STRINGS, nullptr);
+		OPENSSL_init_crypto(OPENSSL_INIT_LOAD_CRYPTO_STRINGS, nullptr);
 		done = true;
 	}
 }

src/tls.hpp

@@ -48,6 +48,11 @@ gnutls_datum_t make_datum(char *data, size_t size);
 
 #else // USE_GNUTLS==0
 
+#ifdef _WIN32
+// Include winsock2.h header first since OpenSSL may include winsock.h
+#include <winsock2.h>
+#endif
+
 #include <openssl/ssl.h>
 
 #include <openssl/bio.h>
@@ -55,6 +60,8 @@ gnutls_datum_t make_datum(char *data, size_t size);
 #include <openssl/err.h>
 #include <openssl/pem.h>
 #include <openssl/x509.h>
+#include <openssl/rsa.h>
+#include <openssl/bn.h>
 
 #ifndef BIO_EOF
 #define BIO_EOF -1

src/tlstransport.cpp

@@ -269,9 +269,19 @@ TlsTransport::TlsTransport(shared_ptr<TcpTransport> lower, string host, state_ca
 		SSL_CTX_set_quiet_shutdown(mCtx, 1);
 		SSL_CTX_set_info_callback(mCtx, InfoCallback);
 
-		SSL_CTX_set_default_verify_paths(mCtx);
+		// SSL_CTX_set_default_verify_paths() does nothing on Windows
+#ifndef _WIN32
+		if (SSL_CTX_set_default_verify_paths(mCtx)) {
+#else
+		if (false) {
+#endif
+			PLOG_INFO << "SSL root CA certificates available, server verification enabled";
 			SSL_CTX_set_verify(mCtx, SSL_VERIFY_PEER, NULL);
 			SSL_CTX_set_verify_depth(mCtx, 4);
+		} else {
+			PLOG_WARNING << "SSL root CA certificates unavailable, server verification disabled";
+			SSL_CTX_set_verify(mCtx, SSL_VERIFY_NONE, NULL);
+		}
 
 		if (!(mSsl = SSL_new(mCtx)))
 			throw std::runtime_error("Failed to create SSL instance");
@@ -337,7 +347,7 @@ bool TlsTransport::send(message_ptr message) {
 	if (message->size() == 0)
 		return true;
 
-	int ret = SSL_write(mSsl, message->data(), message->size());
+	int ret = SSL_write(mSsl, message->data(), int(message->size()));
 	if (!openssl::check(mSsl, ret))
 		return false;
 
@@ -393,7 +403,7 @@ void TlsTransport::runRecvLoop() {
 
 			message_ptr message = *next;
 			if (message->size() > 0)
-				BIO_write(mInBio, message->data(), message->size()); // Input
+				BIO_write(mInBio, message->data(), int(message->size())); // Input
 			else
 				recv(message); // Pass zero-sized messages through
 		}

src/transport.hpp

@@ -41,12 +41,17 @@ public:
 
 	virtual ~Transport() {
 		stop();
-		if (mLower)
-			mLower->onRecv(nullptr); // doing it on stop could cause a deadlock
 	}
 
 	virtual bool stop() {
-		return !mShutdown.exchange(true);
+		if (mShutdown.exchange(true))
+			return false;
+
+		// We don't want incoming() to be called by the lower layer anymore
+		if (mLower)
+			mLower->onRecv(nullptr);
+
+		return true;
 	}
 
 	void registerIncoming() {

src/websocket.cpp

@@ -18,8 +18,9 @@
 
 #if RTC_ENABLE_WEBSOCKET
 
-#include "include.hpp"
 #include "websocket.hpp"
+#include "include.hpp"
+#include "threadpool.hpp"
 
 #include "tcptransport.hpp"
 #include "tlstransport.hpp"
@@ -301,8 +302,7 @@ void WebSocket::closeTransports() {
 	auto ws = std::atomic_exchange(&mWsTransport, decltype(mWsTransport)(nullptr));
 	auto tls = std::atomic_exchange(&mTlsTransport, decltype(mTlsTransport)(nullptr));
 	auto tcp = std::atomic_exchange(&mTcpTransport, decltype(mTcpTransport)(nullptr));
-	if (ws || tls || tcp) {
-		std::thread t([ws, tls, tcp, token = mInitToken]() mutable {
+	ThreadPool::Instance().enqueue([ws, tls, tcp]() mutable {
 		if (ws)
 			ws->stop();
 		if (tls)
@@ -314,8 +314,6 @@ void WebSocket::closeTransports() {
 		tls.reset();
 		tcp.reset();
 	});
-		t.detach();
-	}
 }
 
 } // namespace rtc

src/wstransport.cpp

@@ -50,7 +50,7 @@ using std::to_integer;
 using std::to_string;
 
 using random_bytes_engine =
-    std::independent_bits_engine<std::default_random_engine, CHAR_BIT, unsigned char>;
+    std::independent_bits_engine<std::default_random_engine, CHAR_BIT, unsigned short>;
 
 WsTransport::WsTransport(std::shared_ptr<Transport> lower, string host, string path,
                          message_callback recvCallback, state_callback stateCallback)
@@ -145,12 +145,12 @@ void WsTransport::close() {
 bool WsTransport::sendHttpRequest() {
 	changeState(State::Connecting);
 
-	auto seed = system_clock::now().time_since_epoch().count();
+	auto seed = static_cast<unsigned int>(system_clock::now().time_since_epoch().count());
 	random_bytes_engine generator(seed);
 
 	binary key(16);
 	auto k = reinterpret_cast<uint8_t *>(key.data());
-	std::generate(k, k + key.size(), generator);
+	std::generate(k, k + key.size(), [&]() { return uint8_t(generator()); });
 
 	const string request = "GET " + mPath +
 	                       " HTTP/1.1\r\n"
@@ -283,7 +283,7 @@ size_t WsTransport::readFrame(byte *buffer, size_t size, Frame &frame) {
 		cur += 4;
 	}
 
-	if (end - cur < frame.length)
+	if (size_t(end - cur) < frame.length)
 		return 0;
 
 	frame.payload = cur;
@@ -292,7 +292,7 @@ size_t WsTransport::readFrame(byte *buffer, size_t size, Frame &frame) {
 			frame.payload[i] ^= maskingKey[i % 4];
 	cur += frame.length;
 
-	return cur - buffer;
+	return size_t(cur - buffer);
 }
 
 void WsTransport::recvFrame(const Frame &frame) {
@@ -378,13 +378,13 @@ bool WsTransport::sendFrame(const Frame &frame) {
 	}
 
 	if (frame.mask) {
-		auto seed = system_clock::now().time_since_epoch().count();
+		auto seed = static_cast<unsigned int>(system_clock::now().time_since_epoch().count());
 		random_bytes_engine generator(seed);
 
 		byte *maskingKey = reinterpret_cast<byte *>(cur);
 
 		auto u = reinterpret_cast<uint8_t *>(maskingKey);
-		std::generate(u, u + 4, generator);
+		std::generate(u, u + 4, [&]() { return uint8_t(generator()); });
 		cur += 4;
 
 		for (size_t i = 0; i < frame.length; ++i)

test/benchmark.cpp

@@ -95,13 +95,12 @@ size_t benchmark(milliseconds duration) {
 	fill(messageData.begin(), messageData.end(), byte(0xFF));
 
 	atomic<size_t> receivedSize = 0;
-	atomic<bool> finished = false;
 
 	steady_clock::time_point startTime, openTime, receivedTime, endTime;
 
 	shared_ptr<DataChannel> dc2;
 	pc2->onDataChannel(
-	    [&dc2, &finished, &receivedSize, &receivedTime, &endTime](shared_ptr<DataChannel> dc) {
+	    [&dc2, &receivedSize, &receivedTime](shared_ptr<DataChannel> dc) {
 		    dc->onMessage([&receivedTime, &receivedSize](const variant<binary, string> &message) {
 			    if (holds_alternative<binary>(message)) {
 				    const auto &bin = get<binary>(message);
@@ -111,11 +110,7 @@ size_t benchmark(milliseconds duration) {
 			    }
 		    });
 
-		    dc->onClosed([&finished, &endTime]() {
-			    cout << "DataChannel closed." << endl;
-			    endTime = steady_clock::now();
-			    finished = true;
-		    });
+		    dc->onClosed([]() { cout << "DataChannel closed." << endl; });
 
 		    std::atomic_store(&dc2, dc);
 	    });
@@ -157,11 +152,9 @@ size_t benchmark(milliseconds duration) {
 		cout << "Received: " << receivedSize.load() / 1000 << " KB" << endl;
 	}
 
-	if (auto adc2 = std::atomic_load(&dc2)) {
 	dc1->close();
-		while (!finished && adc2->isOpen())
-			this_thread::sleep_for(100ms);
-	}
+
+	endTime = steady_clock::now();
 
 	auto connectDuration = duration_cast<milliseconds>(openTime - startTime);
 	auto transferDuration = duration_cast<milliseconds>(endTime - receivedTime);

test/capi.cpp

@@ -18,7 +18,6 @@
 
 #include <rtc/rtc.h>
 
-#include <cstdbool>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
@@ -136,7 +135,7 @@ static void deletePeer(Peer *peer) {
 int test_capi_main() {
 	int attempts;
 
-	rtcInitLogger(RTC_LOG_DEBUG);
+	rtcInitLogger(RTC_LOG_DEBUG, nullptr);
 
 	// Create peer 1
 	rtcConfiguration config1;
@@ -201,7 +200,7 @@ int test_capi_main() {
 
 	// You may call rtcCleanup() when finished to free static resources
 	rtcCleanup();
-	sleep(1);
+	sleep(2);
 
 	printf("Success\n");
 	return 0;

test/connectivity.cpp

@@ -150,7 +150,7 @@ void test_connectivity() {
 
 	// You may call rtc::Cleanup() when finished to free static resources
 	rtc::Cleanup();
-	this_thread::sleep_for(1s);
+	this_thread::sleep_for(2s);
 
 	cout << "Success" << endl;
 }

test/main.cpp

@@ -18,6 +18,7 @@
 
 #include <chrono>
 #include <iostream>
+#include <thread>
 
 using namespace std;
 using namespace chrono_literals;
@@ -71,7 +72,10 @@ int main(int argc, char **argv) {
 		cout << "*** Finished WebRTC benchmark" << endl;
 	} catch (const exception &e) {
 		cerr << "WebRTC benchmark failed: " << e.what() << endl;
+		std::this_thread::sleep_for(2s);
 		return -1;
 	}
+
+	std::this_thread::sleep_for(2s);
 	return 0;
 }

test/websocket.cpp

@@ -72,7 +72,7 @@ void test_websocket() {
 		throw runtime_error("Expected message not received");
 
 	ws->close();
-	this_thread::sleep_for(1s);
+	this_thread::sleep_for(2s);
 
 	// You may call rtc::Cleanup() when finished to free static resources
 	rtc::Cleanup();