Merge branch 'v0.16'

.github/workflows/build-gnutls.yml

@@ -4,8 +4,6 @@ on:
     branches:
     - master
   pull_request:
-    branches:
-    - master
 jobs:
   build-linux:
     runs-on: ubuntu-latest

.github/workflows/build-nice.yml

@@ -4,8 +4,6 @@ on:
     branches:
     - master
   pull_request:
-    branches:
-    - master
 jobs:
   build-media:
     runs-on: ubuntu-latest

.github/workflows/build-openssl.yml

@@ -4,8 +4,6 @@ on:
     branches:
     - master
   pull_request:
-    branches:
-    - master
 jobs:
   build-linux:
     runs-on: ubuntu-latest

deps/libjuice

@@ -1 +1 @@
-Subproject commit 017144601b2b0c73baf8ae4e0c36d3637c73c3f9
+Subproject commit 5d154a4143964ccb2bbb69e1a325746b517ba50f

src/impl/datachannel.cpp

@@ -75,8 +75,13 @@ struct CloseMessage {
 };
 #pragma pack(pop)
 
-LogCounter COUNTER_USERNEG_OPEN_MESSAGE(
-    plog::warning, "Number of open messages for a user-negotiated DataChannel received");
+bool DataChannel::IsOpenMessage(message_ptr message) {
+	if (message->type != Message::Control)
+		return false;
+
+	auto raw = reinterpret_cast<const uint8_t *>(message->data());
+	return !message->empty() && raw[0] == MESSAGE_OPEN;
+}
 
 DataChannel::DataChannel(weak_ptr<PeerConnection> pc, uint16_t stream, string label,
                          string protocol, Reliability reliability)
@@ -193,8 +198,7 @@ void DataChannel::open(shared_ptr<SctpTransport> transport) {
 }
 
 void DataChannel::processOpenMessage(message_ptr) {
-	PLOG_DEBUG << "Received an open message for a user-negotiated DataChannel, ignoring";
-	COUNTER_USERNEG_OPEN_MESSAGE++;
+	PLOG_WARNING << "Received an open message for a user-negotiated DataChannel, ignoring";
 }
 
 bool DataChannel::outgoing(message_ptr message) {
@@ -218,7 +222,7 @@ bool DataChannel::outgoing(message_ptr message) {
 }
 
 void DataChannel::incoming(message_ptr message) {
-	if (!message)
+	if (!message || mIsClosed)
 		return;
 
 	switch (message->type) {
@@ -261,12 +265,6 @@ NegotiatedDataChannel::NegotiatedDataChannel(weak_ptr<PeerConnection> pc, uint16
                                              string label, string protocol, Reliability reliability)
     : DataChannel(pc, stream, std::move(label), std::move(protocol), std::move(reliability)) {}
 
-NegotiatedDataChannel::NegotiatedDataChannel(weak_ptr<PeerConnection> pc,
-                                             weak_ptr<SctpTransport> transport, uint16_t stream)
-    : DataChannel(pc, stream, "", "", {}) {
-	mSctpTransport = transport;
-}
-
 NegotiatedDataChannel::~NegotiatedDataChannel() {}
 
 void NegotiatedDataChannel::open(shared_ptr<SctpTransport> transport) {
@@ -314,7 +312,23 @@ void NegotiatedDataChannel::open(shared_ptr<SctpTransport> transport) {
 	transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
 }
 
-void NegotiatedDataChannel::processOpenMessage(message_ptr message) {
+void NegotiatedDataChannel::processOpenMessage(message_ptr) {
+	PLOG_WARNING << "Received an open message for a locally-created DataChannel, ignoring";
+}
+
+IncomingDataChannel::IncomingDataChannel(weak_ptr<PeerConnection> pc,
+                                         weak_ptr<SctpTransport> transport, uint16_t stream)
+    : DataChannel(pc, stream, "", "", {}) {
+	mSctpTransport = transport;
+}
+
+IncomingDataChannel::~IncomingDataChannel() {}
+
+void IncomingDataChannel::open(shared_ptr<SctpTransport>) {
+	// Ignore
+}
+
+void IncomingDataChannel::processOpenMessage(message_ptr message) {
 	std::unique_lock lock(mMutex);
 	auto transport = mSctpTransport.lock();
 	if (!transport)

src/impl/datachannel.hpp

@@ -35,6 +35,8 @@ namespace rtc::impl {
 struct PeerConnection;
 
 struct DataChannel : Channel, std::enable_shared_from_this<DataChannel> {
+	static bool IsOpenMessage(message_ptr message);
+
 	DataChannel(weak_ptr<PeerConnection> pc, uint16_t stream, string label, string protocol,
 	            Reliability reliability);
 	virtual ~DataChannel();
@@ -82,11 +84,18 @@ protected:
 struct NegotiatedDataChannel final : public DataChannel {
 	NegotiatedDataChannel(weak_ptr<PeerConnection> pc, uint16_t stream, string label,
 	                      string protocol, Reliability reliability);
-	NegotiatedDataChannel(weak_ptr<PeerConnection> pc, weak_ptr<SctpTransport> transport,
-	                      uint16_t stream);
 	~NegotiatedDataChannel();
 
-	void open(impl_ptr<SctpTransport> transport) override;
+	void open(shared_ptr<SctpTransport> transport) override;
+	void processOpenMessage(message_ptr message) override;
+};
+
+struct IncomingDataChannel final : public DataChannel {
+	IncomingDataChannel(weak_ptr<PeerConnection> pc, weak_ptr<SctpTransport> transport,
+	                    uint16_t stream);
+	~IncomingDataChannel();
+
+	void open(shared_ptr<SctpTransport> transport) override;
 	void processOpenMessage(message_ptr message) override;
 };
 

src/impl/peerconnection.cpp

@@ -419,42 +419,47 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		return;
 	}
 
-	uint16_t stream = uint16_t(message->stream);
+	const uint16_t stream = uint16_t(message->stream);
 	auto channel = findDataChannel(stream);
-	if (!channel) {
+
+	if (DataChannel::IsOpenMessage(message)) {
 		auto iceTransport = getIceTransport();
 		auto sctpTransport = getSctpTransport();
 		if (!iceTransport || !sctpTransport)
 			return;
 
-		// See https://www.rfc-editor.org/rfc/rfc8832.html
-		const byte dataChannelOpenMessage{0x03};
-		uint16_t remoteParity = (iceTransport->role() == Description::Role::Active) ? 1 : 0;
-		if (message->type == Message::Control) {
-			if (message->size() == 0 || *message->data() != dataChannelOpenMessage)
-				return; // ignore
-
-			if (stream % 2 != remoteParity) {
-				// The odd/even rule is violated, close the DataChannel
-				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
+			PLOG_WARNING << "Got open message violating the odd/even rule on stream " << stream;
+			sctpTransport->closeStream(message->stream);
+			return;
+		}
 
-			channel =
-			    std::make_shared<NegotiatedDataChannel>(weak_from_this(), sctpTransport, stream);
-			channel->openCallback = weak_bind(&PeerConnection::triggerDataChannel, this,
-			                                  weak_ptr<DataChannel>{channel});
+		if (channel && channel->isOpen()) {
+			PLOG_WARNING << "Got open message on stream " << stream
+			             << " for an already open DataChannel, closing it first";
+			channel->close();
+		}
 
-			std::unique_lock lock(mDataChannelsMutex); // we are going to emplace
-			mDataChannels.emplace(stream, channel);
+		channel = std::make_shared<IncomingDataChannel>(weak_from_this(), sctpTransport, stream);
+		channel->openCallback =
+		    weak_bind(&PeerConnection::triggerDataChannel, this, weak_ptr<DataChannel>{channel});
 
-		} else {
-			// Invalid, close the DataChannel
+		std::unique_lock lock(mDataChannelsMutex); // we are going to emplace
+		mDataChannels.emplace(stream, channel);
+	}
+
+	if (!channel) {
+		// Invalid, close the DataChannel
+		PLOG_WARNING << "Got unexpected message on stream " << stream;
+		if (auto sctpTransport = getSctpTransport())
 			sctpTransport->closeStream(message->stream);
-			return;
-		}
+
+		return;
 	}
 
+	// Forward the message
 	channel->incoming(message);
 }
 

test/capi_connectivity.cpp

@@ -97,6 +97,7 @@ static void RTC_API openCallback(int id, void *ptr) {
 static void RTC_API closedCallback(int id, void *ptr) {
 	Peer *peer = (Peer *)ptr;
 	peer->connected = false;
+	printf("DataChannel %d: Closed\n", peer == peer1 ? 1 : 2);
 }
 
 static void RTC_API messageCallback(int id, const char *message, int size, void *ptr) {
@@ -152,9 +153,6 @@ static void RTC_API dataChannelCallback(int pc, int dc, void *ptr) {
 	rtcSetMessageCallback(dc, messageCallback);
 
 	peer->dc = dc;
-
-	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) {

test/connectivity.cpp

@@ -108,8 +108,10 @@ void test_connectivity() {
 		}
 
 		dc->onOpen([wdc = make_weak_ptr(dc)]() {
-			if (auto dc = wdc.lock())
+			if (auto dc = wdc.lock()) {
+				cout << "DataChannel 2: Open" << endl;
 				dc->send("Hello from 2");
+			}
 		});
 
 		dc->onClosed([]() {
@@ -157,7 +159,8 @@ void test_connectivity() {
 	if (!adc2 || !adc2->isOpen() || !dc1->isOpen())
 		throw runtime_error("DataChannel is not open");
 
-	if (dc1->maxMessageSize() != CUSTOM_MAX_MESSAGE_SIZE || dc2->maxMessageSize() != CUSTOM_MAX_MESSAGE_SIZE)
+	if (dc1->maxMessageSize() != CUSTOM_MAX_MESSAGE_SIZE ||
+	    dc2->maxMessageSize() != CUSTOM_MAX_MESSAGE_SIZE)
 		throw runtime_error("DataChannel max message size is incorrect");
 
 	if (auto addr = pc1.localAddress())

test/turn_connectivity.cpp

@@ -105,8 +105,10 @@ void test_turn_connectivity() {
 		}
 
 		dc->onOpen([wdc = make_weak_ptr(dc)]() {
-			if (auto dc = wdc.lock())
+			if (auto dc = wdc.lock()) {
+				cout << "DataChannel 2: Open" << endl;
 				dc->send("Hello from 2");
+			}
 		});
 
 		dc->onMessage([](variant<binary, string> message) {
@@ -168,7 +170,7 @@ void test_turn_connectivity() {
 	cout << "Local candidate 1:  " << local << endl;
 	cout << "Remote candidate 1: " << remote << endl;
 
-	if(local.type() != Candidate::Type::Relayed || remote.type() != Candidate::Type::Relayed)
+	if (local.type() != Candidate::Type::Relayed || remote.type() != Candidate::Type::Relayed)
 		throw runtime_error("Connection is not relayed as expected");
 
 	if (!pc2.getSelectedCandidatePair(&local, &remote))
@@ -177,7 +179,7 @@ void test_turn_connectivity() {
 	cout << "Local candidate 2:  " << local << endl;
 	cout << "Remote candidate 2: " << remote << endl;
 
-	if(local.type() != Candidate::Type::Relayed || remote.type() != Candidate::Type::Relayed)
+	if (local.type() != Candidate::Type::Relayed || remote.type() != Candidate::Type::Relayed)
 		throw runtime_error("Connection is not relayed as expected");
 
 	// Try to open a second data channel with another label