Updated libjuice to v0.2.9

C API update and fixes

.github/workflows/build.yml

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

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.7)
 project (libdatachannel
 	DESCRIPTION "WebRTC DataChannels Library"
-	VERSION 0.4.5
+	VERSION 0.4.9
 	LANGUAGES CXX)
 
 option(USE_GNUTLS "Use GnuTLS instead of OpenSSL" OFF)
@@ -31,13 +31,35 @@ set(LIBDATACHANNEL_SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/src/description.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/dtlstransport.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/icetransport.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/init.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/log.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/peerconnection.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/rtc.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/sctptransport.cpp
 )
 
+set(LIBDATACHANNEL_HEADERS
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/candidate.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/channel.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/configuration.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/configuration.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/datachannel.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/description.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/include.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/init.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/log.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/message.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/peerconnection.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/queue.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/reliability.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtc.h
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtc.hpp
+)
+
 set(TESTS_SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/test/main.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/test/connectivity.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/test/capi.cpp
 )
 
 set(TESTS_OFFERER_SOURCES
@@ -52,6 +74,10 @@ set(THREADS_PREFER_PTHREAD_FLAG ON)
 find_package(Threads REQUIRED)
 
 add_subdirectory(deps/usrsctp EXCLUDE_FROM_ALL)
+if (CMAKE_CXX_COMPILER_ID MATCHES "GNU")
+    target_compile_options(usrsctp PRIVATE -Wno-error=format-truncation)
+	target_compile_options(usrsctp-static PRIVATE -Wno-error=format-truncation)
+endif()
 add_library(Usrsctp::Usrsctp ALIAS usrsctp)
 add_library(Usrsctp::UsrsctpStatic ALIAS usrsctp-static)
 
@@ -64,16 +90,7 @@ target_include_directories(datachannel PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/includ
 target_include_directories(datachannel PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include/rtc)
 target_include_directories(datachannel PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
 target_include_directories(datachannel PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/deps/plog/include)
-target_link_libraries(datachannel
+target_link_libraries(datachannel Threads::Threads Usrsctp::UsrsctpStatic)
-					Threads::Threads
-					Usrsctp::UsrsctpStatic
-					)
-if(WIN32)
-	target_link_libraries(datachannel
-						"wsock32"   #winsock2
-    					"ws2_32"    #winsock2
-						)
-endif()
 
 add_library(datachannel-static STATIC EXCLUDE_FROM_ALL ${LIBDATACHANNEL_SOURCES})
 set_target_properties(datachannel-static PROPERTIES
@@ -84,15 +101,11 @@ target_include_directories(datachannel-static PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}
 target_include_directories(datachannel-static PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include/rtc)
 target_include_directories(datachannel-static PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
 target_include_directories(datachannel-static PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/deps/plog/include)
-target_link_libraries(datachannel-static
+target_link_libraries(datachannel-static Threads::Threads Usrsctp::UsrsctpStatic)
-						Threads::Threads
+
-						Usrsctp::UsrsctpStatic
-						)
 if(WIN32)
-	target_link_libraries(datachannel-static
+	target_link_libraries(datachannel "wsock32" "ws2_32") # winsock2
-						"wsock32"   #winsock2
+	target_link_libraries(datachannel-static "wsock32" "ws2_32") # winsock2
-    					"ws2_32"    #winsock2
-						)
 endif()
 
 if (USE_GNUTLS)
@@ -134,6 +147,9 @@ endif()
 add_library(LibDataChannel::LibDataChannel ALIAS datachannel)
 add_library(LibDataChannel::LibDataChannelStatic ALIAS datachannel-static)
 
+install(TARGETS datachannel LIBRARY DESTINATION lib)
+install(FILES ${LIBDATACHANNEL_HEADERS} DESTINATION include/rtc)
+
 # Main Test
 add_executable(datachannel-tests ${TESTS_SOURCES})
 set_target_properties(datachannel-tests PROPERTIES

Makefile

@@ -5,7 +5,7 @@ CXX=$(CROSS)g++
 AR=$(CROSS)ar
 RM=rm -f
 CXXFLAGS=-std=c++17
-CPPFLAGS=-O2 -pthread -fPIC -Wall -Wno-address-of-packed-member
+CPPFLAGS=-O2 -pthread -fPIC -Wall
 LDFLAGS=-pthread
 LIBS=
 LOCALLIBS=libusrsctp.a
@@ -44,6 +44,9 @@ LDLIBS+=$(LOCALLIBS) $(shell pkg-config --libs $(LIBS))
 SRCS=$(shell printf "%s " src/*.cpp)
 OBJS=$(subst .cpp,.o,$(SRCS))
 
+TEST_SRCS=$(shell printf "%s " test/*.cpp)
+TEST_OBJS=$(subst .cpp,.o,$(TEST_SRCS))
+
 all: $(NAME).a $(NAME).so tests
 
 src/%.o: src/%.cpp
@@ -60,8 +63,8 @@ $(NAME).a: $(OBJS)
 $(NAME).so: $(LOCALLIBS) $(OBJS)
 	$(CXX) $(LDFLAGS) -shared -o $@ $(OBJS) $(LDLIBS)
 
-tests: $(NAME).a test/main.o
+tests: $(NAME).a $(TEST_OBJS)
-	$(CXX) $(LDFLAGS) -o $@ test/main.o $(NAME).a $(LDLIBS)
+	$(CXX) $(LDFLAGS) -o $@ $(TEST_OBJS) $(NAME).a $(LDLIBS)
 
 clean:
 	-$(RM) include/rtc/*.d *.d
@@ -83,7 +86,7 @@ dist-clean: clean
 libusrsctp.a:
 	cd $(USRSCTP_DIR) && \
 		./bootstrap && \
-		./configure --enable-static --disable-debug CFLAGS="$(CPPFLAGS)" && \
+		./configure --enable-static --disable-debug CFLAGS="$(CPPFLAGS) -Wno-error=format-truncation" && \
 		make
 	cp $(USRSCTP_DIR)/usrsctplib/.libs/libusrsctp.a .
 

README.md

@@ -79,11 +79,11 @@ MY_ON_RECV_CANDIDATE_FROM_REMOTE([pc](string candidate, string mid) {
 ### Observe the PeerConnection state
 
 ```cpp
-pc->onStateChanged([](PeerConnection::State state) {
+pc->onStateChange([](PeerConnection::State state) {
     cout << "State: " << state << endl;
 });
 
-pc->onGatheringStateChanged([](PeerConnection::GatheringState state) {
+pc->onGatheringStateChange([](PeerConnection::GatheringState state) {
     cout << "Gathering state: " << state << endl;
 });
 
@@ -114,5 +114,7 @@ pc->onDataChannel([&dc](shared_ptr<rtc::DataChannel> incoming) {
 
 ```
 
-See [test/main.cpp](https://github.com/paullouisageneau/libdatachannel/blob/master/test/main.cpp) for a complete local connection example.
+See [test/connectivity.cpp](https://github.com/paullouisageneau/libdatachannel/blob/master/test/connectivity.cpp) for a complete local connection example.
+
+See [test/cpai.cpp](https://github.com/paullouisageneau/libdatachannel/blob/master/test/capi.cpp) for a C API example.
 

include/rtc/channel.hpp

@@ -31,12 +31,10 @@ class Channel {
 public:
 	virtual void close() = 0;
 	virtual bool send(const std::variant<binary, string> &data) = 0; // returns false if buffered
-	virtual std::optional<std::variant<binary, string>> receive() = 0; // only if onMessage unset
 
 	virtual bool isOpen() const = 0;
 	virtual bool isClosed() const = 0;
-
+	virtual size_t maxMessageSize() const; // max message size in a call to send
-	virtual size_t availableAmount() const; // total size available to receive
 	virtual size_t bufferedAmount() const; // total size buffered to send
 
 	void onOpen(std::function<void()> callback);
@@ -47,11 +45,14 @@ public:
 	void onMessage(std::function<void(const binary &data)> binaryCallback,
 	               std::function<void(const string &data)> stringCallback);
 
-	void onAvailable(std::function<void()> callback);
 	void onBufferedAmountLow(std::function<void()> callback);
-
 	void setBufferedAmountLowThreshold(size_t amount);
 
+	// Extended API
+	virtual std::optional<std::variant<binary, string>> receive() = 0; // only if onMessage unset
+	virtual size_t availableAmount() const; // total size available to receive
+	void onAvailable(std::function<void()> callback);
+
 protected:
 	virtual void triggerOpen();
 	virtual void triggerClosed();

include/rtc/datachannel.hpp

@@ -38,33 +38,31 @@ class PeerConnection;
 
 class DataChannel : public std::enable_shared_from_this<DataChannel>, public Channel {
 public:
-	DataChannel(std::shared_ptr<PeerConnection> pc, unsigned int stream, string label,
+	DataChannel(std::weak_ptr<PeerConnection> pc, unsigned int stream, string label,
 	            string protocol, Reliability reliability);
-	DataChannel(std::shared_ptr<PeerConnection> pc, std::shared_ptr<SctpTransport> transport,
+	DataChannel(std::weak_ptr<PeerConnection> pc, std::shared_ptr<SctpTransport> transport,
 	            unsigned int stream);
 	~DataChannel();
 
-	void close(void) override;
-
-	bool send(const std::variant<binary, string> &data) override;
-	bool send(const byte *data, size_t size);
-
-	template <typename Buffer> bool sendBuffer(const Buffer &buf);
-	template <typename Iterator> bool sendBuffer(Iterator first, Iterator last);
-
-	std::optional<std::variant<binary, string>> receive() override;
-
-	bool isOpen(void) const override;
-	bool isClosed(void) const override;
-	size_t availableAmount() const override;
-
-	size_t maxMessageSize() const;  // maximum message size in a call to send or sendBuffer
-
 	unsigned int stream() const;
 	string label() const;
 	string protocol() const;
 	Reliability reliability() const;
 
+	void close(void) override;
+	bool send(const std::variant<binary, string> &data) override;
+	bool send(const byte *data, size_t size);
+	template <typename Buffer> bool sendBuffer(const Buffer &buf);
+	template <typename Iterator> bool sendBuffer(Iterator first, Iterator last);
+
+	bool isOpen(void) const override;
+	bool isClosed(void) const override;
+	size_t maxMessageSize() const override;
+
+	// Extended API
+	size_t availableAmount() const override;
+	std::optional<std::variant<binary, string>> receive() override;
+
 private:
 	void remoteClose();
 	void open(std::shared_ptr<SctpTransport> sctpTransport);
@@ -72,7 +70,7 @@ private:
 	void incoming(message_ptr message);
 	void processOpenMessage(message_ptr message);
 
-	const std::shared_ptr<PeerConnection> mPeerConnection;
+	const std::weak_ptr<PeerConnection> mPeerConnection;
 	std::shared_ptr<SctpTransport> mSctpTransport;
 
 	unsigned int mStream;

include/rtc/include.hpp

@@ -20,11 +20,14 @@
 #define RTC_INCLUDE_H
 
 #ifdef _WIN32
+#define WIN32_LEAN_AND_MEAN
 #ifndef _WIN32_WINNT
 #define _WIN32_WINNT 0x0602
 #endif
 #endif
 
+#include "log.hpp"
+
 #include <cstddef>
 #include <functional>
 #include <memory>
@@ -33,9 +36,6 @@
 #include <string>
 #include <vector>
 
-#include "plog/Appenders/ColorConsoleAppender.h"
-#include "plog/Log.h"
-
 namespace rtc {
 
 using std::byte;
@@ -50,8 +50,6 @@ using std::uint32_t;
 using std::uint64_t;
 using std::uint8_t;
 
-// Constants
-
 const size_t MAX_NUMERICNODE_LEN = 48; // Max IPv6 string representation length
 const size_t MAX_NUMERICSERV_LEN = 6;  // Max port string representation length
 
@@ -59,29 +57,6 @@ 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
 
-// Log
-
-enum class LogLevel { // Don't change, it must match plog severity
-	None = 0,
-	Fatal = 1,
-	Error = 2,
-	Warning = 3,
-	Info = 4,
-	Debug = 5,
-	Verbose = 6
-};
-
-inline void InitLogger(plog::Severity severity, plog::IAppender *appender = nullptr) {
-	static plog::ColorConsoleAppender<plog::TxtFormatter> consoleAppender;
-	if (!appender)
-		appender = &consoleAppender;
-	plog::init(severity, appender);
-	PLOG_DEBUG << "Logger initialized";
-}
-
-inline void InitLogger(LogLevel level) { InitLogger(static_cast<plog::Severity>(level)); }
-
-// Utils
 
 template <class... Ts> struct overloaded : Ts... { using Ts::operator()...; };
 template <class... Ts> overloaded(Ts...)->overloaded<Ts...>;
@@ -89,6 +64,7 @@ template <class... Ts> overloaded(Ts...)->overloaded<Ts...>;
 template <typename... P> class synchronized_callback {
 public:
 	synchronized_callback() = default;
+	synchronized_callback(std::function<void(P...)> func) { *this = std::move(func); };
 	~synchronized_callback() { *this = nullptr; }
 
 	synchronized_callback &operator=(std::function<void(P...)> func) {

include/rtc/init.hpp

@@ -0,0 +1,50 @@
+/**
+ * Copyright (c) 2020 Paul-Louis Ageneau
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#ifndef RTC_INIT_H
+#define RTC_INIT_H
+
+#include "include.hpp"
+
+#include <mutex>
+
+namespace rtc {
+
+class Init;
+using init_token = std::shared_ptr<Init>;
+
+class Init {
+public:
+	static init_token Token();
+	static void Cleanup();
+
+	~Init();
+
+private:
+	Init();
+
+	static std::weak_ptr<Init> Weak;
+	static init_token Global;
+	static std::mutex Mutex;
+};
+
+inline void Cleanup() { Init::Cleanup(); }
+
+} // namespace rtc
+
+#endif

include/rtc/log.hpp

@@ -0,0 +1,40 @@
+/**
+ * Copyright (c) 2019 Paul-Louis Ageneau
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#ifndef RTC_LOG_H
+#define RTC_LOG_H
+
+#include "plog/Log.h"
+
+namespace rtc {
+
+enum class LogLevel { // Don't change, it must match plog severity
+	None = 0,
+	Fatal = 1,
+	Error = 2,
+	Warning = 3,
+	Info = 4,
+	Debug = 5,
+	Verbose = 6
+};
+
+void InitLogger(LogLevel level);
+void InitLogger(plog::Severity severity, plog::IAppender *appender = nullptr);
+}
+
+#endif

include/rtc/peerconnection.hpp

@@ -24,6 +24,7 @@
 #include "datachannel.hpp"
 #include "description.hpp"
 #include "include.hpp"
+#include "init.hpp"
 #include "message.hpp"
 #include "reliability.hpp"
 #include "rtc.hpp"
@@ -32,6 +33,7 @@
 #include <functional>
 #include <list>
 #include <mutex>
+#include <shared_mutex>
 #include <thread>
 #include <unordered_map>
 
@@ -87,6 +89,8 @@ public:
 	void onGatheringStateChange(std::function<void(GatheringState state)> callback);
 
 private:
+	init_token mInitToken = Init::Token();
+
 	std::shared_ptr<IceTransport> initIceTransport(Description::Role role);
 	std::shared_ptr<DtlsTransport> initDtlsTransport();
 	std::shared_ptr<SctpTransport> initSctpTransport();
@@ -95,6 +99,11 @@ private:
 	bool checkFingerprint(const std::string &fingerprint) const;
 	void forwardMessage(message_ptr message);
 	void forwardBufferedAmount(uint16_t stream, size_t amount);
+
+	std::shared_ptr<DataChannel> emplaceDataChannel(Description::Role role, const string &label,
+	                                                const string &protocol,
+	                                                const Reliability &reliability);
+	std::shared_ptr<DataChannel> findDataChannel(uint16_t stream);
 	void iterateDataChannels(std::function<void(std::shared_ptr<DataChannel> channel)> func);
 	void openDataChannels();
 	void closeDataChannels();
@@ -118,6 +127,7 @@ private:
 	std::recursive_mutex mInitMutex;
 
 	std::unordered_map<unsigned int, std::weak_ptr<DataChannel>> mDataChannels;
+	std::shared_mutex mDataChannelsMutex;
 
 	std::atomic<State> mState;
 	std::atomic<GatheringState> mGatheringState;

include/rtc/queue.hpp

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

include/rtc/rtc.h

@@ -33,13 +33,13 @@ typedef enum {
 	RTC_FAILED = 4,
 	RTC_CLOSED = 5,
 	RTC_DESTROYING = 6 // internal
-} rtc_state_t;
+} rtcState;
 
 typedef enum {
 	RTC_GATHERING_NEW = 0,
 	RTC_GATHERING_INPROGRESS = 1,
 	RTC_GATHERING_COMPLETE = 2
-} rtc_gathering_state_t;
+} rtcGatheringState;
 
 // Don't change, it must match plog severity
 typedef enum {
@@ -50,32 +50,67 @@ typedef enum {
 	RTC_LOG_INFO = 4,
 	RTC_LOG_DEBUG = 5,
 	RTC_LOG_VERBOSE = 6
-} rtc_log_level_t;
+} rtcLogLevel;
 
-void rtcInitLogger(rtc_log_level_t level);
+typedef struct {
+	const char **iceServers;
+	int iceServersCount;
+} rtcConfiguration;
 
-int rtcCreatePeerConnection(const char **iceServers, int iceServersCount);
+typedef void (*dataChannelCallbackFunc)(int dc, void *ptr);
-void rtcDeletePeerConnection(int pc);
+typedef void (*descriptionCallbackFunc)(const char *sdp, const char *type, void *ptr);
-int rtcCreateDataChannel(int pc, const char *label);
+typedef void (*candidateCallbackFunc)(const char *cand, const char *mid, void *ptr);
-void rtcDeleteDataChannel(int dc);
+typedef void (*stateChangeCallbackFunc)(rtcState state, void *ptr);
-void rtcSetDataChannelCallback(int pc, void (*dataChannelCallback)(int, void *));
+typedef void (*gatheringStateCallbackFunc)(rtcGatheringState state, void *ptr);
-void rtcSetLocalDescriptionCallback(int pc, void (*descriptionCallback)(const char *, const char *,
+typedef void (*openCallbackFunc)(void *ptr);
-                                                                        void *));
+typedef void (*closedCallbackFunc)(void *ptr);
-void rtcSetLocalCandidateCallback(int pc,
+typedef void (*errorCallbackFunc)(const char *error, void *ptr);
-                                  void (*candidateCallback)(const char *, const char *, void *));
+typedef void (*messageCallbackFunc)(const char *message, int size, void *ptr);
-void rtcSetStateChangeCallback(int pc, void (*stateCallback)(rtc_state_t state, void *));
+typedef void (*bufferedAmountLowCallbackFunc)(void *ptr);
-void rtcSetGatheringStateChangeCallback(int pc,
+typedef void (*availableCallbackFunc)(void *ptr);
-                                        void (*gatheringStateCallback)(rtc_gathering_state_t state,
+
-                                                                       void *));
+// Log
-void rtcSetRemoteDescription(int pc, const char *sdp, const char *type);
+void rtcInitLogger(rtcLogLevel level);
-void rtcAddRemoteCandidate(int pc, const char *candidate, const char *mid);
+
-int rtcGetDataChannelLabel(int dc, char *data, int size);
+// User pointer
-void rtcSetOpenCallback(int dc, void (*openCallback)(void *));
-void rtcSetErrorCallback(int dc, void (*errorCallback)(const char *, void *));
-void rtcSetMessageCallback(int dc, void (*messageCallback)(const char *, int, void *));
-int rtcSendMessage(int dc, const char *data, int size);
 void rtcSetUserPointer(int i, void *ptr);
 
+// PeerConnection
+int rtcCreatePeerConnection(const rtcConfiguration *config);
+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);
+
+int rtcSetRemoteDescription(int pc, const char *sdp, const char *type);
+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);
+
+// DataChannel
+int rtcCreateDataChannel(int pc, const char *label);
+int rtcDeleteDataChannel(int dc);
+
+int rtcGetDataChannelLabel(int dc, char *buffer, int size);
+int rtcSetOpenCallback(int dc, openCallbackFunc cb);
+int rtcSetClosedCallback(int dc, closedCallbackFunc cb);
+int rtcSetErrorCallback(int dc, errorCallbackFunc cb);
+int rtcSetMessageCallback(int dc, messageCallbackFunc cb);
+int rtcSendMessage(int dc, const char *data, int size);
+
+int rtcGetBufferedAmount(int dc); // total size buffered to send
+int rtcSetBufferedAmountLowThreshold(int dc, int amount);
+int rtcSetBufferedAmountLowCallback(int dc, bufferedAmountLowCallbackFunc cb);
+
+// DataChannel extended API
+int rtcGetAvailableAmount(int dc); // total size available to receive
+int rtcSetAvailableCallback(int dc, availableCallbackFunc cb);
+int rtcReceiveMessage(int dc, char *buffer, int *size);
+
 #ifdef __cplusplus
 } // extern "C"
 #endif

include/rtc/rtc.hpp

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

src/candidate.cpp

@@ -25,7 +25,7 @@
 #ifdef _WIN32
 #include <winsock2.h>
 #include <ws2tcpip.h>
-#elif __linux__
+#else
 #include <netdb.h>
 #include <sys/socket.h>
 #endif

src/certificate.cpp

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

src/channel.cpp

@@ -18,10 +18,14 @@
 
 #include "channel.hpp"
 
-namespace {}
-
 namespace rtc {
 
+size_t Channel::maxMessageSize() const { return DEFAULT_MAX_MESSAGE_SIZE; }
+
+size_t Channel::bufferedAmount() const { return mBufferedAmount; }
+
+size_t Channel::availableAmount() const { return 0; }
+
 void Channel::onOpen(std::function<void()> callback) {
 	mOpenCallback = callback;
 }
@@ -49,20 +53,16 @@ void Channel::onMessage(std::function<void(const binary &data)> binaryCallback,
 	});
 }
 
-void Channel::onAvailable(std::function<void()> callback) {
-	mAvailableCallback = callback;
-}
-
 void Channel::onBufferedAmountLow(std::function<void()> callback) {
 	mBufferedAmountLowCallback = callback;
 }
 
-size_t Channel::availableAmount() const { return 0; }
-
-size_t Channel::bufferedAmount() const { return mBufferedAmount; }
-
 void Channel::setBufferedAmountLowThreshold(size_t amount) { mBufferedAmountLowThreshold = amount; }
 
+void Channel::onAvailable(std::function<void()> callback) {
+	mAvailableCallback = callback;
+}
+
 void Channel::triggerOpen() { mOpenCallback(); }
 
 void Channel::triggerClosed() { mClosedCallback(); }

src/datachannel.cpp

@@ -21,9 +21,16 @@
 #include "peerconnection.hpp"
 #include "sctptransport.hpp"
 
+#ifdef _WIN32
+#include <winsock2.h>
+#else
+#include <arpa/inet.h>
+#endif
+
 namespace rtc {
 
 using std::shared_ptr;
+using std::weak_ptr;
 
 // Messages for the DataChannel establishment protocol
 // See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09
@@ -60,16 +67,16 @@ struct CloseMessage {
 
 const size_t RECV_QUEUE_LIMIT = 1024 * 1024; // 1 MiB
 
-DataChannel::DataChannel(shared_ptr<PeerConnection> pc, unsigned int stream, string label,
+DataChannel::DataChannel(weak_ptr<PeerConnection> pc, unsigned int stream, string label,
                          string protocol, Reliability reliability)
-    : mPeerConnection(std::move(pc)), mStream(stream), mLabel(std::move(label)),
+    : mPeerConnection(pc), mStream(stream), mLabel(std::move(label)),
       mProtocol(std::move(protocol)),
       mReliability(std::make_shared<Reliability>(std::move(reliability))),
       mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
 
-DataChannel::DataChannel(shared_ptr<PeerConnection> pc, shared_ptr<SctpTransport> transport,
+DataChannel::DataChannel(weak_ptr<PeerConnection> pc, shared_ptr<SctpTransport> transport,
                          unsigned int stream)
-    : mPeerConnection(std::move(pc)), mSctpTransport(transport), mStream(stream),
+    : mPeerConnection(pc), mSctpTransport(transport), mStream(stream),
       mReliability(std::make_shared<Reliability>()),
       mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
 
@@ -77,6 +84,14 @@ DataChannel::~DataChannel() {
 	close();
 }
 
+unsigned int DataChannel::stream() const { return mStream; }
+
+string DataChannel::label() const { return mLabel; }
+
+string DataChannel::protocol() const { return mProtocol; }
+
+Reliability DataChannel::reliability() const { return *mReliability; }
+
 void DataChannel::close() {
 	if (mIsOpen.exchange(false) && mSctpTransport)
 		mSctpTransport->reset(mStream);
@@ -131,24 +146,17 @@ bool DataChannel::isOpen(void) const { return mIsOpen; }
 
 bool DataChannel::isClosed(void) const { return mIsClosed; }
 
-size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
-
 size_t DataChannel::maxMessageSize() const {
 	size_t max = DEFAULT_MAX_MESSAGE_SIZE;
-	if (auto description = mPeerConnection->remoteDescription())
+	if (auto pc = mPeerConnection.lock())
-		if (auto maxMessageSize = description->maxMessageSize())
+		if (auto description = pc->remoteDescription())
-			return *maxMessageSize > 0 ? *maxMessageSize : LOCAL_MAX_MESSAGE_SIZE;
+			if (auto maxMessageSize = description->maxMessageSize())
+				return *maxMessageSize > 0 ? *maxMessageSize : LOCAL_MAX_MESSAGE_SIZE;
 
 	return std::min(max, LOCAL_MAX_MESSAGE_SIZE);
 }
 
-unsigned int DataChannel::stream() const { return mStream; }
+size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
-
-string DataChannel::label() const { return mLabel; }
-
-string DataChannel::protocol() const { return mProtocol; }
-
-Reliability DataChannel::reliability() const { return *mReliability; }
 
 void DataChannel::open(shared_ptr<SctpTransport> sctpTransport) {
 	mSctpTransport = sctpTransport;

src/dtlstransport.cpp

@@ -55,6 +55,14 @@ static bool check_gnutls(int ret, const string &message = "GnuTLS error") {
 
 namespace rtc {
 
+void DtlsTransport::Init() {
+	// Nothing to do
+}
+
+void DtlsTransport::Cleanup() {
+	// Nothing to do
+}
+
 DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certificate> certificate,
                              verifier_callback verifierCallback,
                              state_callback stateChangeCallback)
@@ -131,10 +139,13 @@ bool DtlsTransport::send(message_ptr message) {
 }
 
 void DtlsTransport::incoming(message_ptr message) {
-	if (message)
+	if (!message) {
-		mIncomingQueue.push(message);
-	else
 		mIncomingQueue.stop();
+		return;
+	}
+
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	mIncomingQueue.push(message);
 }
 
 void DtlsTransport::changeState(State state) {
@@ -262,10 +273,8 @@ ssize_t DtlsTransport::ReadCallback(gnutls_transport_ptr_t ptr, void *data, size
 
 int DtlsTransport::TimeoutCallback(gnutls_transport_ptr_t ptr, unsigned int ms) {
 	DtlsTransport *t = static_cast<DtlsTransport *>(ptr);
-	if (ms != GNUTLS_INDEFINITE_TIMEOUT)
+	t->mIncomingQueue.wait(ms != GNUTLS_INDEFINITE_TIMEOUT ? std::make_optional(milliseconds(ms))
-		t->mIncomingQueue.wait(milliseconds(ms));
+	                                                       : nullopt);
-	else
-		t->mIncomingQueue.wait();
 	return !t->mIncomingQueue.empty() ? 1 : 0;
 }
 
@@ -323,7 +332,7 @@ BIO_METHOD *DtlsTransport::BioMethods = NULL;
 int DtlsTransport::TransportExIndex = -1;
 std::mutex DtlsTransport::GlobalMutex;
 
-void DtlsTransport::GlobalInit() {
+void DtlsTransport::Init() {
 	std::lock_guard lock(GlobalMutex);
 	if (!BioMethods) {
 		BioMethods = BIO_meth_new(BIO_TYPE_BIO, "DTLS writer");
@@ -339,6 +348,10 @@ void DtlsTransport::GlobalInit() {
 	}
 }
 
+void DtlsTransport::Cleanup() {
+	// Nothing to do
+}
+
 DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certificate> certificate,
                              verifier_callback verifierCallback, state_callback stateChangeCallback)
     : Transport(lower), mCertificate(certificate), mState(State::Disconnected),
@@ -346,7 +359,6 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
       mStateChangeCallback(std::move(stateChangeCallback)) {
 
 	PLOG_DEBUG << "Initializing DTLS transport (OpenSSL)";
-	GlobalInit();
 
 	if (!(mCtx = SSL_CTX_new(DTLS_method())))
 		throw std::runtime_error("Unable to create SSL context");
@@ -432,10 +444,13 @@ bool DtlsTransport::send(message_ptr message) {
 }
 
 void DtlsTransport::incoming(message_ptr message) {
-	if (message)
+	if (!message) {
-		mIncomingQueue.push(message);
-	else
 		mIncomingQueue.stop();
+		return;
+	}
+
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	mIncomingQueue.push(message);
 }
 
 void DtlsTransport::changeState(State state) {
@@ -448,29 +463,47 @@ void DtlsTransport::runRecvLoop() {
 	try {
 		changeState(State::Connecting);
 
-		SSL_do_handshake(mSsl);
+		int ret = SSL_do_handshake(mSsl);
+		check_openssl_ret(mSsl, ret, "Handshake failed");
 
 		const size_t bufferSize = maxMtu;
 		byte buffer[bufferSize];
-		while (auto next = mIncomingQueue.pop()) {
+		while (true) {
-			auto message = *next;
+			std::optional<milliseconds> duration;
-			BIO_write(mInBio, message->data(), message->size());
+			struct timeval timeout = {};
-			int ret = SSL_read(mSsl, buffer, bufferSize);
+			if (DTLSv1_get_timeout(mSsl, &timeout))
-			if (!check_openssl_ret(mSsl, ret))
+				duration = milliseconds(timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
-				break;
 
-			auto decrypted = ret > 0 ? make_message(buffer, buffer + ret) : nullptr;
+			if (!mIncomingQueue.wait(duration))
+				break; // queue is stopped
+
+			message_ptr decrypted;
+			if (!mIncomingQueue.empty()) {
+				auto message = *mIncomingQueue.pop();
+				BIO_write(mInBio, message->data(), message->size());
+
+				int ret = SSL_read(mSsl, buffer, bufferSize);
+				if (!check_openssl_ret(mSsl, ret))
+					break;
+
+				if (ret > 0)
+					decrypted = make_message(buffer, buffer + ret);
+			}
 
 			if (mState == State::Connecting) {
-				if (unsigned long err = ERR_get_error())
-					throw std::runtime_error("handshake failed: " + openssl_error_string(err));
-
 				if (SSL_is_init_finished(mSsl)) {
 					changeState(State::Connected);
 
 					// RFC 8261: DTLS MUST support sending messages larger than the current path MTU
 					// See https://tools.ietf.org/html/rfc8261#section-5
 					SSL_set_mtu(mSsl, maxMtu + 1);
+				} else {
+					// Continue the handshake
+					int ret = SSL_do_handshake(mSsl);
+					if (!check_openssl_ret(mSsl, ret, "Handshake failed"))
+						break;
+
+					DTLSv1_handle_timeout(mSsl);
 				}
 			}
 
@@ -486,7 +519,7 @@ void DtlsTransport::runRecvLoop() {
 		changeState(State::Disconnected);
 		recv(nullptr);
 	} else {
-		PLOG_INFO << "DTLS handshake failed";
+		PLOG_ERROR << "DTLS handshake failed";
 		changeState(State::Failed);
 	}
 }

src/dtlstransport.hpp

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

src/icetransport.cpp

@@ -19,9 +19,15 @@
 #include "icetransport.hpp"
 #include "configuration.hpp"
 
+#include <iostream>
+#include <random>
+#include <sstream>
+
 #ifdef _WIN32
 #include <winsock2.h>
-#elif __linux__
+#include <ws2tcpip.h>
+#else
+#include <arpa/inet.h>
 #include <netdb.h>
 #include <netinet/in.h>
 #include <sys/socket.h>
@@ -29,10 +35,6 @@
 
 #include <sys/types.h>
 
-#include <iostream>
-#include <random>
-#include <sstream>
-
 using namespace std::chrono_literals;
 
 using std::shared_ptr;
@@ -160,7 +162,10 @@ bool IceTransport::send(message_ptr message) {
 	return outgoing(message);
 }
 
-void IceTransport::incoming(message_ptr message) { recv(message); }
+void IceTransport::incoming(message_ptr message) {
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	recv(message);
+}
 
 void IceTransport::incoming(const byte *data, int size) {
 	incoming(make_message(data, data + size));
@@ -243,11 +248,8 @@ void IceTransport::LogCallback(juice_log_level_t level, const char *message) {
 	case JUICE_LOG_LEVEL_INFO:
 		severity = plog::info;
 		break;
-	case JUICE_LOG_LEVEL_DEBUG:
-		severity = plog::debug;
-		break;
 	default:
-		severity = plog::verbose;
+		severity = plog::verbose; // libjuice debug as verbose
 		break;
 	}
 	PLOG(severity) << "juice: " << message;
@@ -517,7 +519,10 @@ bool IceTransport::send(message_ptr message) {
 	return outgoing(message);
 }
 
-void IceTransport::incoming(message_ptr message) { recv(message); }
+void IceTransport::incoming(message_ptr message) {
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	recv(message);
+}
 
 void IceTransport::incoming(const byte *data, int size) {
 	incoming(make_message(data, data + size));

src/init.cpp

@@ -0,0 +1,86 @@
+/**
+ * Copyright (c) 2020 Paul-Louis Ageneau
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include "init.hpp"
+
+#include "dtlstransport.hpp"
+#include "sctptransport.hpp"
+
+#ifdef _WIN32
+#include <winsock2.h>
+#endif
+
+#if USE_GNUTLS
+// Nothing to do
+#else
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+#endif
+
+using std::shared_ptr;
+
+namespace rtc {
+
+std::weak_ptr<Init> Init::Weak;
+init_token Init::Global;
+std::mutex Init::Mutex;
+
+init_token Init::Token() {
+	std::lock_guard lock(Mutex);
+
+	if (!Global) {
+		if (auto token = Weak.lock())
+			Global = token;
+		else
+			Global = shared_ptr<Init>(new Init());
+	}
+	return Global;
+}
+
+void Init::Cleanup() { Global.reset(); }
+
+Init::Init() {
+#ifdef _WIN32
+	WSADATA wsaData;
+	if (WSAStartup(MAKEWORD(2, 2), &wsaData))
+		throw std::runtime_error("WSAStartup failed, error=" + std::to_string(WSAGetLastError()));
+#endif
+
+#if USE_GNUTLS
+		// Nothing to do
+#else
+	OPENSSL_init_ssl(0, NULL);
+	SSL_load_error_strings();
+	ERR_load_crypto_strings();
+#endif
+
+	DtlsTransport::Init();
+	SctpTransport::Init();
+}
+
+Init::~Init() {
+	DtlsTransport::Cleanup();
+	SctpTransport::Cleanup();
+
+#ifdef _WIN32
+	WSACleanup();
+#endif
+}
+
+} // namespace rtc
+

src/log.cpp

@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2019-2020 Paul-Louis Ageneau
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include "log.hpp"
+
+#include "plog/Appenders/ColorConsoleAppender.h"
+#include "plog/Log.h"
+#include "plog/Logger.h"
+
+namespace rtc {
+
+void InitLogger(LogLevel level) { InitLogger(static_cast<plog::Severity>(level)); }
+
+void InitLogger(plog::Severity severity, plog::IAppender *appender) {
+	static plog::ColorConsoleAppender<plog::TxtFormatter> consoleAppender;
+	static plog::Logger<0> *logger = nullptr;
+	if (!logger) {
+		logger = &plog::init(severity, appender ? appender : &consoleAppender);
+		PLOG_DEBUG << "Logger initialized";
+	} else {
+		logger->setMaxSeverity(severity);
+		if (appender)
+			logger->addAppender(appender);
+	}
+}
+}
+

src/peerconnection.cpp

@@ -32,7 +32,8 @@ using namespace std::placeholders;
 using std::shared_ptr;
 using std::weak_ptr;
 
-PeerConnection::PeerConnection() : PeerConnection(Configuration()) {}
+PeerConnection::PeerConnection() : PeerConnection(Configuration()) {
+}
 
 PeerConnection::PeerConnection(const Configuration &config)
     : mConfig(config), mCertificate(make_certificate("libdatachannel")), mState(State::New) {}
@@ -48,7 +49,6 @@ PeerConnection::~PeerConnection() {
 void PeerConnection::close() {
 	// Close DataChannels
 	closeDataChannels();
-	mDataChannels.clear();
 
 	// Close Transports
 	for (int i = 0; i < 2; ++i) { // Make sure a transport wasn't spawn behind our back
@@ -101,12 +101,16 @@ void PeerConnection::setRemoteDescription(Description description) {
 		if (!sctpTransport && iceTransport->role() == Description::Role::Active) {
 			// Since we assumed passive role during DataChannel creation, we need to shift the
 			// stream numbers by one to shift them from odd to even.
+			std::unique_lock lock(mDataChannelsMutex);
 			decltype(mDataChannels) newDataChannels;
-			iterateDataChannels([&](shared_ptr<DataChannel> channel) {
+			auto it = mDataChannels.begin();
+			while (it != mDataChannels.end()) {
+				auto channel = it->second.lock();
 				if (channel->stream() % 2 == 1)
 					channel->mStream -= 1;
 				newDataChannels.emplace(channel->stream(), channel);
-			});
+				++it;
+			}
 			std::swap(mDataChannels, newDataChannels);
 		}
 	}
@@ -158,19 +162,7 @@ shared_ptr<DataChannel> PeerConnection::createDataChannel(const string &label,
 	auto iceTransport = std::atomic_load(&mIceTransport);
 	auto role = iceTransport ? iceTransport->role() : Description::Role::Passive;
 
-	// The active side must use streams with even identifiers, whereas the passive side must use
+	auto channel = emplaceDataChannel(role, label, protocol, reliability);
-	// streams with odd identifiers.
-	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09#section-6
-	unsigned int stream = (role == Description::Role::Active) ? 0 : 1;
-	while (mDataChannels.find(stream) != mDataChannels.end()) {
-		stream += 2;
-		if (stream >= 65535)
-			throw std::runtime_error("Too many DataChannels");
-	}
-
-	auto channel =
-	    std::make_shared<DataChannel>(shared_from_this(), stream, label, protocol, reliability);
-	mDataChannels.insert(std::make_pair(stream, channel));
 
 	if (!iceTransport) {
 		// RFC 5763: The endpoint that is the offerer MUST use the setup attribute value of
@@ -353,14 +345,7 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		return;
 	}
 
-	shared_ptr<DataChannel> channel;
+	auto channel = findDataChannel(message->stream);
-	if (auto it = mDataChannels.find(message->stream); it != mDataChannels.end()) {
-		channel = it->second.lock();
-		if (!channel || channel->isClosed()) {
-			mDataChannels.erase(it);
-			channel = nullptr;
-		}
-	}
 
 	auto iceTransport = std::atomic_load(&mIceTransport);
 	auto sctpTransport = std::atomic_load(&mSctpTransport);
@@ -388,29 +373,54 @@ void PeerConnection::forwardMessage(message_ptr message) {
 }
 
 void PeerConnection::forwardBufferedAmount(uint16_t stream, size_t amount) {
+	if (auto channel = findDataChannel(stream))
+		channel->triggerBufferedAmount(amount);
+}
+
+shared_ptr<DataChannel> PeerConnection::emplaceDataChannel(Description::Role role,
+                                                           const string &label,
+                                                           const string &protocol,
+                                                           const Reliability &reliability) {
+	// The active side must use streams with even identifiers, whereas the passive side must use
+	// streams with odd identifiers.
+	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09#section-6
+	std::unique_lock lock(mDataChannelsMutex);
+	unsigned int stream = (role == Description::Role::Active) ? 0 : 1;
+	while (mDataChannels.find(stream) != mDataChannels.end()) {
+		stream += 2;
+		if (stream >= 65535)
+			throw std::runtime_error("Too many DataChannels");
+	}
+	auto channel =
+	    std::make_shared<DataChannel>(shared_from_this(), stream, label, protocol, reliability);
+	mDataChannels.emplace(std::make_pair(stream, channel));
+	return channel;
+}
+
+shared_ptr<DataChannel> PeerConnection::findDataChannel(uint16_t stream) {
+	std::shared_lock lock(mDataChannelsMutex);
 	shared_ptr<DataChannel> channel;
 	if (auto it = mDataChannels.find(stream); it != mDataChannels.end()) {
 		channel = it->second.lock();
-		if (!channel || channel->isClosed()) {
+		if (!channel)
 			mDataChannels.erase(it);
-			channel = nullptr;
-		}
 	}
-
+	return channel;
-	if (channel)
-		channel->triggerBufferedAmount(amount);
 }
 
 void PeerConnection::iterateDataChannels(
     std::function<void(shared_ptr<DataChannel> channel)> func) {
+	std::shared_lock lock(mDataChannelsMutex);
 	auto it = mDataChannels.begin();
 	while (it != mDataChannels.end()) {
 		auto channel = it->second.lock();
-		if (!channel || channel->isClosed()) {
+		if (!channel) {
 			it = mDataChannels.erase(it);
 			continue;
 		}
-		func(channel);
+		if (!channel->isClosed()) {
+			func(channel);
+		}
 		++it;
 	}
 }

src/rtc.cpp

@@ -22,191 +22,82 @@
 
 #include <rtc.h>
 
+#include <exception>
+#include <mutex>
 #include <unordered_map>
-
+#include <utility>
-#include <plog/Appenders/ColorConsoleAppender.h>
 
 using namespace rtc;
 using std::shared_ptr;
 using std::string;
 
+#define CATCH(statement)                                                                           \
+	try {                                                                                          \
+		statement;                                                                                 \
+	} catch (const std::exception &e) {                                                            \
+		PLOG_ERROR << e.what();                                                                    \
+		return -1;                                                                                 \
+	}
+
 namespace {
 
 std::unordered_map<int, shared_ptr<PeerConnection>> peerConnectionMap;
 std::unordered_map<int, shared_ptr<DataChannel>> dataChannelMap;
 std::unordered_map<int, void *> userPointerMap;
+std::mutex mutex;
 int lastId = 0;
 
 void *getUserPointer(int id) {
+	std::lock_guard lock(mutex);
 	auto it = userPointerMap.find(id);
 	return it != userPointerMap.end() ? it->second : nullptr;
 }
 
-} // namespace
+shared_ptr<PeerConnection> getPeerConnection(int id) {
+	std::lock_guard lock(mutex);
+	auto it = peerConnectionMap.find(id);
+	return it != peerConnectionMap.end() ? it->second : nullptr;
+}
 
-void rtcInitLogger(rtc_log_level_t level) { InitLogger(static_cast<LogLevel>(level)); }
+shared_ptr<DataChannel> getDataChannel(int id) {
+	std::lock_guard lock(mutex);
+	auto it = dataChannelMap.find(id);
+	return it != dataChannelMap.end() ? it->second : nullptr;
+}
 
-int rtcCreatePeerConnection(const char **iceServers, int iceServersCount) {
+int emplacePeerConnection(shared_ptr<PeerConnection> ptr) {
-	Configuration config;
+	std::lock_guard lock(mutex);
-	for (int i = 0; i < iceServersCount; ++i) {
-		config.iceServers.emplace_back(IceServer(string(iceServers[i])));
-	}
 	int pc = ++lastId;
-	peerConnectionMap.emplace(std::make_pair(pc, std::make_shared<PeerConnection>(config)));
+	peerConnectionMap.emplace(std::make_pair(pc, ptr));
 	return pc;
 }
 
-void rtcDeletePeerConnection(int pc) { peerConnectionMap.erase(pc); }
+int emplaceDataChannel(shared_ptr<DataChannel> ptr) {
-
+	std::lock_guard lock(mutex);
-int rtcCreateDataChannel(int pc, const char *label) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return 0;
-	auto dataChannel = it->second->createDataChannel(string(label));
 	int dc = ++lastId;
-	dataChannelMap.emplace(std::make_pair(dc, dataChannel));
+	dataChannelMap.emplace(std::make_pair(dc, ptr));
 	return dc;
 }
 
-void rtcDeleteDataChannel(int dc) { dataChannelMap.erase(dc); }
+bool erasePeerConnection(int pc) {
-
+	std::lock_guard lock(mutex);
-void rtcSetDataChannelCallback(int pc, void (*dataChannelCallback)(int, void *)) {
+	if (peerConnectionMap.erase(pc) == 0)
-	auto it = peerConnectionMap.find(pc);
+		return false;
-	if (it == peerConnectionMap.end())
+	userPointerMap.erase(pc);
-		return;
+	return true;
-
-	it->second->onDataChannel([pc, dataChannelCallback](std::shared_ptr<DataChannel> dataChannel) {
-		int dc = ++lastId;
-		dataChannelMap.emplace(std::make_pair(dc, dataChannel));
-		dataChannelCallback(dc, getUserPointer(pc));
-	});
 }
 
-void rtcSetLocalDescriptionCallback(int pc, void (*descriptionCallback)(const char *, const char *,
+bool eraseDataChannel(int dc) {
-                                                                        void *)) {
+	std::lock_guard lock(mutex);
-	auto it = peerConnectionMap.find(pc);
+	if (dataChannelMap.erase(dc) == 0)
-	if (it == peerConnectionMap.end())
+		return false;
-		return;
+	userPointerMap.erase(dc);
-
+	return true;
-	it->second->onLocalDescription([pc, descriptionCallback](const Description &description) {
-		descriptionCallback(string(description).c_str(), description.typeString().c_str(),
-		                    getUserPointer(pc));
-	});
 }
 
-void rtcSetLocalCandidateCallback(int pc,
+} // namespace
-                                  void (*candidateCallback)(const char *, const char *, void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
 
-	it->second->onLocalCandidate([pc, candidateCallback](const Candidate &candidate) {
+void rtcInitLogger(rtcLogLevel level) { InitLogger(static_cast<LogLevel>(level)); }
-		candidateCallback(candidate.candidate().c_str(), candidate.mid().c_str(),
-		                  getUserPointer(pc));
-	});
-}
-
-void rtcSetStateChangeCallback(int pc, void (*stateCallback)(rtc_state_t state, void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
-
-	it->second->onStateChange([pc, stateCallback](PeerConnection::State state) {
-		stateCallback(static_cast<rtc_state_t>(state), getUserPointer(pc));
-	});
-}
-
-void rtcSetGatheringStateChangeCallback(int pc,
-                                        void (*gatheringStateCallback)(rtc_gathering_state_t state,
-                                                                       void *)) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
-
-	it->second->onGatheringStateChange(
-	    [pc, gatheringStateCallback](PeerConnection::GatheringState state) {
-		    gatheringStateCallback(static_cast<rtc_gathering_state_t>(state), getUserPointer(pc));
-	    });
-}
-
-void rtcSetRemoteDescription(int pc, const char *sdp, const char *type) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
-
-	it->second->setRemoteDescription(Description(string(sdp), type ? string(type) : ""));
-}
-
-void rtcAddRemoteCandidate(int pc, const char *candidate, const char *mid) {
-	auto it = peerConnectionMap.find(pc);
-	if (it == peerConnectionMap.end())
-		return;
-
-	it->second->addRemoteCandidate(Candidate(string(candidate), mid ? string(mid) : ""));
-}
-
-int rtcGetDataChannelLabel(int dc, char *buffer, int size) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return 0;
-
-	if (!size)
-		return 0;
-
-	string label = it->second->label();
-	size = std::min(size_t(size - 1), label.size());
-	std::copy(label.data(), label.data() + size, buffer);
-	buffer[size] = '\0';
-	return size + 1;
-}
-
-void rtcSetOpenCallback(int dc, void (*openCallback)(void *)) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return;
-
-	it->second->onOpen([dc, openCallback]() { openCallback(getUserPointer(dc)); });
-}
-
-void rtcSetErrorCallback(int dc, void (*errorCallback)(const char *, void *)) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return;
-
-	it->second->onError([dc, errorCallback](const string &error) {
-		errorCallback(error.c_str(), getUserPointer(dc));
-	});
-}
-
-void rtcSetMessageCallback(int dc, void (*messageCallback)(const char *, int, void *)) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return;
-
-	it->second->onMessage(
-	    [dc, messageCallback](const binary &b) {
-		    messageCallback(reinterpret_cast<const char *>(b.data()), b.size(), getUserPointer(dc));
-	    },
-	    [dc, messageCallback](const string &s) {
-		    messageCallback(s.c_str(), -1, getUserPointer(dc));
-	    });
-}
-
-int rtcSendMessage(int dc, const char *data, int size) {
-	auto it = dataChannelMap.find(dc);
-	if (it == dataChannelMap.end())
-		return 0;
-
-	if (size >= 0) {
-		auto b = reinterpret_cast<const byte *>(data);
-		it->second->send(b, size);
-		return size;
-	} else {
-		string s(data);
-		it->second->send(s);
-		return s.size();
-	}
-}
 
 void rtcSetUserPointer(int i, void *ptr) {
 	if (ptr)
@@ -214,3 +105,340 @@ void rtcSetUserPointer(int i, void *ptr) {
 	else
 		userPointerMap.erase(i);
 }
+
+int rtcCreatePeerConnection(const rtcConfiguration *config) {
+	Configuration c;
+	for (int i = 0; i < config->iceServersCount; ++i)
+		c.iceServers.emplace_back(string(config->iceServers[i]));
+
+	return emplacePeerConnection(std::make_shared<PeerConnection>(c));
+}
+
+int rtcDeletePeerConnection(int pc) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	peerConnection->onDataChannel(nullptr);
+	peerConnection->onLocalDescription(nullptr);
+	peerConnection->onLocalCandidate(nullptr);
+	peerConnection->onStateChange(nullptr);
+	peerConnection->onGatheringStateChange(nullptr);
+
+	erasePeerConnection(pc);
+	return 0;
+}
+
+int rtcCreateDataChannel(int pc, const char *label) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	int dc = emplaceDataChannel(peerConnection->createDataChannel(string(label)));
+	void *ptr = getUserPointer(pc);
+	rtcSetUserPointer(dc, ptr);
+	return dc;
+}
+
+int rtcDeleteDataChannel(int dc) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	dataChannel->onOpen(nullptr);
+	dataChannel->onClosed(nullptr);
+	dataChannel->onError(nullptr);
+	dataChannel->onMessage(nullptr);
+	dataChannel->onBufferedAmountLow(nullptr);
+	dataChannel->onAvailable(nullptr);
+
+	eraseDataChannel(dc);
+	return 0;
+}
+
+int rtcSetDataChannelCallback(int pc, dataChannelCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	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);
+		});
+	else
+		peerConnection->onDataChannel(nullptr);
+	return 0;
+}
+
+int rtcSetLocalDescriptionCallback(int pc, descriptionCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	if (cb)
+		peerConnection->onLocalDescription([pc, cb](const Description &desc) {
+			cb(string(desc).c_str(), desc.typeString().c_str(), getUserPointer(pc));
+		});
+	else
+		peerConnection->onLocalDescription(nullptr);
+	return 0;
+}
+
+int rtcSetLocalCandidateCallback(int pc, candidateCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	if (cb)
+		peerConnection->onLocalCandidate([pc, cb](const Candidate &cand) {
+			cb(cand.candidate().c_str(), cand.mid().c_str(), getUserPointer(pc));
+		});
+	else
+		peerConnection->onLocalCandidate(nullptr);
+	return 0;
+}
+
+int rtcSetStateChangeCallback(int pc, stateChangeCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	if (cb)
+		peerConnection->onStateChange([pc, cb](PeerConnection::State state) {
+			cb(static_cast<rtcState>(state), getUserPointer(pc));
+		});
+	else
+		peerConnection->onStateChange(nullptr);
+	return 0;
+}
+
+int rtcSetGatheringStateChangeCallback(int pc, gatheringStateCallbackFunc cb) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	if (cb)
+		peerConnection->onGatheringStateChange([pc, cb](PeerConnection::GatheringState state) {
+			cb(static_cast<rtcGatheringState>(state), getUserPointer(pc));
+		});
+	else
+		peerConnection->onGatheringStateChange(nullptr);
+	return 0;
+}
+
+int rtcSetRemoteDescription(int pc, const char *sdp, const char *type) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	CATCH(peerConnection->setRemoteDescription({string(sdp), type ? string(type) : ""}));
+	return 0;
+}
+
+int rtcAddRemoteCandidate(int pc, const char *cand, const char *mid) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	CATCH(peerConnection->addRemoteCandidate({string(cand), mid ? string(mid) : ""}))
+	return 0;
+}
+
+int rtcGetLocalAddress(int pc, char *buffer, int size) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	if (auto addr = peerConnection->localAddress()) {
+		size = std::min(size_t(size - 1), addr->size());
+		std::copy(addr->data(), addr->data() + size, buffer);
+		buffer[size] = '\0';
+		return size + 1;
+	}
+	return -1;
+}
+
+int rtcGetRemoteAddress(int pc, char *buffer, int size) {
+	auto peerConnection = getPeerConnection(pc);
+	if (!peerConnection)
+		return -1;
+
+	if (auto addr = peerConnection->remoteAddress()) {
+		size = std::min(size_t(size - 1), addr->size());
+		std::copy(addr->data(), addr->data() + size, buffer);
+		buffer[size] = '\0';
+		return size + 1;
+	}
+	return -1;
+}
+
+int rtcGetDataChannelLabel(int dc, char *buffer, int size) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (!size)
+		return 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;
+}
+
+int rtcSetOpenCallback(int dc, openCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (cb)
+		dataChannel->onOpen([dc, cb]() { cb(getUserPointer(dc)); });
+	else
+		dataChannel->onOpen(nullptr);
+	return 0;
+}
+
+int rtcSetClosedCallback(int dc, closedCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (cb)
+		dataChannel->onClosed([dc, cb]() { cb(getUserPointer(dc)); });
+	else
+		dataChannel->onClosed(nullptr);
+	return 0;
+}
+
+int rtcSetErrorCallback(int dc, errorCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (cb)
+		dataChannel->onError(
+		    [dc, cb](const string &error) { cb(error.c_str(), getUserPointer(dc)); });
+	else
+		dataChannel->onError(nullptr);
+	return 0;
+}
+
+int rtcSetMessageCallback(int dc, messageCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (cb)
+		dataChannel->onMessage(
+		    [dc, cb](const binary &b) {
+			    cb(reinterpret_cast<const char *>(b.data()), b.size(), getUserPointer(dc));
+		    },
+		    [dc, cb](const string &s) { cb(s.c_str(), -1, getUserPointer(dc)); });
+	else
+		dataChannel->onMessage(nullptr);
+
+	return 0;
+}
+
+int rtcSendMessage(int dc, const char *data, int size) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (size >= 0) {
+		auto b = reinterpret_cast<const byte *>(data);
+		CATCH(dataChannel->send(b, size));
+		return size;
+	} else {
+		string s(data);
+		CATCH(dataChannel->send(s));
+		return s.size();
+	}
+}
+
+int rtcGetBufferedAmount(int dc) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	CATCH(return int(dataChannel->bufferedAmount()));
+}
+
+int rtcSetBufferedAmountLowThreshold(int dc, int amount) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	CATCH(dataChannel->setBufferedAmountLowThreshold(size_t(amount)));
+	return 0;
+}
+
+int rtcSetBufferedAmountLowCallback(int dc, bufferedAmountLowCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (cb)
+		dataChannel->onBufferedAmountLow([dc, cb]() { cb(getUserPointer(dc)); });
+	else
+		dataChannel->onBufferedAmountLow(nullptr);
+	return 0;
+}
+
+int rtcGetAvailableAmount(int dc) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	CATCH(return int(dataChannel->availableAmount()));
+}
+
+int rtcSetAvailableCallback(int dc, availableCallbackFunc cb) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (cb)
+		dataChannel->onOpen([dc, cb]() { cb(getUserPointer(dc)); });
+	else
+		dataChannel->onOpen(nullptr);
+	return 0;
+}
+
+int rtcReceiveMessage(int dc, char *buffer, int *size) {
+	auto dataChannel = getDataChannel(dc);
+	if (!dataChannel)
+		return -1;
+
+	if (!size)
+		return -1;
+
+	CATCH({
+		auto message = dataChannel->receive();
+		if (!message)
+			return 0;
+
+		return std::visit( //
+		    overloaded{    //
+		               [&](const binary &b) {
+			               *size = std::min(*size, int(b.size()));
+			               auto data = reinterpret_cast<const char *>(b.data());
+			               std::copy(data, data + *size, buffer);
+			               return *size;
+		               },
+		               [&](const string &s) {
+			               int len = std::min(*size - 1, int(s.size()));
+			               if (len >= 0) {
+				               std::copy(s.data(), s.data() + len, buffer);
+				               buffer[len] = '\0';
+			               }
+			               *size = -(len + 1);
+			               return len + 1;
+		               }},
+		    *message);
+	});
+}

src/sctptransport.cpp

@@ -23,10 +23,6 @@
 #include <iostream>
 #include <vector>
 
-#ifdef __linux__
-#include <arpa/inet.h>
-#endif
-
 #ifdef USE_JUICE
 #ifndef __APPLE__
 // libjuice enables Linux path MTU discovery or sets the DF flag
@@ -53,31 +49,20 @@ using std::shared_ptr;
 
 namespace rtc {
 
-std::mutex SctpTransport::GlobalMutex;
+void SctpTransport::Init() {
-int SctpTransport::InstancesCount = 0;
+	usrsctp_init(0, &SctpTransport::WriteCallback, nullptr);
-
+	usrsctp_sysctl_set_sctp_ecn_enable(0);
-void SctpTransport::GlobalInit() {
+	usrsctp_sysctl_set_sctp_init_rtx_max_default(5);
-	std::lock_guard lock(GlobalMutex);
+	usrsctp_sysctl_set_sctp_path_rtx_max_default(5);
-	if (InstancesCount++ == 0) {
+	usrsctp_sysctl_set_sctp_assoc_rtx_max_default(5);              // single path
-		usrsctp_init(0, &SctpTransport::WriteCallback, nullptr);
+	usrsctp_sysctl_set_sctp_rto_min_default(1 * 1000);             // ms
-		usrsctp_sysctl_set_sctp_ecn_enable(0);
+	usrsctp_sysctl_set_sctp_rto_max_default(10 * 1000);            // ms
-		usrsctp_sysctl_set_sctp_init_rtx_max_default(5);
+	usrsctp_sysctl_set_sctp_rto_initial_default(1 * 1000);         // ms
-		usrsctp_sysctl_set_sctp_path_rtx_max_default(5);
+	usrsctp_sysctl_set_sctp_init_rto_max_default(10 * 1000);       // ms
-		usrsctp_sysctl_set_sctp_assoc_rtx_max_default(5);              // single path
+	usrsctp_sysctl_set_sctp_heartbeat_interval_default(10 * 1000); // ms
-		usrsctp_sysctl_set_sctp_rto_min_default(1 * 1000);             // ms
-		usrsctp_sysctl_set_sctp_rto_max_default(10 * 1000);            // ms
-		usrsctp_sysctl_set_sctp_rto_initial_default(1 * 1000);         // ms
-		usrsctp_sysctl_set_sctp_init_rto_max_default(10 * 1000);       // ms
-		usrsctp_sysctl_set_sctp_heartbeat_interval_default(10 * 1000); // ms
-	}
 }
 
-void SctpTransport::GlobalCleanup() {
+void SctpTransport::Cleanup() { usrsctp_finish(); }
-	std::lock_guard lock(GlobalMutex);
-	if (--InstancesCount == 0) {
-		usrsctp_finish();
-	}
-}
 
 SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
                              message_callback recvCallback, amount_callback bufferedAmountCallback,
@@ -88,7 +73,6 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 	onRecv(recvCallback);
 
 	PLOG_DEBUG << "Initializing SCTP transport";
-	GlobalInit();
 
 	usrsctp_register_address(this);
 	mSock = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP, &SctpTransport::RecvCallback,
@@ -179,8 +163,6 @@ SctpTransport::~SctpTransport() {
 
 	usrsctp_close(mSock);
 	usrsctp_deregister_address(this);
-
-	GlobalCleanup();
 }
 
 SctpTransport::State SctpTransport::state() const { return mState; }
@@ -191,7 +173,7 @@ void SctpTransport::stop() {
 
 	if (!mShutdown.exchange(true)) {
 		mSendQueue.stop();
-		flush();
+		safeFlush();
 		shutdown();
 	}
 }
@@ -281,13 +263,15 @@ void SctpTransport::incoming(message_ptr message) {
 		mWrittenCondition.wait(lock, [&]() { return mWrittenOnce || mState != State::Connected; });
 	}
 
-	if (message) {
+	if (!message) {
-		usrsctp_conninput(this, message->data(), message->size(), 0);
-	} else {
 		PLOG_INFO << "SCTP disconnected";
 		changeState(State::Disconnected);
 		recv(nullptr);
+		return;
 	}
+
+	PLOG_VERBOSE << "Incoming size=" << message->size();
+	usrsctp_conninput(this, message->data(), message->size(), 0);
 }
 
 void SctpTransport::changeState(State state) {
@@ -385,18 +369,33 @@ bool SctpTransport::trySendMessage(message_ptr message) {
 void SctpTransport::updateBufferedAmount(uint16_t streamId, long delta) {
 	// Requires mSendMutex to be locked
 	auto it = mBufferedAmount.insert(std::make_pair(streamId, 0)).first;
-	size_t amount = it->second;
+	size_t amount = size_t(std::max(long(it->second) + delta, long(0)));
-	amount = size_t(std::max(long(amount) + delta, long(0)));
 	if (amount == 0)
 		mBufferedAmount.erase(it);
+	else
+		it->second = amount;
+
 	mBufferedAmountCallback(streamId, amount);
 }
 
+bool SctpTransport::safeFlush() {
+	try {
+		flush();
+		return true;
+
+	} catch (const std::exception &e) {
+		PLOG_ERROR << "SCTP flush: " << e.what();
+		return false;
+	}
+}
+
 int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, const byte *data,
                               size_t len, struct sctp_rcvinfo info, int flags) {
 	try {
+		PLOG_VERBOSE << "Handle recv, len=" << len;
 		if (!len)
 			return -1;
+
 		if (flags & MSG_EOR) {
 			if (!mPartialRecv.empty()) {
 				mPartialRecv.insert(mPartialRecv.end(), data, data + len);
@@ -422,24 +421,21 @@ int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, co
 }
 
 int SctpTransport::handleSend(size_t free) {
-	try {
+	PLOG_VERBOSE << "Handle send, free=" << free;
-		std::lock_guard lock(mSendMutex);
+	return safeFlush() ? 0 : -1;
-		trySendQueue();
-	} catch (const std::exception &e) {
-		PLOG_ERROR << "SCTP send: " << e.what();
-		return -1;
-	}
-	return 0; // success
 }
 
 int SctpTransport::handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_df) {
 	try {
+		PLOG_VERBOSE << "Handle write, len=" << len;
+
 		std::unique_lock lock(mWriteMutex);
 		if (!outgoing(make_message(data, data + len)))
 			return -1;
 		mWritten = true;
 		mWrittenOnce = true;
 		mWrittenCondition.notify_all();
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "SCTP write: " << e.what();
 		return -1;
@@ -448,6 +444,8 @@ int SctpTransport::handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_
 }
 
 void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, PayloadId ppid) {
+	PLOG_VERBOSE << "Process data, len=" << len;
+
 	// The usage of the PPIDs "WebRTC String Partial" and "WebRTC Binary Partial" is deprecated.
 	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13#section-6.6
 	// We handle them at reception for compatibility reasons but should never send them.
@@ -508,10 +506,15 @@ void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, Payl
 }
 
 void SctpTransport::processNotification(const union sctp_notification *notify, size_t len) {
-	if (len != size_t(notify->sn_header.sn_length))
+	if (len != size_t(notify->sn_header.sn_length)) {
+		PLOG_WARNING << "Invalid notification length";
 		return;
+	}
 
-	switch (notify->sn_header.sn_type) {
+	auto type = notify->sn_header.sn_type;
+	PLOG_VERBOSE << "Process notification, type=" << type;
+
+	switch (type) {
 	case SCTP_ASSOC_CHANGE: {
 		const struct sctp_assoc_change &assoc_change = notify->sn_assoc_change;
 		if (assoc_change.sac_state == SCTP_COMM_UP) {
@@ -527,13 +530,16 @@ void SctpTransport::processNotification(const union sctp_notification *notify, s
 			}
 			mWrittenCondition.notify_all();
 		}
+		break;
 	}
+
 	case SCTP_SENDER_DRY_EVENT: {
 		// It not should be necessary since the send callback should have been called already,
 		// but to be sure, let's try to send now.
-		std::lock_guard lock(mSendMutex);
+		safeFlush();
-		trySendQueue();
+		break;
 	}
+
 	case SCTP_STREAM_RESET_EVENT: {
 		const struct sctp_stream_reset_event &reset_event = notify->sn_strreset_event;
 		const int count = (reset_event.strreset_length - sizeof(reset_event)) / sizeof(uint16_t);

src/sctptransport.hpp

@@ -29,20 +29,15 @@
 #include <map>
 #include <mutex>
 
-#ifdef _WIN32
-#include <winsock2.h>
-#elif __linux__
-#include <sys/socket.h>
-#endif
-
-#include <sys/types.h>
-
 #include "usrsctp.h"
 
 namespace rtc {
 
 class SctpTransport : public Transport {
 public:
+	static void Init();
+	static void Cleanup();
+
 	enum class State { Disconnected, Connecting, Connected, Failed };
 
 	using amount_callback = std::function<void(uint16_t streamId, size_t amount)>;
@@ -80,6 +75,7 @@ private:
 	bool trySendQueue();
 	bool trySendMessage(message_ptr message);
 	void updateBufferedAmount(uint16_t streamId, long delta);
+	bool safeFlush();
 
 	int handleRecv(struct socket *sock, union sctp_sockstore addr, const byte *data, size_t len,
 	               struct sctp_rcvinfo recv_info, int flags);
@@ -113,12 +109,6 @@ 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);
-
-	void GlobalInit();
-	void GlobalCleanup();
-
-	static std::mutex GlobalMutex;
-	static int InstancesCount;
 };
 
 } // namespace rtc

test/capi.cpp

@@ -0,0 +1,189 @@
+/**
+ * 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 <rtc/rtc.h>
+
+#include <cstdbool>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+
+#include <unistd.h> // for sleep
+
+using namespace std;
+
+typedef struct {
+	rtcState state;
+	rtcGatheringState gatheringState;
+	int pc;
+	int dc;
+	bool connected;
+} Peer;
+
+Peer *peer1 = NULL;
+Peer *peer2 = NULL;
+
+static void descriptionCallback(const char *sdp, const char *type, void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	printf("Description %d:\n%s\n", peer == peer1 ? 1 : 2, sdp);
+	Peer *other = peer == peer1 ? peer2 : peer1;
+	rtcSetRemoteDescription(other->pc, sdp, type);
+}
+
+static void candidateCallback(const char *cand, const char *mid, void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	printf("Candidate %d: %s\n", peer == peer1 ? 1 : 2, cand);
+	Peer *other = peer == peer1 ? peer2 : peer1;
+	rtcAddRemoteCandidate(other->pc, cand, mid);
+}
+
+static void stateChangeCallback(rtcState state, void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	peer->state = state;
+	printf("State %d: %d\n", peer == peer1 ? 1 : 2, (int)state);
+}
+
+static void gatheringStateCallback(rtcGatheringState state, void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	peer->gatheringState = state;
+	printf("Gathering state %d: %d\n", peer == peer1 ? 1 : 2, (int)state);
+}
+
+static void openCallback(void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	peer->connected = true;
+	printf("DataChannel %d: Open\n", peer == peer1 ? 1 : 2);
+
+	const char *message = peer == peer1 ? "Hello from 1" : "Hello from 2";
+	rtcSendMessage(peer->dc, message, -1); // negative size indicates a null-terminated string
+}
+
+static void closedCallback(void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	peer->connected = false;
+}
+
+static void messageCallback(const char *message, int size, void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	if (size < 0) { // negative size indicates a null-terminated string
+		printf("Message %d: %s\n", peer == peer1 ? 1 : 2, message);
+	} else {
+		printf("Message %d: [binary of size %d]\n", peer == peer1 ? 1 : 2, size);
+	}
+}
+
+static void dataChannelCallback(int dc, void *ptr) {
+	Peer *peer = (Peer *)ptr;
+	peer->dc = dc;
+	peer->connected = true;
+	rtcSetClosedCallback(dc, closedCallback);
+	rtcSetMessageCallback(dc, messageCallback);
+
+	char buffer[256];
+	if (rtcGetDataChannelLabel(dc, buffer, 256) >= 0)
+		printf("DataChannel %d: Received with label \"%s\"\n", peer == peer1 ? 1 : 2, buffer);
+
+	const char *message = peer == peer1 ? "Hello from 1" : "Hello from 2";
+	rtcSendMessage(peer->dc, message, -1); // negative size indicates a null-terminated string
+}
+
+static Peer *createPeer(const rtcConfiguration *config) {
+	Peer *peer = (Peer *)malloc(sizeof(Peer));
+	if (!peer)
+		return nullptr;
+	memset(peer, 0, sizeof(Peer));
+
+	// Create peer connection
+	peer->pc = rtcCreatePeerConnection(config);
+	rtcSetUserPointer(peer->pc, peer);
+	rtcSetDataChannelCallback(peer->pc, dataChannelCallback);
+	rtcSetLocalDescriptionCallback(peer->pc, descriptionCallback);
+	rtcSetLocalCandidateCallback(peer->pc, candidateCallback);
+	rtcSetStateChangeCallback(peer->pc, stateChangeCallback);
+	rtcSetGatheringStateChangeCallback(peer->pc, gatheringStateCallback);
+
+	return peer;
+}
+
+static void deletePeer(Peer *peer) {
+	if (peer) {
+		if (peer->dc)
+			rtcDeleteDataChannel(peer->dc);
+		if (peer->pc)
+			rtcDeletePeerConnection(peer->pc);
+		free(peer);
+	}
+}
+
+int test_capi_main() {
+	rtcInitLogger(RTC_LOG_DEBUG);
+
+	rtcConfiguration config;
+	memset(&config, 0, sizeof(config));
+	// const char *iceServers[1] = {"stun:stun.l.google.com:19302"};
+	// config.iceServers = iceServers;
+	// config.iceServersCount = 1;
+
+	// Create peer 1
+	peer1 = createPeer(&config);
+	if (!peer1)
+		goto error;
+
+	// Create peer 2
+	peer2 = createPeer(&config);
+	if (!peer2)
+		goto error;
+
+	// Peer 1: Create data channel
+	peer1->dc = rtcCreateDataChannel(peer1->pc, "test");
+	rtcSetOpenCallback(peer1->dc, openCallback);
+	rtcSetClosedCallback(peer1->dc, closedCallback);
+	rtcSetMessageCallback(peer1->dc, messageCallback);
+
+	sleep(3);
+
+	char buffer[256];
+	if (rtcGetLocalAddress(peer1->pc, buffer, 256) >= 0)
+		printf("Local address 1:  %s\n", buffer);
+	if (rtcGetRemoteAddress(peer1->pc, buffer, 256) >= 0)
+		printf("Remote address 1: %s\n", buffer);
+	if (rtcGetLocalAddress(peer2->pc, buffer, 256) >= 0)
+		printf("Local address 2:  %s\n", buffer);
+	if (rtcGetRemoteAddress(peer2->pc, buffer, 256) >= 0)
+		printf("Remote address 2: %s\n", buffer);
+
+	if (peer1->connected && peer2->connected) {
+		deletePeer(peer1);
+		deletePeer(peer2);
+		sleep(1);
+		printf("Success\n");
+		return 0;
+	}
+
+error:
+	deletePeer(peer1);
+	deletePeer(peer2);
+	return -1;
+}
+
+#include <stdexcept>
+
+void test_capi() {
+	if (test_capi_main())
+		throw std::runtime_error("Connection failed");
+}

test/connectivity.cpp

@@ -0,0 +1,129 @@
+/**
+ * Copyright (c) 2019 Paul-Louis Ageneau
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include "rtc/rtc.hpp"
+
+#include <chrono>
+#include <iostream>
+#include <memory>
+#include <thread>
+
+using namespace rtc;
+using namespace std;
+
+template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
+
+void test_connectivity() {
+	InitLogger(LogLevel::Debug);
+
+	Configuration config;
+	// config.iceServers.emplace_back("stun:stun.l.google.com:19302");
+
+	auto pc1 = std::make_shared<PeerConnection>(config);
+
+	auto pc2 = std::make_shared<PeerConnection>(config);
+
+	pc1->onLocalDescription([wpc2 = make_weak_ptr(pc2)](const Description &sdp) {
+		auto pc2 = wpc2.lock();
+		if (!pc2)
+			return;
+		cout << "Description 1: " << sdp << endl;
+		pc2->setRemoteDescription(sdp);
+	});
+
+	pc1->onLocalCandidate([wpc2 = make_weak_ptr(pc2)](const Candidate &candidate) {
+		auto pc2 = wpc2.lock();
+		if (!pc2)
+			return;
+		cout << "Candidate 1: " << candidate << endl;
+		pc2->addRemoteCandidate(candidate);
+	});
+
+	pc1->onStateChange([](PeerConnection::State state) { cout << "State 1: " << state << endl; });
+	pc1->onGatheringStateChange([](PeerConnection::GatheringState state) {
+		cout << "Gathering state 1: " << state << endl;
+	});
+
+	pc2->onLocalDescription([wpc1 = make_weak_ptr(pc1)](const Description &sdp) {
+		auto pc1 = wpc1.lock();
+		if (!pc1)
+			return;
+		cout << "Description 2: " << sdp << endl;
+		pc1->setRemoteDescription(sdp);
+	});
+
+	pc2->onLocalCandidate([wpc1 = make_weak_ptr(pc1)](const Candidate &candidate) {
+		auto pc1 = wpc1.lock();
+		if (!pc1)
+			return;
+		cout << "Candidate 2: " << candidate << endl;
+		pc1->addRemoteCandidate(candidate);
+	});
+
+	pc2->onStateChange([](PeerConnection::State state) { cout << "State 2: " << state << endl; });
+	pc2->onGatheringStateChange([](PeerConnection::GatheringState state) {
+		cout << "Gathering state 2: " << state << endl;
+	});
+
+	shared_ptr<DataChannel> dc2;
+	pc2->onDataChannel([&dc2](shared_ptr<DataChannel> dc) {
+		cout << "DataChannel 2: Received with label \"" << dc->label() << "\"" << endl;
+		dc2 = dc;
+		dc2->onMessage([](const variant<binary, string> &message) {
+			if (holds_alternative<string>(message)) {
+				cout << "Message 2: " << get<string>(message) << endl;
+			}
+		});
+		dc2->send("Hello from 2");
+	});
+
+	auto dc1 = pc1->createDataChannel("test");
+	dc1->onOpen([wdc1 = make_weak_ptr(dc1)]() {
+		auto dc1 = wdc1.lock();
+		if (!dc1)
+			return;
+		cout << "DataChannel 1: Open" << endl;
+		dc1->send("Hello from 1");
+	});
+	dc1->onMessage([](const variant<binary, string> &message) {
+		if (holds_alternative<string>(message)) {
+			cout << "Message 1: " << get<string>(message) << endl;
+		}
+	});
+
+	this_thread::sleep_for(3s);
+
+	if (auto addr = pc1->localAddress())
+		cout << "Local address 1:  " << *addr << endl;
+	if (auto addr = pc1->remoteAddress())
+		cout << "Remote address 1: " << *addr << endl;
+	if (auto addr = pc2->localAddress())
+		cout << "Local address 2:  " << *addr << endl;
+	if (auto addr = pc2->remoteAddress())
+		cout << "Remote address 2: " << *addr << endl;
+
+	if (!dc1->isOpen() || !dc2->isOpen())
+		throw runtime_error("DataChannel is not open");
+
+	pc1->close();
+	pc2->close();
+
+	this_thread::sleep_for(1s);
+
+	cout << "Success" << endl;
+}

test/main.cpp

@@ -16,131 +16,29 @@
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  */
 
-#include "rtc/rtc.hpp"
-
-#include <chrono>
 #include <iostream>
-#include <memory>
-#include <thread>
 
-#ifdef _WIN32
-#include <winsock2.h>
-#endif
-
-using namespace rtc;
 using namespace std;
 
-template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
+void test_connectivity();
+void test_capi();
 
 int main(int argc, char **argv) {
-	InitLogger(LogLevel::Warning);
+	try {
-
+		std::cout << "*** Running connectivity test..." << std::endl;
-#ifdef _WIN32
+		test_connectivity();
-	WSADATA wsaData;
+		std::cout << "*** Finished connectivity test" << std::endl;
-	if (WSAStartup(MAKEWORD(2, 2), &wsaData))
+	} catch (const exception &e) {
-		throw std::runtime_error("WSAStartup failed, error=" + std::to_string(WSAGetLastError()));
+		std::cerr << "Connectivity test failed: " << e.what() << endl;
-#endif
+		return -1;
-
-	Configuration config;
-	// config.iceServers.emplace_back("stun:stun.l.google.com:19302");
-	// config.iceServers.emplace_back(IceServer("TURN_SERVER_URL", "PORT", "USERNAME", "PASSWORD",
-	//                                         IceServer::RelayType::TurnUdp)); // libnice only
-	// config.enableIceTcp = true; // libnice only
-
-	auto pc1 = std::make_shared<PeerConnection>(config);
-	auto pc2 = std::make_shared<PeerConnection>(config);
-
-	pc1->onLocalDescription([wpc2 = make_weak_ptr(pc2)](const Description &sdp) {
-		auto pc2 = wpc2.lock();
-		if (!pc2)
-			return;
-		cout << "Description 1: " << sdp << endl;
-		pc2->setRemoteDescription(sdp);
-	});
-
-	pc1->onLocalCandidate([wpc2 = make_weak_ptr(pc2)](const Candidate &candidate) {
-		auto pc2 = wpc2.lock();
-		if (!pc2)
-			return;
-		cout << "Candidate 1: " << candidate << endl;
-		pc2->addRemoteCandidate(candidate);
-	});
-
-	pc1->onStateChange([](PeerConnection::State state) { cout << "State 1: " << state << endl; });
-	pc1->onGatheringStateChange([](PeerConnection::GatheringState state) {
-		cout << "Gathering state 1: " << state << endl;
-	});
-
-	pc2->onLocalDescription([wpc1 = make_weak_ptr(pc1)](const Description &sdp) {
-		auto pc1 = wpc1.lock();
-		if (!pc1)
-			return;
-		cout << "Description 2: " << sdp << endl;
-		pc1->setRemoteDescription(sdp);
-	});
-
-	pc2->onLocalCandidate([wpc1 = make_weak_ptr(pc1)](const Candidate &candidate) {
-		auto pc1 = wpc1.lock();
-		if (!pc1)
-			return;
-		cout << "Candidate 2: " << candidate << endl;
-		pc1->addRemoteCandidate(candidate);
-	});
-
-	pc2->onStateChange([](PeerConnection::State state) { cout << "State 2: " << state << endl; });
-	pc2->onGatheringStateChange([](PeerConnection::GatheringState state) {
-		cout << "Gathering state 2: " << state << endl;
-	});
-
-	shared_ptr<DataChannel> dc2;
-	pc2->onDataChannel([&dc2](shared_ptr<DataChannel> dc) {
-		cout << "Got a DataChannel with label: " << dc->label() << endl;
-		dc2 = dc;
-		dc2->onMessage([](const variant<binary, string> &message) {
-			if (holds_alternative<string>(message)) {
-				cout << "Received 2: " << get<string>(message) << endl;
-			}
-		});
-		dc2->send("Hello from 2");
-	});
-
-	auto dc1 = pc1->createDataChannel("test");
-	dc1->onOpen([wdc1 = make_weak_ptr(dc1)]() {
-		auto dc1 = wdc1.lock();
-		if (!dc1)
-			return;
-		cout << "DataChannel open: " << dc1->label() << endl;
-		dc1->send("Hello from 1");
-	});
-	dc1->onMessage([](const variant<binary, string> &message) {
-		if (holds_alternative<string>(message)) {
-			cout << "Received 1: " << get<string>(message) << endl;
-		}
-	});
-
-	this_thread::sleep_for(3s);
-
-	if (auto addr = pc1->localAddress())
-		cout << "Local address 1:  " << *addr << endl;
-	if (auto addr = pc1->remoteAddress())
-		cout << "Remote address 1: " << *addr << endl;
-	if (auto addr = pc2->localAddress())
-		cout << "Local address 2:  " << *addr << endl;
-	if (auto addr = pc2->remoteAddress())
-		cout << "Remote address 2: " << *addr << endl;
-
-	bool success;
-	if ((success = dc1->isOpen() && dc2->isOpen())) {
-		pc1->close();
-		pc2->close();
-		cout << "Success" << endl;
-	} else {
-		cout << "Failure" << endl;
 	}
-
+	try {
-#ifdef _WIN32
+		std::cout << "*** Running C API test..." << std::endl;
-		WSACleanup();
+		test_capi();
-#endif
+		std::cout << "*** Finished C API test" << std::endl;
-
+	} catch (const exception &e) {
-	    return success ? 0 : 1;
+		std::cerr << "C API test failed: " << e.what() << endl;
+		return -1;
+	}
+	return 0;
 }

test/p2p/answerer.cpp

@@ -23,23 +23,13 @@
 #include <memory>
 #include <thread>
 
-#ifdef _WIN32
-#include <winsock2.h>
-#endif
-
 using namespace rtc;
 using namespace std;
 
 template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
 
 int main(int argc, char **argv) {
-	InitLogger(LogLevel::Debug);
+	InitLogger(LogLevel::Warning);
-
-#ifdef _WIN32
-	WSADATA wsaData;
-	if (WSAStartup(MAKEWORD(2, 2), &wsaData))
-		throw std::runtime_error("WSAStartup failed, error=" + std::to_string(WSAGetLastError()));
-#endif
 
 	Configuration config;
 	// config.iceServers.emplace_back("stun.l.google.com:19302");
@@ -141,8 +131,4 @@ int main(int argc, char **argv) {
 		dc->close();
 	if (pc)
 		pc->close();
-
-#ifdef _WIN32
-	WSACleanup();
-#endif
 }

test/p2p/offerer.cpp

@@ -23,10 +23,6 @@
 #include <memory>
 #include <thread>
 
-#ifdef _WIN32
-#include <winsock2.h>
-#endif
-
 using namespace rtc;
 using namespace std;
 
@@ -35,12 +31,6 @@ template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
 int main(int argc, char **argv) {
 	InitLogger(LogLevel::Warning);
 
-#ifdef _WIN32
-	WSADATA wsaData;
-	if (WSAStartup(MAKEWORD(2, 2), &wsaData))
-		throw std::runtime_error("WSAStartup failed, error=" + std::to_string(WSAGetLastError()));
-#endif
-
 	Configuration config;
 	// config.iceServers.emplace_back("stun.l.google.com:19302");
 
@@ -141,8 +131,4 @@ int main(int argc, char **argv) {
 		dc->close();
 	if (pc)
 		pc->close();
-
-#ifdef _WIN32
-	WSACleanup();
-#endif
 }