Bumped version to 0.7.1

Close data channels  asynchronously for safety

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required(VERSION 3.7)
 project(libdatachannel
 	DESCRIPTION "WebRTC Data Channels Library"
-	VERSION 0.7.0
+	VERSION 0.7.1
 	LANGUAGES CXX)
 
 # Options

README.md

@@ -197,5 +197,6 @@ ws->open("wss://my.websocket/service");
 ```
 
 ## External resources
-- Rust wrappers for libdatachannel: [lerouxrgd/datachannel-rs](https://github.com/lerouxrgd/datachannel-rs)
+- Rust wrappers for libdatachannel: [datachannel-rs](https://github.com/lerouxrgd/datachannel-rs)
+- WebAssembly wrappers compatible with libdatachannel: [datachannel-wasm](https://github.com/paullouisageneau/datachannel-wasm)
 

include/rtc/include.hpp

@@ -84,9 +84,24 @@ template <typename F, typename T, typename... Args> auto weak_bind(F &&f, T *t,
 template <typename... P> class synchronized_callback {
 public:
 	synchronized_callback() = default;
+	synchronized_callback(synchronized_callback &&cb) { *this = std::move(cb); }
+	synchronized_callback(const synchronized_callback &cb) { *this = cb; }
 	synchronized_callback(std::function<void(P...)> func) { *this = std::move(func); }
 	~synchronized_callback() { *this = nullptr; }
 
+	synchronized_callback &operator=(synchronized_callback &&cb) {
+		std::scoped_lock lock(mutex, cb.mutex);
+		callback = std::move(cb.callback);
+		cb.callback = nullptr;
+		return *this;
+	}
+
+	synchronized_callback &operator=(const synchronized_callback &cb) {
+		std::scoped_lock lock(mutex, cb.mutex);
+		callback = cb.callback;
+		return *this;
+	}
+
 	synchronized_callback &operator=(std::function<void(P...)> func) {
 		std::lock_guard lock(mutex);
 		callback = std::move(func);

include/rtc/message.hpp

@@ -37,6 +37,8 @@ struct Message : binary {
 	Message(Iterator begin_, Iterator end_, Type type_ = Binary)
 	    : binary(begin_, end_), type(type_) {}
 
+	Message(binary &&data, Type type_ = Binary) : binary(std::move(data)), type(type_) {}
+
 	Type type;
 	unsigned int stream = 0;
 	std::shared_ptr<Reliability> reliability;
@@ -68,6 +70,15 @@ inline message_ptr make_message(size_t size, Message::Type type = Message::Binar
 	return message;
 }
 
+inline message_ptr make_message(binary &&data, Message::Type type = Message::Binary,
+                                unsigned int stream = 0,
+                                std::shared_ptr<Reliability> reliability = nullptr) {
+	auto message = std::make_shared<Message>(std::move(data), type);
+	message->stream = stream;
+	message->reliability = reliability;
+	return message;
+}
+
 } // namespace rtc
 
 #endif

src/peerconnection.cpp

@@ -51,13 +51,17 @@ PeerConnection::PeerConnection(const Configuration &config)
 }
 
 PeerConnection::~PeerConnection() {
-	close();
 	PLOG_VERBOSE << "Destroying PeerConnection";
+	close();
+	mProcessor->join();
 }
 
 void PeerConnection::close() {
 	PLOG_VERBOSE << "Closing PeerConnection";
-	closeDataChannels();
+
+	// Close data channels asynchronously
+	mProcessor->enqueue(std::bind(&PeerConnection::closeDataChannels, this));
+
 	closeTransports();
 }
 
@@ -439,13 +443,15 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 void PeerConnection::closeTransports() {
 	PLOG_VERBOSE << "Closing transports";
 
-	// Change state to sink state Closed to block init methods
+	// Change state to sink state Closed
 	changeState(State::Closed);
 
 	// Reset callbacks now that state is changed
 	resetCallbacks();
 
-	// Pass the references to a thread, allowing to terminate a transport from its own thread
+	// Initiate transport stop on the processor after closing the data channels
+	mProcessor->enqueue([this]() {
+		// Pass the pointers to a thread
 		auto sctp = std::atomic_exchange(&mSctpTransport, decltype(mSctpTransport)(nullptr));
 		auto dtls = std::atomic_exchange(&mDtlsTransport, decltype(mDtlsTransport)(nullptr));
 		auto ice = std::atomic_exchange(&mIceTransport, decltype(mIceTransport)(nullptr));
@@ -461,6 +467,7 @@ void PeerConnection::closeTransports() {
 			dtls.reset();
 			ice.reset();
 		});
+	});
 }
 
 void PeerConnection::endLocalCandidates() {
@@ -648,7 +655,12 @@ bool PeerConnection::changeState(State state) {
 
 	} while (!mState.compare_exchange_weak(current, state));
 
+	if (state == State::Closed)
+		// This is the last state change, so we may steal the callback
+		mProcessor->enqueue([this, cb = std::move(mStateChangeCallback)]() { cb(State::Closed); });
+	else
 		mProcessor->enqueue([this, state]() { mStateChangeCallback(state); });
+
 	return true;
 }
 

src/sctptransport.cpp

@@ -451,7 +451,7 @@ void SctpTransport::sendReset(uint16_t streamId) {
 		mWrittenCondition.wait_for(lock, 1000ms,
 		                           [&]() { return mWritten || state() != State::Connected; });
 	} else if (errno == EINVAL) {
-		PLOG_VERBOSE << "SCTP stream " << streamId << " already reset";
+		PLOG_DEBUG << "SCTP stream " << streamId << " already reset";
 	} else {
 		PLOG_WARNING << "SCTP reset stream " << streamId << " failed, errno=" << errno;
 	}
@@ -479,32 +479,22 @@ int SctpTransport::handleRecv(struct socket * /*sock*/, union sctp_sockstore /*a
 		// therefore partial notifications and messages need to be handled separately.
 		if (flags & MSG_NOTIFICATION) {
 			// SCTP event notification
+			mPartialNotification.insert(mPartialNotification.end(), data, data + len);
 			if (flags & MSG_EOR) {
-				if (!mPartialNotification.empty()) {
-					mPartialNotification.insert(mPartialNotification.end(), data, data + len);
-					data = mPartialNotification.data();
-					len = mPartialNotification.size();
-				}
 				// Notification is complete, process it
-				processNotification(reinterpret_cast<const union sctp_notification *>(data), len);
+				processNotification(
+				    reinterpret_cast<const union sctp_notification *>(mPartialNotification.data()),
+				    mPartialNotification.size());
 				mPartialNotification.clear();
-			} else {
-				mPartialNotification.insert(mPartialNotification.end(), data, data + len);
 			}
-
 		} else {
 			// SCTP message
+			mPartialMessage.insert(mPartialMessage.end(), data, data + len);
 			if (flags & MSG_EOR) {
-				if (!mPartialMessage.empty()) {
-					mPartialMessage.insert(mPartialMessage.end(), data, data + len);
-					data = mPartialMessage.data();
-					len = mPartialMessage.size();
-				}
 				// Message is complete, process it
-				processData(data, len, info.rcv_sid, PayloadId(htonl(info.rcv_ppid)));
+				processData(std::move(mPartialMessage), info.rcv_sid,
+				            PayloadId(htonl(info.rcv_ppid)));
 				mPartialMessage.clear();
-			} else {
-				mPartialMessage.insert(mPartialMessage.end(), data, data + len);
 			}
 		}
 
@@ -539,62 +529,56 @@ int SctpTransport::handleWrite(byte *data, size_t len, uint8_t /*tos*/, uint8_t
 	return 0; // success
 }
 
-void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, PayloadId ppid) {
-	PLOG_VERBOSE << "Process data, len=" << len;
+void SctpTransport::processData(binary &&data, uint16_t sid, PayloadId ppid) {
+	PLOG_VERBOSE << "Process data, size=" << data.size();
 
 	// 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.
 	switch (ppid) {
 	case PPID_CONTROL:
-		recv(make_message(data, data + len, Message::Control, sid));
+		recv(make_message(std::move(data), Message::Control, sid));
 		break;
 
 	case PPID_STRING_PARTIAL: // deprecated
-		mPartialStringData.insert(mPartialStringData.end(), data, data + len);
+		mPartialStringData.insert(mPartialStringData.end(), data.begin(), data.end());
 		break;
 
 	case PPID_STRING:
 		if (mPartialStringData.empty()) {
-			mBytesReceived += len;
-			recv(make_message(data, data + len, Message::String, sid));
+			mBytesReceived += data.size();
+			recv(make_message(std::move(data), Message::String, sid));
 		} else {
-			mPartialStringData.insert(mPartialStringData.end(), data, data + len);
+			mPartialStringData.insert(mPartialStringData.end(), data.begin(), data.end());
 			mBytesReceived += mPartialStringData.size();
-			recv(make_message(mPartialStringData.begin(), mPartialStringData.end(), Message::String,
-			                  sid));
+			recv(make_message(std::move(mPartialStringData), Message::String, sid));
 			mPartialStringData.clear();
 		}
 		break;
 
 	case PPID_STRING_EMPTY:
-		// This only accounts for when the partial data is empty
-		recv(make_message(mPartialStringData.begin(), mPartialStringData.end(), Message::String,
-		                  sid));
+		recv(make_message(std::move(mPartialStringData), Message::String, sid));
 		mPartialStringData.clear();
 		break;
 
 	case PPID_BINARY_PARTIAL: // deprecated
-		mPartialBinaryData.insert(mPartialBinaryData.end(), data, data + len);
+		mPartialBinaryData.insert(mPartialBinaryData.end(), data.begin(), data.end());
 		break;
 
 	case PPID_BINARY:
 		if (mPartialBinaryData.empty()) {
-			mBytesReceived += len;
-			recv(make_message(data, data + len, Message::Binary, sid));
+			mBytesReceived += data.size();
+			recv(make_message(std::move(data), Message::Binary, sid));
 		} else {
-			mPartialBinaryData.insert(mPartialBinaryData.end(), data, data + len);
-			mBytesReceived += mPartialStringData.size();
-			recv(make_message(mPartialBinaryData.begin(), mPartialBinaryData.end(), Message::Binary,
-			                  sid));
+			mPartialBinaryData.insert(mPartialBinaryData.end(), data.begin(), data.end());
+			mBytesReceived += mPartialBinaryData.size();
+			recv(make_message(std::move(mPartialBinaryData), Message::Binary, sid));
 			mPartialBinaryData.clear();
 		}
 		break;
 
 	case PPID_BINARY_EMPTY:
-		// This only accounts for when the partial data is empty
-		recv(make_message(mPartialBinaryData.begin(), mPartialBinaryData.end(), Message::Binary,
-		                  sid));
+		recv(make_message(std::move(mPartialBinaryData), Message::Binary, sid));
 		mPartialBinaryData.clear();
 		break;
 

src/sctptransport.hpp

@@ -86,7 +86,7 @@ private:
 	int handleSend(size_t free);
 	int handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_df);
 
-	void processData(const byte *data, size_t len, uint16_t streamId, PayloadId ppid);
+	void processData(binary &&data, uint16_t streamId, PayloadId ppid);
 	void processNotification(const union sctp_notification *notify, size_t len);
 
 	const uint16_t mPort;

src/transport.hpp

@@ -65,10 +65,20 @@ public:
 	virtual bool send(message_ptr message) { return outgoing(message); }
 
 protected:
-	void recv(message_ptr message) { mRecvCallback(message); }
+	void recv(message_ptr message) {
+		try {
+			mRecvCallback(message);
+		} catch (const std::exception &e) {
+			PLOG_WARNING << e.what();
+		}
+	}
 	void changeState(State state) {
+		try {
 			if (mState.exchange(state) != state)
 				mStateChangeCallback(state);
+		} catch (const std::exception &e) {
+			PLOG_WARNING << e.what();
+		}
 	}
 
 	virtual void incoming(message_ptr message) { recv(message); }

src/websocket.cpp

@@ -317,7 +317,7 @@ void WebSocket::closeTransports() {
 	// Reset callbacks now that state is changed
 	resetCallbacks();
 
-	// Pass the references to a thread, allowing to terminate a transport from its own thread
+	// Pass the pointers to a thread, allowing to terminate a transport from its own thread
 	auto ws = std::atomic_exchange(&mWsTransport, decltype(mWsTransport)(nullptr));
 	auto tls = std::atomic_exchange(&mTlsTransport, decltype(mTlsTransport)(nullptr));
 	auto tcp = std::atomic_exchange(&mTcpTransport, decltype(mTcpTransport)(nullptr));