Bumped version to 0.9.4

Improve consistency of C API

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required(VERSION 3.7)
 project(libdatachannel
 	DESCRIPTION "WebRTC Data Channels Library"
-	VERSION 0.9.2
+	VERSION 0.9.4
 	LANGUAGES CXX)
 
 # Options
@@ -11,6 +11,7 @@ option(NO_WEBSOCKET "Disable WebSocket support" OFF)
 option(NO_EXAMPLES "Disable examples" OFF)
 option(NO_TESTS "Disable tests build" OFF)
 option(WARNINGS_AS_ERRORS "Treat warnings as errors" OFF)
+option(CAPI_STDCALL "Set calling convention of C API callbacks stdcall" OFF)
 
 if(USE_NICE)
 	option(USE_JUICE "Use libjuice" OFF)
@@ -217,6 +218,11 @@ else()
 	target_link_libraries(datachannel-static PRIVATE LibJuice::LibJuiceStatic)
 endif()
 
+if(CAPI_STDCALL)
+	target_compile_definitions(datachannel PUBLIC CAPI_STDCALL)
+	target_compile_definitions(datachannel-static PUBLIC CAPI_STDCALL)
+endif()
+
 add_library(LibDataChannel::LibDataChannel ALIAS datachannel)
 add_library(LibDataChannel::LibDataChannelStatic ALIAS datachannel-static)
 

README.md

@@ -202,5 +202,6 @@ ws->open("wss://my.websocket/service");
 ## External resources
 - Rust wrapper for libdatachannel: [datachannel-rs](https://github.com/lerouxrgd/datachannel-rs)
 - Node.js wrapper for libdatachannel: [node-datachannel](https://github.com/murat-dogan/node-datachannel)
+- Unity wrapper for Windows 10 and Hololens: [datachannel-unity](https://github.com/hanseuljun/datachannel-unity)
 - WebAssembly wrapper compatible with libdatachannel: [datachannel-wasm](https://github.com/paullouisageneau/datachannel-wasm)
 

include/rtc/queue.hpp

@@ -38,6 +38,7 @@ public:
 	~Queue();
 
 	void stop();
+	bool running() const;
 	bool empty() const;
 	bool full() const;
 	size_t size() const;   // elements
@@ -80,6 +81,11 @@ template <typename T> void Queue<T>::stop() {
 	mPushCondition.notify_all();
 }
 
+template <typename T> bool Queue<T>::running() const {
+	std::lock_guard lock(mMutex);
+	return !mQueue.empty() || !mStopping;
+}
+
 template <typename T> bool Queue<T>::empty() const {
 	std::lock_guard lock(mMutex);
 	return mQueue.empty();

include/rtc/rtc.h

@@ -29,6 +29,12 @@ extern "C" {
 #define RTC_EXPORT
 #endif
 
+#ifdef CAPI_STDCALL
+#define RTC_API __stdcall
+#else
+#define RTC_API
+#endif
+
 #ifndef RTC_ENABLE_WEBSOCKET
 #define RTC_ENABLE_WEBSOCKET 1
 #endif
@@ -81,22 +87,23 @@ typedef struct {
 	unsigned int maxRetransmits;    // ignored if reliable
 } rtcReliability;
 
-typedef void (*rtcLogCallbackFunc)(rtcLogLevel level, const char *message);
+typedef void (RTC_API *rtcLogCallbackFunc)(rtcLogLevel level, const char *message);
-typedef void (*rtcDescriptionCallbackFunc)(const char *sdp, const char *type, void *ptr);
+typedef void (RTC_API *rtcDescriptionCallbackFunc)(const char *sdp, const char *type, void *ptr);
-typedef void (*rtcCandidateCallbackFunc)(const char *cand, const char *mid, void *ptr);
+typedef void (RTC_API *rtcCandidateCallbackFunc)(const char *cand, const char *mid, void *ptr);
-typedef void (*rtcStateChangeCallbackFunc)(rtcState state, void *ptr);
+typedef void (RTC_API *rtcStateChangeCallbackFunc)(rtcState state, void *ptr);
-typedef void (*rtcGatheringStateCallbackFunc)(rtcGatheringState state, void *ptr);
+typedef void (RTC_API *rtcGatheringStateCallbackFunc)(rtcGatheringState state, void *ptr);
-typedef void (*rtcDataChannelCallbackFunc)(int dc, void *ptr);
+typedef void (RTC_API *rtcDataChannelCallbackFunc)(int dc, void *ptr);
-typedef void (*rtcTrackCallbackFunc)(int tr, void *ptr);
+typedef void (RTC_API *rtcTrackCallbackFunc)(int tr, void *ptr);
-typedef void (*rtcOpenCallbackFunc)(void *ptr);
+typedef void (RTC_API *rtcOpenCallbackFunc)(void *ptr);
-typedef void (*rtcClosedCallbackFunc)(void *ptr);
+typedef void (RTC_API *rtcClosedCallbackFunc)(void *ptr);
-typedef void (*rtcErrorCallbackFunc)(const char *error, void *ptr);
+typedef void (RTC_API *rtcErrorCallbackFunc)(const char *error, void *ptr);
-typedef void (*rtcMessageCallbackFunc)(const char *message, int size, void *ptr);
+typedef void (RTC_API *rtcMessageCallbackFunc)(const char *message, int size, void *ptr);
-typedef void (*rtcBufferedAmountLowCallbackFunc)(void *ptr);
+typedef void (RTC_API *rtcBufferedAmountLowCallbackFunc)(void *ptr);
-typedef void (*rtcAvailableCallbackFunc)(void *ptr);
+typedef void (RTC_API *rtcAvailableCallbackFunc)(void *ptr);
 
 // Log
-RTC_EXPORT void rtcInitLogger(rtcLogLevel level, rtcLogCallbackFunc cb); // NULL cb to log to stdout
+// NULL cb on the first call will log to stdout
+RTC_EXPORT void rtcInitLogger(rtcLogLevel level, rtcLogCallbackFunc cb);
 
 // User pointer
 RTC_EXPORT void rtcSetUserPointer(int id, void *ptr);
@@ -114,6 +121,9 @@ RTC_EXPORT int rtcSetLocalDescription(int pc);
 RTC_EXPORT int rtcSetRemoteDescription(int pc, const char *sdp, const char *type);
 RTC_EXPORT int rtcAddRemoteCandidate(int pc, const char *cand, const char *mid);
 
+RTC_EXPORT int rtcGetLocalDescription(int pc, char *buffer, int size);
+RTC_EXPORT int rtcGetRemoteDescription(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);
 

src/capi.cpp

@@ -195,12 +195,17 @@ template <typename F> int wrap(F func) {
 		return RTC_ERR_SUCCESS;                                                                    \
 	})
 
-class plog_appender : public plog::IAppender {
+class plogAppender : public plog::IAppender {
 public:
-	plog_appender(rtcLogCallbackFunc cb = nullptr) { set_callback(cb); }
+	plogAppender(rtcLogCallbackFunc cb = nullptr) { setCallback(cb); }
 
-	void set_callback(rtcLogCallbackFunc cb) {
+	plogAppender(plogAppender &&appender) : callback(nullptr) {
-		std::lock_guard lock(mutex);
+		std::lock_guard lock(appender.callbackMutex);
+		std::swap(appender.callback, callback);
+	}
+
+	void setCallback(rtcLogCallbackFunc cb) {
+		std::lock_guard lock(callbackMutex);
 		callback = cb;
 	}
 
@@ -215,7 +220,7 @@ public:
 #else
 		std::string str = formatted;
 #endif
-		std::lock_guard lock(mutex);
+		std::lock_guard lock(callbackMutex);
 		if (callback)
 			callback(static_cast<rtcLogLevel>(record.getSeverity()), str.c_str());
 		else
@@ -224,18 +229,24 @@ public:
 
 private:
 	rtcLogCallbackFunc callback;
+	std::mutex callbackMutex;
 };
 
 } // namespace
 
 void rtcInitLogger(rtcLogLevel level, rtcLogCallbackFunc cb) {
-	static std::optional<plog_appender> appender;
+	static std::optional<plogAppender> appender;
-	if (appender)
+	const auto severity = static_cast<plog::Severity>(level);
-		appender->set_callback(cb);
+	std::lock_guard lock(mutex);
-	else if (cb)
+	if (appender) {
-		appender.emplace(plog_appender(cb));
+		appender->setCallback(cb);
-
+		InitLogger(severity, nullptr); // change the severity
-	InitLogger(static_cast<plog::Severity>(level), appender ? &appender.value() : nullptr);
+	} else if (cb) {
+		appender.emplace(plogAppender(cb));
+		InitLogger(severity, &appender.value());
+	} else {
+		InitLogger(severity, nullptr); // log to stdout
+	}
 }
 
 void rtcSetUserPointer(int i, void *ptr) { setUserPointer(i, ptr); }
@@ -518,6 +529,52 @@ int rtcAddRemoteCandidate(int pc, const char *cand, const char *mid) {
 	});
 }
 
+int rtcGetLocalDescription(int pc, char *buffer, int size) {
+	return WRAP({
+		auto peerConnection = getPeerConnection(pc);
+
+		if (size <= 0)
+			return 0;
+
+		if (!buffer)
+			throw std::invalid_argument("Unexpected null pointer for buffer");
+
+		if (auto desc = peerConnection->localDescription()) {
+			auto sdp = string(*desc);
+			const char *data = sdp.data();
+			size = std::min(size - 1, int(sdp.size()));
+			std::copy(data, data + size, buffer);
+			buffer[size] = '\0';
+			return size + 1;
+		}
+
+		return RTC_ERR_FAILURE;
+	});
+}
+
+int rtcGetRemoteDescription(int pc, char *buffer, int size) {
+	return WRAP({
+		auto peerConnection = getPeerConnection(pc);
+
+		if (size <= 0)
+			return 0;
+
+		if (!buffer)
+			throw std::invalid_argument("Unexpected null pointer for buffer");
+
+		if (auto desc = peerConnection->remoteDescription()) {
+			auto sdp = string(*desc);
+			const char *data = sdp.data();
+			size = std::min(size - 1, int(sdp.size()));
+			std::copy(data, data + size, buffer);
+			buffer[size] = '\0';
+			return size + 1;
+		}
+
+		return RTC_ERR_FAILURE;
+	});
+}
+
 int rtcGetLocalAddress(int pc, char *buffer, int size) {
 	return WRAP({
 		auto peerConnection = getPeerConnection(pc);
@@ -535,6 +592,8 @@ int rtcGetLocalAddress(int pc, char *buffer, int size) {
 			buffer[size] = '\0';
 			return size + 1;
 		}
+
+		return RTC_ERR_FAILURE;
 	});
 }
 
@@ -555,6 +614,8 @@ int rtcGetRemoteAddress(int pc, char *buffer, int size) {
 			buffer[size] = '\0';
 			return int(size + 1);
 		}
+
+		return RTC_ERR_FAILURE;
 	});
 }
 

src/datachannel.cpp

@@ -218,6 +218,8 @@ void DataChannel::incoming(message_ptr message) {
 
 	switch (message->type) {
 	case Message::Control: {
+		if (message->size() == 0)
+			break; // Ignore
 		auto raw = reinterpret_cast<const uint8_t *>(message->data());
 		switch (raw[0]) {
 		case MESSAGE_OPEN:

src/dtlstransport.cpp

@@ -435,7 +435,7 @@ void DtlsTransport::runRecvLoop() {
 
 		const size_t bufferSize = maxMtu;
 		byte buffer[bufferSize];
-		while (true) {
+		while (mIncomingQueue.running()) {
 			// Process pending messages
 			while (auto next = mIncomingQueue.tryPop()) {
 				message_ptr message = std::move(*next);
@@ -492,8 +492,7 @@ void DtlsTransport::runRecvLoop() {
 				}
 			}
 
-			if (!mIncomingQueue.wait(duration))
+			mIncomingQueue.wait(duration);
-				break; // queue is stopped
 		}
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "DTLS recv: " << e.what();

src/icetransport.cpp

@@ -60,7 +60,8 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 	}
 
 	juice_log_level_t level;
-	switch (plog::get()->getMaxSeverity()) {
+	auto logger = plog::get();
+	switch (logger ? logger->getMaxSeverity() : plog::none) {
 	case plog::none:
 		level = JUICE_LOG_LEVEL_NONE;
 		break;

src/log.cpp

@@ -24,6 +24,8 @@
 #include "plog/Log.h"
 #include "plog/Logger.h"
 
+#include <mutex>
+
 namespace rtc {
 
 void InitLogger(LogLevel level) { InitLogger(static_cast<plog::Severity>(level)); }
@@ -31,6 +33,8 @@ 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;
+	static std::mutex mutex;
+	std::lock_guard lock(mutex);
 	if (!logger) {
 		logger = &plog::init(severity, appender ? appender : &consoleAppender);
 		PLOG_DEBUG << "Logger initialized";

src/peerconnection.cpp

@@ -102,6 +102,7 @@ void PeerConnection::setLocalDescription() {
 
 	if (std::atomic_load(&mIceTransport)) {
 		PLOG_DEBUG << "Local description is already set, ignoring";
+		return;
 	}
 
 	// RFC 5763: The endpoint that is the offerer MUST use the setup attribute value of

src/sctptransport.cpp

@@ -476,8 +476,6 @@ int SctpTransport::handleRecv(struct socket * /*sock*/, union sctp_sockstore /*a
                               const byte *data, size_t len, struct sctp_rcvinfo info, int flags) {
 	try {
 		PLOG_VERBOSE << "Handle recv, len=" << len;
-		if (!len)
-			return 0; // Ignore
 
 		// SCTP_FRAGMENT_INTERLEAVE does not seem to work as expected for messages > 64KB,
 		// therefore partial notifications and messages need to be handled separately.
@@ -497,7 +495,7 @@ int SctpTransport::handleRecv(struct socket * /*sock*/, union sctp_sockstore /*a
 			if (flags & MSG_EOR) {
 				// Message is complete, process it
 				processData(std::move(mPartialMessage), info.rcv_sid,
-				            PayloadId(htonl(info.rcv_ppid)));
+				            PayloadId(ntohl(info.rcv_ppid)));
 				mPartialMessage.clear();
 			}
 		}

src/websocket.cpp

@@ -51,31 +51,45 @@ WebSocket::~WebSocket() {
 WebSocket::State WebSocket::readyState() const { return mState; }
 
 void WebSocket::open(const string &url) {
+	PLOG_VERBOSE << "Opening WebSocket to URL: " << url;
+
 	if (mState != State::Closed)
-		throw std::runtime_error("WebSocket must be closed before opening");
+		throw std::logic_error("WebSocket must be closed before opening");
 
-	static const char *rs = R"(^(([^:\/?#]+):)?(//([^\/?#]*))?([^?#]*)(\?([^#]*))?(#(.*))?)";
+	// Modified regex from RFC 3986, see https://tools.ietf.org/html/rfc3986#appendix-B
-	static std::regex regex(rs, std::regex::extended);
+	static const char *rs =
+	    R"(^(([^:.@/?#]+):)?(/{0,2}((([^:@]*)(:([^@]*))?)@)?(([^:/?#]*)(:([^/?#]*))?))?([^?#]*)(\?([^#]*))?(#(.*))?)";
 
-	std::smatch match;
+	static const std::regex r(rs, std::regex::extended);
-	if (!std::regex_match(url, match, regex))
-		throw std::invalid_argument("Malformed WebSocket URL: " + url);
 
-	mScheme = match[2];
+	std::smatch m;
-	if (mScheme != "ws" && mScheme != "wss")
+	if (!std::regex_match(url, m, r) || m[10].length() == 0)
+		throw std::invalid_argument("Invalid WebSocket URL: " + url);
+
+	mScheme = m[2];
+	if (mScheme.empty())
+		mScheme = "ws";
+	else if (mScheme != "ws" && mScheme != "wss")
 		throw std::invalid_argument("Invalid WebSocket scheme: " + mScheme);
 
-	mHost = match[4];
+	mHostname = m[10];
-	if (auto pos = mHost.find(':'); pos != string::npos) {
+	mService = m[12];
-		mHostname = mHost.substr(0, pos);
+	if (mService.empty()) {
-		mService = mHost.substr(pos + 1);
-	} else {
-		mHostname = mHost;
 		mService = mScheme == "ws" ? "80" : "443";
+		mHost = mHostname;
+	} else {
+		mHost = mHostname + ':' + mService;
 	}
 
-	mPath = match[5];
+	while (!mHostname.empty() && mHostname.front() == '[')
-	if (string query = match[7]; !query.empty())
+		mHostname.erase(mHostname.begin());
+	while (!mHostname.empty() && mHostname.back() == ']')
+		mHostname.pop_back();
+
+	mPath = m[13];
+	if (mPath.empty())
+		mPath += '/';
+	if (string query = m[15]; !query.empty())
 		mPath += "?" + query;
 
 	changeState(State::Connecting);

src/wstransport.cpp

@@ -58,6 +58,12 @@ WsTransport::WsTransport(std::shared_ptr<Transport> lower, string host, string p
 	onRecv(recvCallback);
 
 	PLOG_DEBUG << "Initializing WebSocket transport";
+
+	if (mHost.empty())
+		throw std::invalid_argument("WebSocket HTTP host cannot be empty");
+
+	if (mPath.empty())
+		throw std::invalid_argument("WebSocket HTTP path cannot be empty");
 }
 
 WsTransport::~WsTransport() { stop(); }
@@ -147,7 +153,7 @@ void WsTransport::close() {
 }
 
 bool WsTransport::sendHttpRequest() {
-	PLOG_DEBUG << "Sending WebSocket HTTP request";
+	PLOG_DEBUG << "Sending WebSocket HTTP request for path " << mPath;
 	changeState(State::Connecting);
 
 	auto seed = static_cast<unsigned int>(system_clock::now().time_since_epoch().count());

test/capi_connectivity.cpp

@@ -29,6 +29,8 @@ static void sleep(unsigned int secs) { Sleep(secs * 1000); }
 #include <unistd.h> // for sleep
 #endif
 
+#define BUFFER_SIZE 4096
+
 typedef struct {
 	rtcState state;
 	rtcGatheringState gatheringState;
@@ -183,15 +185,55 @@ int test_capi_connectivity_main() {
 		goto error;
 	}
 
-	char buffer[256];
+	char buffer[BUFFER_SIZE];
-	if (rtcGetLocalAddress(peer1->pc, buffer, 256) >= 0)
+
-		printf("Local address 1:  %s\n", buffer);
+	if (rtcGetLocalDescription(peer1->pc, buffer, BUFFER_SIZE) < 0) {
-	if (rtcGetRemoteAddress(peer1->pc, buffer, 256) >= 0)
+		fprintf(stderr, "rtcGetLocalDescription failed\n");
-		printf("Remote address 1: %s\n", buffer);
+		goto error;
-	if (rtcGetLocalAddress(peer2->pc, buffer, 256) >= 0)
+	}
-		printf("Local address 2:  %s\n", buffer);
+	printf("Local description 1:  %s\n", buffer);
-	if (rtcGetRemoteAddress(peer2->pc, buffer, 256) >= 0)
+
-		printf("Remote address 2: %s\n", buffer);
+	if (rtcGetRemoteDescription(peer1->pc, buffer, BUFFER_SIZE) < 0) {
+		fprintf(stderr, "rtcGetRemoteDescription failed\n");
+		goto error;
+	}
+	printf("Remote description 1:  %s\n", buffer);
+
+	if (rtcGetLocalDescription(peer2->pc, buffer, BUFFER_SIZE) < 0) {
+		fprintf(stderr, "rtcGetLocalDescription failed\n");
+		goto error;
+	}
+	printf("Local description 2:  %s\n", buffer);
+
+	if (rtcGetRemoteDescription(peer2->pc, buffer, BUFFER_SIZE) < 0) {
+		fprintf(stderr, "rtcGetRemoteDescription failed\n");
+		goto error;
+	}
+	printf("Remote description 2:  %s\n", buffer);
+
+	if (rtcGetLocalAddress(peer1->pc, buffer, BUFFER_SIZE) < 0) {
+		fprintf(stderr, "rtcGetLocalAddress failed\n");
+		goto error;
+	}
+	printf("Local address 1:  %s\n", buffer);
+
+	if (rtcGetRemoteAddress(peer1->pc, buffer, BUFFER_SIZE) < 0) {
+		fprintf(stderr, "rtcGetRemoteAddress failed\n");
+		goto error;
+	}
+	printf("Remote address 1: %s\n", buffer);
+
+	if (rtcGetLocalAddress(peer2->pc, buffer, BUFFER_SIZE) < 0) {
+		fprintf(stderr, "rtcGetLocalAddress failed\n");
+		goto error;
+	}
+	printf("Local address 2:  %s\n", buffer);
+
+	if (rtcGetRemoteAddress(peer2->pc, buffer, BUFFER_SIZE) < 0) {
+		fprintf(stderr, "rtcGetRemoteAddress failed\n");
+		goto error;
+	}
+	printf("Remote address 2: %s\n", buffer);
 
 	deletePeer(peer1);
 	sleep(1);

test/connectivity.cpp

@@ -125,6 +125,7 @@ void test_connectivity() {
 		}
 	});
 
+	// Wait a bit
 	int attempts = 10;
 	shared_ptr<DataChannel> adc2;
 	while ((!(adc2 = std::atomic_load(&dc2)) || !adc2->isOpen() || !dc1->isOpen()) && attempts--)
@@ -146,6 +147,49 @@ void test_connectivity() {
 	if (auto addr = pc2->remoteAddress())
 		cout << "Remote address 2: " << *addr << endl;
 
+	// Try to open a second data channel with another label
+	shared_ptr<DataChannel> second2;
+	pc2->onDataChannel([&second2](shared_ptr<DataChannel> dc) {
+		cout << "Second DataChannel 2: Received with label \"" << dc->label() << "\"" << endl;
+		if (dc->label() != "second") {
+			cerr << "Wrong second DataChannel label" << endl;
+			return;
+		}
+
+		dc->onMessage([](variant<binary, string> message) {
+			if (holds_alternative<string>(message)) {
+				cout << "Second Message 2: " << get<string>(message) << endl;
+			}
+		});
+
+		dc->send("Send hello from 2");
+
+		std::atomic_store(&second2, dc);
+	});
+
+	auto second1 = pc1->createDataChannel("second");
+	second1->onOpen([wsecond1 = make_weak_ptr(dc1)]() {
+		auto second1 = wsecond1.lock();
+		if (!second1)
+			return;
+
+		cout << "Second DataChannel 1: Open" << endl;
+		second1->send("Second hello from 1");
+	});
+	dc1->onMessage([](const variant<binary, string> &message) {
+		if (holds_alternative<string>(message)) {
+			cout << "Second Message 1: " << get<string>(message) << endl;
+		}
+	});
+
+	// Wait a bit
+	attempts = 10;
+	shared_ptr<DataChannel> asecond2;
+	while (
+	    (!(asecond2 = std::atomic_load(&second2)) || !asecond2->isOpen() || !second1->isOpen()) &&
+	    attempts--)
+		this_thread::sleep_for(1s);
+
 	// Delay close of peer 2 to check closing works properly
 	pc1->close();
 	this_thread::sleep_for(1s);