Merge branch 'v0.18'

Author: paullouisageneau

src/datachannel.cpp

@@ -26,13 +26,7 @@ DataChannel::DataChannel(impl_ptr<impl::DataChannel> impl)
     : CheshireCat<impl::DataChannel>(impl),
       Channel(std::dynamic_pointer_cast<impl::Channel>(impl)) {}
 
-DataChannel::~DataChannel() {
-	try {
-		close();
-	} catch (const std::exception &e) {
-		PLOG_ERROR << e.what();
-	}
-}
+DataChannel::~DataChannel() {}
 
 void DataChannel::close() { return impl()->close(); }
 

src/impl/datachannel.cpp

@@ -79,8 +79,11 @@ DataChannel::DataChannel(weak_ptr<PeerConnection> pc, string label, string proto
 
 DataChannel::~DataChannel() {
 	PLOG_VERBOSE << "Destroying DataChannel";
-
-	close();
+	try {
+		close();
+	} catch (const std::exception &e) {
+		PLOG_ERROR << e.what();
+	}
 }
 
 void DataChannel::close() {
@@ -102,9 +105,7 @@ void DataChannel::close() {
 }
 
 void DataChannel::remoteClose() {
-	mIsOpen = false;
-	if (!mIsClosed.exchange(true))
-		triggerClosed();
+	close();
 }
 
 optional<message_variant> DataChannel::receive() {

src/impl/peerconnection.cpp

@@ -440,23 +440,28 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		return;
 
 	const uint16_t stream = uint16_t(message->stream);
-	auto channel = findDataChannel(stream);
+	auto [channel, found] = findDataChannel(stream);
 
 	if (DataChannel::IsOpenMessage(message)) {
+		if (found) {
+			// The stream is already used, the receiver must close the DataChannel
+			PLOG_WARNING << "Got open message on already used stream " << stream;
+			if(channel && channel->isOpen())
+				channel->close();
+			else
+				sctpTransport->closeStream(message->stream);
+
+			return;
+		}
+
 		const uint16_t remoteParity = (iceTransport->role() == Description::Role::Active) ? 1 : 0;
 		if (stream % 2 != remoteParity) {
-			// The odd/even rule is violated, close the DataChannel
+			// The odd/even rule is violated, the receiver must close the DataChannel
 			PLOG_WARNING << "Got open message violating the odd/even rule on stream " << stream;
 			sctpTransport->closeStream(message->stream);
 			return;
 		}
 
-		if (channel && channel->isOpen()) {
-			PLOG_WARNING << "Got open message on stream " << stream
-			             << " for an already open DataChannel, closing it first";
-			channel->close();
-		}
-
 		channel = std::make_shared<IncomingDataChannel>(weak_from_this(), sctpTransport);
 		channel->assignStream(stream);
 		channel->openCallback =
@@ -465,8 +470,7 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		std::unique_lock lock(mDataChannelsMutex); // we are going to emplace
 		mDataChannels.emplace(stream, channel);
 	}
-
-	if (!channel) {
+	else if (!found) {
 		if (message->type == Message::Reset)
 			return; // ignore
 
@@ -476,8 +480,18 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		return;
 	}
 
-	// Forward the message
-	channel->incoming(message);
+	if (message->type == Message::Reset) {
+		// Incoming stream is reset, unregister it
+		removeDataChannel(stream);
+	}
+
+	if (channel) {
+		// Forward the message
+		channel->incoming(message);
+	} else {
+		// DataChannel was destroyed, ignore
+		PLOG_DEBUG << "Ignored message on stream " << stream << ", DataChannel is destroyed";
+	}
 }
 
 void PeerConnection::forwardMedia(message_ptr message) {
@@ -571,12 +585,12 @@ void PeerConnection::forwardMedia(message_ptr message) {
 }
 
 void PeerConnection::forwardBufferedAmount(uint16_t stream, size_t amount) {
-	if (auto channel = findDataChannel(stream))
+	[[maybe_unused]] auto [channel, found] = findDataChannel(stream);
+	if (channel)
 		channel->triggerBufferedAmount(amount);
 }
 
 shared_ptr<DataChannel> PeerConnection::emplaceDataChannel(string label, DataChannelInit init) {
-	cleanupDataChannels();
 	std::unique_lock lock(mDataChannelsMutex); // we are going to emplace
 
 	// If the DataChannel is user-negotiated, do not negotiate it in-band
@@ -613,13 +627,17 @@ shared_ptr<DataChannel> PeerConnection::emplaceDataChannel(string label, DataCha
 	return channel;
 }
 
-shared_ptr<DataChannel> PeerConnection::findDataChannel(uint16_t stream) {
+std::pair<shared_ptr<DataChannel>, bool> PeerConnection::findDataChannel(uint16_t stream) {
 	std::shared_lock lock(mDataChannelsMutex); // read-only
 	if (auto it = mDataChannels.find(stream); it != mDataChannels.end())
-		if (auto channel = it->second.lock())
-			return channel;
+		return std::make_pair(it->second.lock(), true);
+	else
+		return std::make_pair(nullptr, false);
+}
 
-	return nullptr;
+bool PeerConnection::removeDataChannel(uint16_t stream) {
+		std::unique_lock lock(mDataChannelsMutex); // we are going to erase
+		return mDataChannels.erase(stream) != 0;
 }
 
 uint16_t PeerConnection::maxDataChannelStream() const {
@@ -650,8 +668,7 @@ void PeerConnection::assignDataChannels() {
 			if (stream > maxStream)
 				throw std::runtime_error("Too many DataChannels");
 
-			auto it = mDataChannels.find(stream);
-			if (it == mDataChannels.end() || !it->second.lock())
+			if (mDataChannels.find(stream) == mDataChannels.end())
 				break;
 
 			stream += 2;
@@ -691,19 +708,6 @@ void PeerConnection::iterateDataChannels(
 	}
 }
 
-void PeerConnection::cleanupDataChannels() {
-	std::unique_lock lock(mDataChannelsMutex); // we are going to erase
-	auto it = mDataChannels.begin();
-	while (it != mDataChannels.end()) {
-		if (!it->second.lock()) {
-			it = mDataChannels.erase(it);
-			continue;
-		}
-
-		++it;
-	}
-}
-
 void PeerConnection::openDataChannels() {
 	if (auto transport = std::atomic_load(&mSctpTransport))
 		iterateDataChannels([&](shared_ptr<DataChannel> channel) {

src/impl/peerconnection.hpp

@@ -59,11 +59,11 @@ struct PeerConnection : std::enable_shared_from_this<PeerConnection> {
 	void forwardBufferedAmount(uint16_t stream, size_t amount);
 
 	shared_ptr<DataChannel> emplaceDataChannel(string label, DataChannelInit init);
-	shared_ptr<DataChannel> findDataChannel(uint16_t stream);
+	std::pair<shared_ptr<DataChannel>, bool> findDataChannel(uint16_t stream);
+	bool removeDataChannel(uint16_t stream);
 	uint16_t maxDataChannelStream() const;
 	void assignDataChannels();
 	void iterateDataChannels(std::function<void(shared_ptr<DataChannel> channel)> func);
-	void cleanupDataChannels();
 	void openDataChannels();
 	void closeDataChannels();
 	void remoteCloseDataChannels();

src/impl/track.cpp

@@ -30,8 +30,11 @@ Track::Track(weak_ptr<PeerConnection> pc, Description::Media description)
 
 Track::~Track() {
 	PLOG_VERBOSE << "Destroying Track";
-
-	close();
+	try {
+		close();
+	} catch (const std::exception &e) {
+		PLOG_ERROR << e.what();
+	}
 }
 
 string Track::mid() const {