Bumped version to 0.10.2

Schedule SCTP recv task only if necessary

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.7)
 project(libdatachannel
-	VERSION 0.10.0
+	VERSION 0.10.2
 	LANGUAGES CXX)
 set(PROJECT_DESCRIPTION "WebRTC Data Channels Library")
 
@@ -115,17 +115,15 @@ set(CMAKE_POLICY_DEFAULT_CMP0048 NEW)
 add_subdirectory(deps/plog)
 
 option(sctp_build_programs 0)
+option(sctp_build_shared_lib 0)
 add_subdirectory(deps/usrsctp EXCLUDE_FROM_ALL)
 if (MSYS OR MINGW)
 	target_compile_definitions(usrsctp PUBLIC -DSCTP_STDINT_INCLUDE=<stdint.h>)
-	target_compile_definitions(usrsctp-static PUBLIC -DSCTP_STDINT_INCLUDE=<stdint.h>)
 endif()
 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)
 
 if (NO_WEBSOCKET)
 	add_library(datachannel SHARED
@@ -156,13 +154,13 @@ 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_link_libraries(datachannel PUBLIC Threads::Threads plog::plog)
-target_link_libraries(datachannel PRIVATE Usrsctp::UsrsctpStatic)
+target_link_libraries(datachannel PRIVATE Usrsctp::Usrsctp)
 
 target_include_directories(datachannel-static PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
 target_include_directories(datachannel-static PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include/rtc)
 target_include_directories(datachannel-static PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
 target_link_libraries(datachannel-static PUBLIC Threads::Threads plog::plog)
-target_link_libraries(datachannel-static PRIVATE Usrsctp::UsrsctpStatic)
+target_link_libraries(datachannel-static PRIVATE Usrsctp::Usrsctp)
 
 if(WIN32)
 	target_link_libraries(datachannel PRIVATE ws2_32) # winsock2

include/rtc/candidate.hpp

@@ -40,29 +40,38 @@ public:
 	enum class ResolveMode { Simple, Lookup };
 	bool resolve(ResolveMode mode = ResolveMode::Simple);
 
+	Type type() const;
+	TransportType transportType() const;
+	uint32_t priority() const;
 	string candidate() const;
 	string mid() const;
 	operator string() const;
 
+	bool operator==(const Candidate &other) const;
+	bool operator!=(const Candidate &other) const;
+
 	bool isResolved() const;
 	Family family() const;
-	Type type() const;
-	TransportType transportType() const;
 	std::optional<string> address() const;
 	std::optional<uint16_t> port() const;
-	std::optional<uint32_t> priority() const;
 
 private:
-	string mCandidate;
+	void parse(string candidate);
+
+	string mFoundation;
+	uint32_t mComponent, mPriority;
+	string mTypeString, mTransportString;
+	Type mType;
+	TransportType mTransportType;
+	string mNode, mService;
+	string mTail;
+
 	std::optional<string> mMid;
 
 	// Extracted on resolution
 	Family mFamily;
-	Type mType;
-	TransportType mTransportType;
 	string mAddress;
 	uint16_t mPort;
-	uint32_t mPriority;
 };
 
 } // namespace rtc

include/rtc/description.hpp

@@ -53,6 +53,7 @@ public:
 	void hintType(Type type);
 	void setFingerprint(string fingerprint);
 
+	bool hasCandidate(const Candidate &candidate) const;
 	void addCandidate(Candidate candidate);
 	void addCandidates(std::vector<Candidate> candidates);
 	void endCandidates();

include/rtc/include.hpp

@@ -102,12 +102,12 @@ private:
 	std::function<void()> function;
 };
 
-template <typename... P> class synchronized_callback {
+template <typename... Args> 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(std::function<void(Args...)> func) { *this = std::move(func); }
 	~synchronized_callback() { *this = nullptr; }
 
 	synchronized_callback &operator=(synchronized_callback &&cb) {
@@ -123,16 +123,16 @@ public:
 		return *this;
 	}
 
-	synchronized_callback &operator=(std::function<void(P...)> func) {
+	synchronized_callback &operator=(std::function<void(Args...)> func) {
 		std::lock_guard lock(mutex);
 		callback = std::move(func);
 		return *this;
 	}
 
-	void operator()(P... args) const {
+	void operator()(Args... args) const {
 		std::lock_guard lock(mutex);
 		if (callback)
-			callback(args...);
+			callback(std::move(args)...);
 	}
 
 	operator bool() const {
@@ -140,8 +140,12 @@ public:
 		return callback ? true : false;
 	}
 
+	std::function<void(Args...)> wrap() const {
+		return [this](Args... args) { (*this)(std::move(args)...); };
+	}
+
 private:
-	std::function<void(P...)> callback;
+	std::function<void(Args...)> callback;
 	mutable std::recursive_mutex mutex;
 };
 } // namespace rtc

include/rtc/peerconnection.hpp

@@ -129,7 +129,7 @@ public:
 	void onTrack(std::function<void(std::shared_ptr<Track> track)> callback);
 
 private:
-	std::shared_ptr<IceTransport> initIceTransport(Description::Role role);
+	std::shared_ptr<IceTransport> initIceTransport();
 	std::shared_ptr<DtlsTransport> initDtlsTransport();
 	std::shared_ptr<SctpTransport> initSctpTransport();
 	void closeTransports();

include/rtc/rtp.hpp

@@ -57,6 +57,7 @@ public:
 
 	inline uint8_t version() const { return _first >> 6; }
 	inline bool padding() const { return (_first >> 5) & 0x01; }
+	inline bool extension() const { return (_first >> 4) & 0x01; }
 	inline uint8_t csrcCount() const { return _first & 0x0F; }
 	inline uint8_t marker() const { return _payloadType & 0b10000000; }
 	inline uint8_t payloadType() const { return _payloadType & 0b01111111; }
@@ -77,6 +78,17 @@ public:
 	inline void setSsrc(uint32_t ssrc) { _ssrc = htonl(ssrc); }
 
 	void setTimestamp(uint32_t i) { _timestamp = htonl(i); }
+
+	void log() {
+		PLOG_VERBOSE  << "RTP V: " << (int) version()
+		            << " P: " << (padding() ? "P" : " ")
+		            << " X: " << (extension() ? "X" : " ")
+		            << " CC: "  << (int) csrcCount()
+		            << " M: " << (marker() ? "M" : " ")
+		            << " PT: " << (int) payloadType()
+		            << " SEQNO: " << seqNumber()
+		            << " TS: " << timestamp();
+	}
 };
 
 struct RTCP_ReportBlock {

src/candidate.cpp

@@ -20,6 +20,7 @@
 
 #include <algorithm>
 #include <array>
+#include <cctype>
 #include <sstream>
 #include <unordered_map>
 
@@ -39,39 +40,44 @@ using std::string;
 
 namespace {
 
-inline bool hasprefix(const string &str, const string &prefix) {
+inline bool match_prefix(const string &str, const string &prefix) {
 	return str.size() >= prefix.size() &&
 	       std::mismatch(prefix.begin(), prefix.end(), str.begin()).first == prefix.end();
 }
 
+inline void trim_begin(string &str) {
+	str.erase(str.begin(),
+	          std::find_if(str.begin(), str.end(), [](char c) { return !std::isspace(c); }));
+}
+
+inline void trim_end(string &str) {
+	str.erase(
+	    std::find_if(str.rbegin(), str.rend(), [](char c) { return !std::isspace(c); }).base(),
+	    str.end());
+}
+
 } // namespace
 
 namespace rtc {
 
 Candidate::Candidate()
-    : mFamily(Family::Unresolved), mType(Type::Unknown), mTransportType(TransportType::Unknown),
-      mPort(0), mPriority(0) {}
+    : mFoundation("none"), mComponent(0), mPriority(0), mTypeString("unknown"),
+      mTransportString("unknown"), mType(Type::Unknown), mTransportType(TransportType::Unknown),
+      mNode("0.0.0.0"), mService("9"), mFamily(Family::Unresolved), mPort(0) {}
 
 Candidate::Candidate(string candidate) : Candidate() {
-	const std::array prefixes{"a=", "candidate:"};
-	for (const string &prefix : prefixes)
-		if (hasprefix(candidate, prefix))
-			candidate.erase(0, prefix.size());
-
-	mCandidate = std::move(candidate);
+	if (!candidate.empty())
+		parse(std::move(candidate));
 }
 
-Candidate::Candidate(string candidate, string mid) : Candidate(std::move(candidate)) {
+Candidate::Candidate(string candidate, string mid) : Candidate() {
+	if (!candidate.empty())
+		parse(std::move(candidate));
 	if (!mid.empty())
 		mMid.emplace(std::move(mid));
 }
 
-void Candidate::hintMid(string mid) {
-	if (!mMid)
-		mMid.emplace(std::move(mid));
-}
-
-bool Candidate::resolve(ResolveMode mode) {
+void Candidate::parse(string candidate) {
 	using TypeMap_t = std::unordered_map<string, Type>;
 	using TcpTypeMap_t = std::unordered_map<string, TransportType>;
 
@@ -84,99 +90,122 @@ bool Candidate::resolve(ResolveMode mode) {
 	                                        {"passive", TransportType::TcpPassive},
 	                                        {"so", TransportType::TcpSo}};
 
-	if (mFamily != Family::Unresolved)
-		return true;
+	const std::array prefixes{"a=", "candidate:"};
+	for (const string &prefix : prefixes)
+		if (match_prefix(candidate, prefix))
+			candidate.erase(0, prefix.size());
 
-	if (mCandidate.empty())
-		throw std::logic_error("Candidate is empty");
-
-	PLOG_VERBOSE << "Resolving candidate (mode="
-	             << (mode == ResolveMode::Simple ? "simple" : "lookup") << "): " << mCandidate;
+	PLOG_VERBOSE << "Parsing candidate: " << candidate;
 
 	// See RFC 8445 for format
-	std::istringstream iss(mCandidate);
-	int component{0}, priority{0};
-	string foundation, transport, node, service, typ_, type;
-	if (iss >> foundation >> component >> transport >> priority &&
-	    iss >> node >> service >> typ_ >> type && typ_ == "typ") {
+	std::istringstream iss(candidate);
+	string transport, typ_, type;
+	if (!(iss >> mFoundation >> mComponent >> mTransportString >> mPriority &&
+	      iss >> mNode >> mService >> typ_ >> mTypeString && typ_ == "typ"))
+		throw std::invalid_argument("Invalid candidate format");
 
-		string left;
-		std::getline(iss, left);
+	std::getline(iss, mTail);
+	trim_begin(mTail);
+	trim_end(mTail);
 
-		if (auto it = TypeMap.find(type); it != TypeMap.end())
-			mType = it->second;
-		else
-			mType = Type::Unknown;
+	if (auto it = TypeMap.find(type); it != TypeMap.end())
+		mType = it->second;
+	else
+		mType = Type::Unknown;
 
-		if (transport == "UDP" || transport == "udp") {
-			mTransportType = TransportType::Udp;
-		} else if (transport == "TCP" || transport == "tcp") {
-			std::istringstream iss(left);
-			string tcptype_, tcptype;
-			if (iss >> tcptype_ >> tcptype && tcptype_ == "tcptype") {
-				if (auto it = TcpTypeMap.find(tcptype); it != TcpTypeMap.end())
-					mTransportType = it->second;
-				else
-					mTransportType = TransportType::TcpUnknown;
-
-			} else {
+	if (transport == "UDP" || transport == "udp") {
+		mTransportType = TransportType::Udp;
+	} else if (transport == "TCP" || transport == "tcp") {
+		// Peek tail to find TCP type
+		std::istringstream iss(mTail);
+		string tcptype_, tcptype;
+		if (iss >> tcptype_ >> tcptype && tcptype_ == "tcptype") {
+			if (auto it = TcpTypeMap.find(tcptype); it != TcpTypeMap.end())
+				mTransportType = it->second;
+			else
 				mTransportType = TransportType::TcpUnknown;
-			}
+
 		} else {
-			mTransportType = TransportType::Unknown;
+			mTransportType = TransportType::TcpUnknown;
 		}
+	} else {
+		mTransportType = TransportType::Unknown;
+	}
+}
 
-		// Try to resolve the node
-		struct addrinfo hints = {};
-		hints.ai_family = AF_UNSPEC;
-		hints.ai_flags = AI_ADDRCONFIG;
-		if (mTransportType == TransportType::Udp) {
-			hints.ai_socktype = SOCK_DGRAM;
-			hints.ai_protocol = IPPROTO_UDP;
-		} else if (mTransportType != TransportType::Unknown) {
-			hints.ai_socktype = SOCK_STREAM;
-			hints.ai_protocol = IPPROTO_TCP;
-		}
+void Candidate::hintMid(string mid) {
+	if (!mMid)
+		mMid.emplace(std::move(mid));
+}
 
-		if (mode == ResolveMode::Simple)
-			hints.ai_flags |= AI_NUMERICHOST;
+bool Candidate::resolve(ResolveMode mode) {
+	PLOG_VERBOSE << "Resolving candidate (mode="
+	             << (mode == ResolveMode::Simple ? "simple" : "lookup") << "): " << mNode << ' '
+	             << mService;
 
-		struct addrinfo *result = nullptr;
-		if (getaddrinfo(node.c_str(), service.c_str(), &hints, &result) == 0) {
-			for (auto p = result; p; p = p->ai_next) {
-				if (p->ai_family == AF_INET || p->ai_family == AF_INET6) {
-					// Rewrite the candidate
-					char nodebuffer[MAX_NUMERICNODE_LEN];
-					char servbuffer[MAX_NUMERICSERV_LEN];
-					if (getnameinfo(p->ai_addr, socklen_t(p->ai_addrlen), nodebuffer,
-					                MAX_NUMERICNODE_LEN, servbuffer, MAX_NUMERICSERV_LEN,
-					                NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
+	// Try to resolve the node and service
+	struct addrinfo hints = {};
+	hints.ai_family = AF_UNSPEC;
+	hints.ai_flags = AI_ADDRCONFIG;
+	if (mTransportType == TransportType::Udp) {
+		hints.ai_socktype = SOCK_DGRAM;
+		hints.ai_protocol = IPPROTO_UDP;
+	} else if (mTransportType != TransportType::Unknown) {
+		hints.ai_socktype = SOCK_STREAM;
+		hints.ai_protocol = IPPROTO_TCP;
+	}
 
-						mAddress = nodebuffer;
-						mPort = uint16_t(std::stoul(servbuffer));
-						mFamily = p->ai_family == AF_INET6 ? Family::Ipv6 : Family::Ipv4;
+	if (mode == ResolveMode::Simple)
+		hints.ai_flags |= AI_NUMERICHOST;
 
-						const char sp{' '};
-						std::ostringstream oss;
-						oss << foundation << sp << component << sp << transport << sp << priority;
-						oss << sp << nodebuffer << sp << servbuffer << sp << "typ" << sp << type;
-						oss << left;
-						mCandidate = oss.str();
+	struct addrinfo *result = nullptr;
+	if (getaddrinfo(mNode.c_str(), mService.c_str(), &hints, &result) == 0) {
+		for (auto p = result; p; p = p->ai_next) {
+			if (p->ai_family == AF_INET || p->ai_family == AF_INET6) {
+				char nodebuffer[MAX_NUMERICNODE_LEN];
+				char servbuffer[MAX_NUMERICSERV_LEN];
+				if (getnameinfo(p->ai_addr, socklen_t(p->ai_addrlen), nodebuffer,
+				                MAX_NUMERICNODE_LEN, servbuffer, MAX_NUMERICSERV_LEN,
+				                NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
 
-						PLOG_VERBOSE << "Resolved candidate: " << mCandidate;
-						break;
-					}
+					mAddress = nodebuffer;
+					mPort = uint16_t(std::stoul(servbuffer));
+					mFamily = p->ai_family == AF_INET6 ? Family::Ipv6 : Family::Ipv4;
+					PLOG_VERBOSE << "Resolved candidate: " << mAddress << ' ' << mPort;
+					break;
 				}
 			}
-
-			freeaddrinfo(result);
 		}
+
+		freeaddrinfo(result);
 	}
 
 	return mFamily != Family::Unresolved;
 }
 
-string Candidate::candidate() const { return "candidate:" + mCandidate; }
+Candidate::Type Candidate::type() const { return mType; }
+
+Candidate::TransportType Candidate::transportType() const { return mTransportType; }
+
+uint32_t Candidate::priority() const { return mPriority; }
+
+string Candidate::candidate() const {
+	const char sp{' '};
+	std::ostringstream oss;
+	oss << "candidate:";
+	oss << mFoundation << sp << mComponent << sp << mTransportString << sp << mPriority << sp;
+	if (isResolved())
+		oss << mAddress << sp << mPort;
+	else
+		oss << mNode << sp << mService;
+
+	oss << sp << "typ" << sp << mTypeString;
+
+	if (!mTail.empty())
+		oss << sp << mTail;
+
+	return oss.str();
+}
 
 string Candidate::mid() const { return mMid.value_or("0"); }
 
@@ -186,14 +215,18 @@ Candidate::operator string() const {
 	return line.str();
 }
 
+bool Candidate::operator==(const Candidate &other) const {
+	return mFoundation == other.mFoundation;
+}
+
+bool Candidate::operator!=(const Candidate &other) const {
+	return mFoundation != other.mFoundation;
+}
+
 bool Candidate::isResolved() const { return mFamily != Family::Unresolved; }
 
 Candidate::Family Candidate::family() const { return mFamily; }
 
-Candidate::Type Candidate::type() const { return mType; }
-
-Candidate::TransportType Candidate::transportType() const { return mTransportType; }
-
 std::optional<string> Candidate::address() const {
 	return isResolved() ? std::make_optional(mAddress) : nullopt;
 }
@@ -202,10 +235,6 @@ std::optional<uint16_t> Candidate::port() const {
 	return isResolved() ? std::make_optional(mPort) : nullopt;
 }
 
-std::optional<uint32_t> Candidate::priority() const {
-	return isResolved() ? std::make_optional(mPriority) : nullopt;
-}
-
 } // namespace rtc
 
 std::ostream &operator<<(std::ostream &out, const rtc::Candidate &candidate) {
@@ -217,11 +246,11 @@ std::ostream &operator<<(std::ostream &out, const rtc::Candidate::Type &type) {
 	case rtc::Candidate::Type::Host:
 		return out << "host";
 	case rtc::Candidate::Type::PeerReflexive:
-		return out << "peer_reflexive";
+		return out << "prflx";
 	case rtc::Candidate::Type::ServerReflexive:
-		return out << "server_reflexive";
+		return out << "srflx";
 	case rtc::Candidate::Type::Relayed:
-		return out << "relayed";
+		return out << "relay";
 	default:
 		return out << "unknown";
 	}

src/datachannel.cpp

@@ -306,7 +306,7 @@ void NegociatedDataChannel::processOpenMessage(message_ptr message) {
 	mLabel.assign(end, open.labelLength);
 	mProtocol.assign(end + open.labelLength, open.protocolLength);
 
-	mReliability->unordered = (open.reliabilityParameter & 0x80) != 0;
+	mReliability->unordered = (open.channelType & 0x80) != 0;
 	switch (open.channelType & 0x7F) {
 	case CHANNEL_PARTIAL_RELIABLE_REXMIT:
 		mReliability->type = Reliability::Type::Rexmit;

src/description.cpp

@@ -106,7 +106,7 @@ Description::Description(const string &sdp, Type type, Role role)
 					               mFingerprint->begin(),
 					               [](char c) { return char(std::toupper(c)); });
 				} else {
-					PLOG_WARNING << "Unknown SDP fingerprint type: " << value;
+					PLOG_WARNING << "Unknown SDP fingerprint format: " << value;
 				}
 			} else if (key == "ice-ufrag") {
 				mIceUfrag = value;
@@ -171,16 +171,27 @@ void Description::setFingerprint(string fingerprint) {
 	mFingerprint.emplace(std::move(fingerprint));
 }
 
+bool Description::hasCandidate(const Candidate &candidate) const {
+	for (const Candidate &other : mCandidates)
+		if (candidate == other)
+			return true;
+
+	return false;
+}
+
 void Description::addCandidate(Candidate candidate) {
 	candidate.hintMid(bundleMid());
+
+	for (const Candidate &other : mCandidates)
+		if (candidate == other)
+			return;
+
 	mCandidates.emplace_back(std::move(candidate));
 }
 
 void Description::addCandidates(std::vector<Candidate> candidates) {
-	for (Candidate candidate : candidates) {
-		candidate.hintMid(bundleMid());
-		mCandidates.emplace_back(std::move(candidate));
-	}
+	for (Candidate candidate : candidates)
+		addCandidate(std::move(candidate));
 }
 
 void Description::endCandidates() { mEnded = true; }

src/icetransport.cpp

@@ -46,11 +46,12 @@ using std::chrono::system_clock;
 
 namespace rtc {
 
-IceTransport::IceTransport(const Configuration &config, Description::Role role,
-                           candidate_callback candidateCallback, state_callback stateChangeCallback,
+IceTransport::IceTransport(const Configuration &config, candidate_callback candidateCallback,
+                           state_callback stateChangeCallback,
                            gathering_state_callback gatheringStateChangeCallback)
-    : Transport(nullptr, std::move(stateChangeCallback)), mRole(role), mMid("0"),
-      mGatheringState(GatheringState::New), mCandidateCallback(std::move(candidateCallback)),
+    : Transport(nullptr, std::move(stateChangeCallback)), mRole(Description::Role::ActPass),
+      mMid("0"), mGatheringState(GatheringState::New),
+      mCandidateCallback(std::move(candidateCallback)),
       mGatheringStateChangeCallback(std::move(gatheringStateChangeCallback)),
       mAgent(nullptr, nullptr) {
 
@@ -139,13 +140,19 @@ Description IceTransport::getLocalDescription(Description::Type type) const {
 	if (juice_get_local_description(mAgent.get(), sdp, JUICE_MAX_SDP_STRING_LEN) < 0)
 		throw std::runtime_error("Failed to generate local SDP");
 
+	// RFC 5763: The endpoint that is the offerer MUST use the setup attribute value of
+	// setup:actpass.
+	// See https://tools.ietf.org/html/rfc5763#section-5
 	return Description(string(sdp), type,
 	                   type == Description::Type::Offer ? Description::Role::ActPass : mRole);
 }
 
 void IceTransport::setRemoteDescription(const Description &description) {
-	mRole = description.role() == Description::Role::Active ? Description::Role::Passive
-	                                                        : Description::Role::Active;
+	if (mRole == Description::Role::ActPass)
+		mRole = description.role() == Description::Role::Active ? Description::Role::Passive
+		                                                        : Description::Role::Active;
+	if (mRole == description.role())
+		throw std::logic_error("Incompatible roles with remote description");
 
 	mMid = description.bundleMid();
 	if (juice_set_remote_description(mAgent.get(),
@@ -316,11 +323,12 @@ void IceTransport::LogCallback(juice_log_level_t level, const char *message) {
 
 namespace rtc {
 
-IceTransport::IceTransport(const Configuration &config, Description::Role role,
-                           candidate_callback candidateCallback, state_callback stateChangeCallback,
+IceTransport::IceTransport(const Configuration &config, candidate_callback candidateCallback,
+                           state_callback stateChangeCallback,
                            gathering_state_callback gatheringStateChangeCallback)
-    : Transport(nullptr, std::move(stateChangeCallback)), mRole(role), mMid("0"),
-      mGatheringState(GatheringState::New), mCandidateCallback(std::move(candidateCallback)),
+    : Transport(nullptr, std::move(stateChangeCallback)), mRole(Description::Role::ActPass),
+      mMid("0"), mGatheringState(GatheringState::New),
+      mCandidateCallback(std::move(candidateCallback)),
       mGatheringStateChangeCallback(std::move(gatheringStateChangeCallback)),
       mNiceAgent(nullptr, nullptr), mMainLoop(nullptr, nullptr) {
 
@@ -526,12 +534,21 @@ Description IceTransport::getLocalDescription(Description::Type type) const {
 
 	std::unique_ptr<gchar[], void (*)(void *)> sdp(nice_agent_generate_local_sdp(mNiceAgent.get()),
 	                                               g_free);
-	return Description(string(sdp.get()), type, mRole);
+
+	// RFC 5763: The endpoint that is the offerer MUST use the setup attribute value of
+	// setup:actpass.
+	// See https://tools.ietf.org/html/rfc5763#section-5
+	return Description(string(sdp.get()), type,
+	                   type == Description::Type::Offer ? Description::Role::ActPass : mRole);
 }
 
 void IceTransport::setRemoteDescription(const Description &description) {
-	mRole = description.role() == Description::Role::Active ? Description::Role::Passive
-	                                                        : Description::Role::Active;
+	if (mRole == Description::Role::ActPass)
+		mRole = description.role() == Description::Role::Active ? Description::Role::Passive
+		                                                        : Description::Role::Active;
+	if (mRole == description.role())
+		throw std::logic_error("Incompatible roles with remote description");
+
 	mMid = description.bundleMid();
 	mTrickleTimeout = !description.ended() ? 30s : 0s;
 
@@ -547,12 +564,14 @@ bool IceTransport::addRemoteCandidate(const Candidate &candidate) {
 		return false;
 
 	// Warning: the candidate string must start with "a=candidate:" and it must not end with a
-	// newline, else libnice will reject it.
+	// newline or whitespace, else libnice will reject it.
 	string sdp(candidate);
 	NiceCandidate *cand =
 	    nice_agent_parse_remote_candidate_sdp(mNiceAgent.get(), mStreamId, sdp.c_str());
-	if (!cand)
+	if (!cand) {
+		PLOG_WARNING << "Rejected ICE candidate: " << sdp;
 		return false;
+	}
 
 	GSList *list = g_slist_append(nullptr, cand);
 	int ret = nice_agent_set_remote_candidates(mNiceAgent.get(), mStreamId, 1, list);

src/icetransport.hpp

@@ -45,8 +45,8 @@ public:
 	using candidate_callback = std::function<void(const Candidate &candidate)>;
 	using gathering_state_callback = std::function<void(GatheringState state)>;
 
-	IceTransport(const Configuration &config, Description::Role role,
-	             candidate_callback candidateCallback, state_callback stateChangeCallback,
+	IceTransport(const Configuration &config, candidate_callback candidateCallback,
+	             state_callback stateChangeCallback,
 	             gathering_state_callback gatheringStateChangeCallback);
 	~IceTransport();
 

src/peerconnection.cpp

@@ -77,7 +77,7 @@ void PeerConnection::close() {
 	mNegotiationNeeded = false;
 
 	// Close data channels asynchronously
-	mProcessor->enqueue(std::bind(&PeerConnection::closeDataChannels, this));
+	mProcessor->enqueue(&PeerConnection::closeDataChannels, this);
 
 	closeTransports();
 }
@@ -187,13 +187,7 @@ void PeerConnection::setLocalDescription(Description::Type type) {
 	}
 	}
 
-	auto iceTransport = std::atomic_load(&mIceTransport);
-	if (!iceTransport) {
-		// RFC 5763: The endpoint that is the offerer MUST use the setup attribute value of
-		// setup:actpass.
-		// See https://tools.ietf.org/html/rfc5763#section-5
-		iceTransport = initIceTransport(Description::Role::ActPass);
-	}
+	auto iceTransport = initIceTransport();
 
 	Description localDescription = iceTransport->getLocalDescription(type);
 	processLocalDescription(std::move(localDescription));
@@ -272,12 +266,6 @@ void PeerConnection::setRemoteDescription(Description description) {
 	// Candidates will be added at the end, extract them for now
 	auto remoteCandidates = description.extractCandidates();
 	auto type = description.type();
-
-	auto iceTransport = std::atomic_load(&mIceTransport);
-	if (!iceTransport)
-		iceTransport = initIceTransport(Description::Role::ActPass);
-
-	iceTransport->setRemoteDescription(description);
 	processRemoteDescription(std::move(description));
 
 	changeSignalingState(newSignalingState);
@@ -287,8 +275,9 @@ void PeerConnection::setRemoteDescription(Description description) {
 		setLocalDescription(Description::Type::Answer);
 	} else {
 		// This is an answer
+		auto iceTransport = std::atomic_load(&mIceTransport);
 		auto sctpTransport = std::atomic_load(&mSctpTransport);
-		if (!sctpTransport && iceTransport->role() == Description::Role::Active) {
+		if (!sctpTransport && iceTransport && 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); // we are going to swap the container
@@ -405,14 +394,14 @@ void PeerConnection::onTrack(std::function<void(std::shared_ptr<Track>)> callbac
 	mTrackCallback = callback;
 }
 
-shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role) {
+shared_ptr<IceTransport> PeerConnection::initIceTransport() {
 	PLOG_VERBOSE << "Starting ICE transport";
 	try {
 		if (auto transport = std::atomic_load(&mIceTransport))
 			return transport;
 
 		auto transport = std::make_shared<IceTransport>(
-		    mConfig, role, weak_bind(&PeerConnection::processLocalCandidate, this, _1),
+		    mConfig, weak_bind(&PeerConnection::processLocalCandidate, this, _1),
 		    [this, weak_this = weak_from_this()](IceTransport::State state) {
 			    auto shared_this = weak_this.lock();
 			    if (!shared_this)
@@ -490,7 +479,7 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 				else
 					changeState(State::Connected);
 
-				mProcessor->enqueue(std::bind(&PeerConnection::openTracks, this));
+				mProcessor->enqueue(&PeerConnection::openTracks, this);
 				break;
 			case DtlsTransport::State::Failed:
 				changeState(State::Failed);
@@ -561,16 +550,16 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 			    switch (state) {
 			    case SctpTransport::State::Connected:
 				    changeState(State::Connected);
-				    mProcessor->enqueue(std::bind(&PeerConnection::openDataChannels, this));
+				    mProcessor->enqueue(&PeerConnection::openDataChannels, this);
 				    break;
 			    case SctpTransport::State::Failed:
 				    LOG_WARNING << "SCTP transport failed";
 				    changeState(State::Failed);
-				    mProcessor->enqueue(std::bind(&PeerConnection::remoteCloseDataChannels, this));
+				    mProcessor->enqueue(&PeerConnection::remoteCloseDataChannels, this);
 				    break;
 			    case SctpTransport::State::Disconnected:
 				    changeState(State::Disconnected);
-				    mProcessor->enqueue(std::bind(&PeerConnection::remoteCloseDataChannels, this));
+				    mProcessor->enqueue(&PeerConnection::remoteCloseDataChannels, this);
 				    break;
 			    default:
 				    // Ignore
@@ -1069,19 +1058,17 @@ void PeerConnection::processLocalDescription(Description description) {
 			mCurrentLocalDescription.emplace(std::move(*mLocalDescription));
 		}
 
-		mLocalDescription.emplace(std::move(description));
+		mLocalDescription.emplace(description);
 		mLocalDescription->addCandidates(std::move(existingCandidates));
 	}
 
-	mProcessor->enqueue([this, description = *mLocalDescription]() {
-		PLOG_VERBOSE << "Issuing local description: " << description;
-		mLocalDescriptionCallback(std::move(description));
-	});
+	PLOG_VERBOSE << "Issuing local description: " << description;
+	mProcessor->enqueue(mLocalDescriptionCallback.wrap(), std::move(description));
 
 	// Reciprocated tracks might need to be open
 	if (auto dtlsTransport = std::atomic_load(&mDtlsTransport);
 	    dtlsTransport && dtlsTransport->state() == Transport::State::Connected)
-		mProcessor->enqueue(std::bind(&PeerConnection::openTracks, this));
+		mProcessor->enqueue(&PeerConnection::openTracks, this);
 }
 
 void PeerConnection::processLocalCandidate(Candidate candidate) {
@@ -1089,13 +1076,11 @@ void PeerConnection::processLocalCandidate(Candidate candidate) {
 	if (!mLocalDescription)
 		throw std::logic_error("Got a local candidate without local description");
 
-	candidate.resolve(Candidate::ResolveMode::Simple); // for proper SDP generation later
+	candidate.resolve(Candidate::ResolveMode::Simple);
 	mLocalDescription->addCandidate(candidate);
 
-	mProcessor->enqueue([this, candidate = std::move(candidate)]() {
-		PLOG_VERBOSE << "Issuing local candidate: " << candidate;
-		mLocalCandidateCallback(std::move(candidate));
-	});
+	PLOG_VERBOSE << "Issuing local candidate: " << candidate;
+	mProcessor->enqueue(mLocalCandidateCallback.wrap(), std::move(candidate));
 }
 
 void PeerConnection::processRemoteDescription(Description description) {
@@ -1107,10 +1092,13 @@ void PeerConnection::processRemoteDescription(Description description) {
 		if (mRemoteDescription)
 			existingCandidates = mRemoteDescription->extractCandidates();
 
-		mRemoteDescription.emplace(std::move(description));
+		mRemoteDescription.emplace(description);
 		mRemoteDescription->addCandidates(std::move(existingCandidates));
 	}
 
+	auto iceTransport = initIceTransport();
+	iceTransport->setRemoteDescription(std::move(description));
+
 	if (description.hasApplication()) {
 		auto dtlsTransport = std::atomic_load(&mDtlsTransport);
 		auto sctpTransport = std::atomic_load(&mSctpTransport);
@@ -1121,28 +1109,40 @@ void PeerConnection::processRemoteDescription(Description description) {
 }
 
 void PeerConnection::processRemoteCandidate(Candidate candidate) {
-	std::lock_guard lock(mRemoteDescriptionMutex);
 	auto iceTransport = std::atomic_load(&mIceTransport);
-	if (!mRemoteDescription || !iceTransport)
-		throw std::logic_error("Got a remote candidate without remote description");
+	{
+		// Set as remote candidate
+		std::lock_guard lock(mRemoteDescriptionMutex);
+		if (!mRemoteDescription)
+			throw std::logic_error("Got a remote candidate without remote description");
 
-	candidate.hintMid(mRemoteDescription->bundleMid());
+		if (!iceTransport)
+			throw std::logic_error("Got a remote candidate without ICE transport");
 
-	if (candidate.resolve(Candidate::ResolveMode::Simple)) {
-		iceTransport->addRemoteCandidate(candidate);
-	} else {
-		// OK, we might need a lookup, do it asynchronously
-		// We don't use the thread pool because we have no control on the timeout
-		weak_ptr<IceTransport> weakIceTransport{iceTransport};
-		std::thread t([weakIceTransport, candidate]() mutable {
-			if (candidate.resolve(Candidate::ResolveMode::Lookup))
-				if (auto iceTransport = weakIceTransport.lock())
-					iceTransport->addRemoteCandidate(candidate);
-		});
-		t.detach();
+		candidate.hintMid(mRemoteDescription->bundleMid());
+
+		if (mRemoteDescription->hasCandidate(candidate))
+			return; // already in description, ignore
+
+		candidate.resolve(Candidate::ResolveMode::Simple);
+		mRemoteDescription->addCandidate(candidate);
 	}
 
-	mRemoteDescription->addCandidate(std::move(candidate));
+	if (candidate.isResolved()) {
+		iceTransport->addRemoteCandidate(std::move(candidate));
+	} else {
+		// We might need a lookup, do it asynchronously
+		// We don't use the thread pool because we have no control on the timeout
+		if (auto iceTransport = std::atomic_load(&mIceTransport)) {
+			weak_ptr<IceTransport> weakIceTransport{iceTransport};
+			std::thread t([weakIceTransport, candidate = std::move(candidate)]() mutable {
+				if (candidate.resolve(Candidate::ResolveMode::Lookup))
+					if (auto iceTransport = weakIceTransport.lock())
+						iceTransport->addRemoteCandidate(std::move(candidate));
+			});
+			t.detach();
+		}
+	}
 }
 
 void PeerConnection::triggerDataChannel(weak_ptr<DataChannel> weakDataChannel) {
@@ -1150,12 +1150,11 @@ void PeerConnection::triggerDataChannel(weak_ptr<DataChannel> weakDataChannel) {
 	if (!dataChannel)
 		return;
 
-	mProcessor->enqueue(
-	    [this, dataChannel = std::move(dataChannel)]() { mDataChannelCallback(dataChannel); });
+	mProcessor->enqueue(mDataChannelCallback.wrap(), std::move(dataChannel));
 }
 
 void PeerConnection::triggerTrack(std::shared_ptr<Track> track) {
-	mProcessor->enqueue([this, track = std::move(track)]() { mTrackCallback(track); });
+	mProcessor->enqueue(mTrackCallback.wrap(), std::move(track));
 }
 
 bool PeerConnection::changeState(State state) {
@@ -1177,7 +1176,7 @@ bool PeerConnection::changeState(State state) {
 		// This is the last state change, so we may steal the callback
 		mProcessor->enqueue([cb = std::move(mStateChangeCallback)]() { cb(State::Closed); });
 	else
-		mProcessor->enqueue([this, state]() { mStateChangeCallback(state); });
+		mProcessor->enqueue(mStateChangeCallback.wrap(), state);
 
 	return true;
 }
@@ -1189,7 +1188,7 @@ bool PeerConnection::changeGatheringState(GatheringState state) {
 	std::ostringstream s;
 	s << state;
 	PLOG_INFO << "Changed gathering state to " << s.str();
-	mProcessor->enqueue([this, state] { mGatheringStateChangeCallback(state); });
+	mProcessor->enqueue(mGatheringStateChangeCallback.wrap(), state);
 	return true;
 }
 
@@ -1200,7 +1199,7 @@ bool PeerConnection::changeSignalingState(SignalingState state) {
 	std::ostringstream s;
 	s << state;
 	PLOG_INFO << "Changed signaling state to " << s.str();
-	mProcessor->enqueue([this, state] { mSignalingStateChangeCallback(state); });
+	mProcessor->enqueue(mSignalingStateChangeCallback.wrap(), state);
 	return true;
 }
 

src/processor.cpp

@@ -20,6 +20,8 @@
 
 namespace rtc {
 
+Processor::Processor(size_t limit) : mTasks(limit) {}
+
 Processor::~Processor() { join(); }
 
 void Processor::join() {
@@ -29,15 +31,13 @@ void Processor::join() {
 
 void Processor::schedule() {
 	std::unique_lock lock(mMutex);
-	if (mTasks.empty()) {
+	if (auto next = mTasks.tryPop()) {
+		ThreadPool::Instance().enqueue(std::move(*next));
+	} else {
 		// No more tasks
 		mPending = false;
 		mCondition.notify_all();
-		return;
 	}
-
-	ThreadPool::Instance().enqueue(std::move(mTasks.front()));
-	mTasks.pop();
 }
 
 } // namespace rtc

src/processor.hpp

@@ -22,6 +22,7 @@
 #include "include.hpp"
 #include "init.hpp"
 #include "threadpool.hpp"
+#include "queue.hpp"
 
 #include <condition_variable>
 #include <future>
@@ -34,7 +35,7 @@ namespace rtc {
 // Processed tasks in order by delegating them to the thread pool
 class Processor final {
 public:
-	Processor() = default;
+	Processor(size_t limit = 0);
 	~Processor();
 
 	Processor(const Processor &) = delete;
@@ -52,7 +53,7 @@ protected:
 	// Keep an init token
 	const init_token mInitToken = Init::Token();
 
-	std::queue<std::function<void()>> mTasks;
+	Queue<std::function<void()>> mTasks;
 	bool mPending = false; // true iff a task is pending in the thread pool
 
 	mutable std::mutex mMutex;
@@ -71,7 +72,7 @@ template <class F, class... Args> void Processor::enqueue(F &&f, Args &&... args
 		ThreadPool::Instance().enqueue(std::move(task));
 		mPending = true;
 	} else {
-		mTasks.emplace(std::move(task));
+		mTasks.push(std::move(task));
 	}
 }
 

src/sctptransport.cpp

@@ -88,7 +88,7 @@ void SctpTransport::Cleanup() {
 SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
                              message_callback recvCallback, amount_callback bufferedAmountCallback,
                              state_callback stateChangeCallback)
-    : Transport(lower, std::move(stateChangeCallback)), mPort(port),
+    : Transport(lower, std::move(stateChangeCallback)), mPort(port), mPendingRecvCount(0),
       mSendQueue(0, message_size_func), mBufferedAmountCallback(std::move(bufferedAmountCallback)) {
 	onRecv(recvCallback);
 
@@ -100,11 +100,12 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 		Instances.insert(this);
 	}
 
-	mSock = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP, &SctpTransport::RecvCallback,
-	                       &SctpTransport::SendCallback, 0, this);
+	mSock = usrsctp_socket(AF_CONN, SOCK_STREAM, IPPROTO_SCTP, nullptr, nullptr, 0, nullptr);
 	if (!mSock)
 		throw std::runtime_error("Could not create SCTP socket, errno=" + std::to_string(errno));
 
+	usrsctp_set_upcall(mSock, &SctpTransport::UpcallCallback, this);
+
 	if (usrsctp_set_non_blocking(mSock, 1))
 		throw std::runtime_error("Unable to set non-blocking mode, errno=" + std::to_string(errno));
 
@@ -122,6 +123,10 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 	if (usrsctp_setsockopt(mSock, IPPROTO_SCTP, SCTP_ENABLE_STREAM_RESET, &av, sizeof(av)))
 		throw std::runtime_error("Could not set socket option SCTP_ENABLE_STREAM_RESET, errno=" +
 		                         std::to_string(errno));
+	int on = 1;
+	if (usrsctp_setsockopt(mSock, IPPROTO_SCTP, SCTP_RECVRCVINFO, &on, sizeof(on)))
+		throw std::runtime_error("Could set socket option SCTP_RECVRCVINFO, errno=" +
+		                         std::to_string(errno));
 
 	struct sctp_event se = {};
 	se.se_assoc_id = SCTP_ALL_ASSOC;
@@ -225,6 +230,7 @@ bool SctpTransport::stop() {
 
 void SctpTransport::close() {
 	if (mSock) {
+		mProcessor.join();
 		usrsctp_close(mSock);
 		mSock = nullptr;
 	}
@@ -319,6 +325,60 @@ void SctpTransport::incoming(message_ptr message) {
 	usrsctp_conninput(this, message->data(), message->size(), 0);
 }
 
+void SctpTransport::doRecv() {
+	std::lock_guard lock(mRecvMutex);
+	--mPendingRecvCount;
+	try {
+		while (true) {
+			const size_t bufferSize = 65536;
+			byte buffer[bufferSize];
+			socklen_t fromlen = 0;
+			struct sctp_rcvinfo info = {};
+			socklen_t infolen = sizeof(info);
+			unsigned int infotype = 0;
+			int flags = 0;
+			ssize_t len = usrsctp_recvv(mSock, buffer, bufferSize, nullptr, &fromlen, &info,
+			                            &infolen, &infotype, &flags);
+			if (len < 0) {
+				if (errno == EWOULDBLOCK || errno == EAGAIN || errno == ECONNRESET)
+					break;
+				else
+					throw std::runtime_error("SCTP recv failed, errno=" + std::to_string(errno));
+			}
+
+			PLOG_VERBOSE << "SCTP recv, len=" << len;
+
+			// SCTP_FRAGMENT_INTERLEAVE does not seem to work as expected for messages > 64KB,
+			// therefore partial notifications and messages need to be handled separately.
+			if (flags & MSG_NOTIFICATION) {
+				// SCTP event notification
+				mPartialNotification.insert(mPartialNotification.end(), buffer, buffer + len);
+				if (flags & MSG_EOR) {
+					// Notification is complete, process it
+					auto notification =
+					    reinterpret_cast<union sctp_notification *>(mPartialNotification.data());
+					processNotification(notification, mPartialNotification.size());
+					mPartialNotification.clear();
+				}
+			} else {
+				// SCTP message
+				mPartialMessage.insert(mPartialMessage.end(), buffer, buffer + len);
+				if (flags & MSG_EOR) {
+					// Message is complete, process it
+					if (infotype != SCTP_RECVV_RCVINFO)
+						throw std::runtime_error("Missing SCTP recv info");
+
+					processData(std::move(mPartialMessage), info.rcv_sid,
+					            PayloadId(ntohl(info.rcv_ppid)));
+					mPartialMessage.clear();
+				}
+			}
+		}
+	} catch (const std::exception &e) {
+		PLOG_WARNING << e.what();
+	}
+}
+
 bool SctpTransport::trySendQueue() {
 	// Requires mSendMutex to be locked
 	while (auto next = mSendQueue.peek()) {
@@ -472,44 +532,21 @@ bool SctpTransport::safeFlush() {
 	}
 }
 
-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;
+void SctpTransport::handleUpcall() {
+	if (!mSock)
+		return;
 
-		// SCTP_FRAGMENT_INTERLEAVE does not seem to work as expected for messages > 64KB,
-		// 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) {
-				// Notification is complete, process it
-				processNotification(
-				    reinterpret_cast<const union sctp_notification *>(mPartialNotification.data()),
-				    mPartialNotification.size());
-				mPartialNotification.clear();
-			}
-		} else {
-			// SCTP message
-			mPartialMessage.insert(mPartialMessage.end(), data, data + len);
-			if (flags & MSG_EOR) {
-				// Message is complete, process it
-				processData(std::move(mPartialMessage), info.rcv_sid,
-				            PayloadId(ntohl(info.rcv_ppid)));
-				mPartialMessage.clear();
-			}
-		}
+	PLOG_VERBOSE << "Handle upcall";
 
-	} catch (const std::exception &e) {
-		PLOG_ERROR << "SCTP recv: " << e.what();
-		return -1;
+	int events = usrsctp_get_events(mSock);
+
+	if (events & SCTP_EVENT_READ && mPendingRecvCount == 0) {
+		++mPendingRecvCount;
+		mProcessor.enqueue(&SctpTransport::doRecv, this);
 	}
-	return 0; // success
-}
 
-int SctpTransport::handleSend(size_t free) {
-	PLOG_VERBOSE << "Handle send, free=" << free;
-	return safeFlush() ? 0 : -1;
+	if (events & SCTP_EVENT_WRITE)
+		mProcessor.enqueue(&SctpTransport::safeFlush, this);
 }
 
 int SctpTransport::handleWrite(byte *data, size_t len, uint8_t /*tos*/, uint8_t /*set_df*/) {
@@ -699,31 +736,14 @@ std::optional<milliseconds> SctpTransport::rtt() {
 	return milliseconds(status.sstat_primary.spinfo_srtt);
 }
 
-int SctpTransport::RecvCallback(struct socket *sock, union sctp_sockstore addr, void *data,
-                                size_t len, struct sctp_rcvinfo recv_info, int flags,
-                                void *ulp_info) {
-	auto *transport = static_cast<SctpTransport *>(ulp_info);
-
-	std::shared_lock lock(InstancesMutex);
-	if (Instances.find(transport) == Instances.end()) {
-		free(data);
-		return -1;
-	}
-
-	int ret =
-	    transport->handleRecv(sock, addr, static_cast<const byte *>(data), len, recv_info, flags);
-	free(data);
-	return ret;
-}
-
-int SctpTransport::SendCallback(struct socket *, uint32_t sb_free, void *ulp_info) {
-	auto *transport = static_cast<SctpTransport *>(ulp_info);
+void SctpTransport::UpcallCallback(struct socket *, void *arg, int /* flags */) {
+	auto *transport = static_cast<SctpTransport *>(arg);
 
 	std::shared_lock lock(InstancesMutex);
 	if (Instances.find(transport) == Instances.end())
-		return -1;
+		return;
 
-	return transport->handleSend(size_t(sb_free));
+	transport->handleUpcall();
 }
 
 int SctpTransport::WriteCallback(void *ptr, void *data, size_t len, uint8_t tos, uint8_t set_df) {

src/sctptransport.hpp

@@ -21,8 +21,10 @@
 
 #include "include.hpp"
 #include "peerconnection.hpp"
+#include "processor.hpp"
 #include "queue.hpp"
 #include "transport.hpp"
+#include "processor.hpp"
 
 #include <condition_variable>
 #include <functional>
@@ -35,7 +37,7 @@
 
 namespace rtc {
 
-class SctpTransport : public Transport {
+class SctpTransport final : public Transport {
 public:
 	static void Init();
 	static void Cleanup();
@@ -76,15 +78,14 @@ private:
 	void close();
 	void incoming(message_ptr message) override;
 
+	void doRecv();
 	bool trySendQueue();
 	bool trySendMessage(message_ptr message);
 	void updateBufferedAmount(uint16_t streamId, long delta);
 	void sendReset(uint16_t streamId);
 	bool safeFlush();
 
-	int handleRecv(struct socket *sock, union sctp_sockstore addr, const byte *data, size_t len,
-	               struct sctp_rcvinfo recv_info, int flags);
-	int handleSend(size_t free);
+	void handleUpcall();
 	int handleWrite(byte *data, size_t len, uint8_t tos, uint8_t set_df);
 
 	void processData(binary &&data, uint16_t streamId, PayloadId ppid);
@@ -93,7 +94,9 @@ private:
 	const uint16_t mPort;
 	struct socket *mSock;
 
-	std::mutex mSendMutex;
+	Processor mProcessor;
+	std::atomic<int> mPendingRecvCount;
+	std::mutex mRecvMutex, mSendMutex;
 	Queue<message_ptr> mSendQueue;
 	std::map<uint16_t, size_t> mBufferedAmount;
 	amount_callback mBufferedAmountCallback;
@@ -109,9 +112,7 @@ private:
 	// Stats
 	std::atomic<size_t> mBytesSent = 0, mBytesReceived = 0;
 
-	static int RecvCallback(struct socket *sock, union sctp_sockstore addr, void *data, size_t len,
-	                        struct sctp_rcvinfo recv_info, int flags, void *ulp_info);
-	static int SendCallback(struct socket *sock, uint32_t sb_free, void *ulp_info);
+	static void UpcallCallback(struct socket *sock, void *arg, int flags);
 	static int WriteCallback(void *sctp_ptr, void *data, size_t len, uint8_t tos, uint8_t set_df);
 
 	static std::unordered_set<SctpTransport *> Instances;

test/capi_connectivity.cpp

@@ -100,14 +100,48 @@ static void RTC_API messageCallback(int id, const char *message, int size, void
 
 static void RTC_API dataChannelCallback(int pc, int dc, void *ptr) {
 	Peer *peer = (Peer *)ptr;
-	peer->dc = dc;
-	peer->connected = true;
+
+	char label[256];
+	if (rtcGetDataChannelLabel(dc, label, 256) < 0) {
+		fprintf(stderr, "rtcGetDataChannelLabel failed\n");
+		return;
+	}
+
+	char protocol[256];
+	if (rtcGetDataChannelProtocol(dc, protocol, 256) < 0) {
+		fprintf(stderr, "rtcGetDataChannelProtocol failed\n");
+		return;
+	}
+
+	rtcReliability reliability;
+	if (rtcGetDataChannelReliability(dc, &reliability) < 0) {
+		fprintf(stderr, "rtcGetDataChannelReliability failed\n");
+		return;
+	}
+
+	printf("DataChannel %d: Received with label \"%s\" and protocol \"%s\"\n",
+	       peer == peer1 ? 1 : 2, label, protocol);
+
+	if (strcmp(label, "test") != 0) {
+		fprintf(stderr, "Wrong DataChannel label\n");
+		return;
+	}
+
+	if (strcmp(protocol, "protocol") != 0) {
+		fprintf(stderr, "Wrong DataChannel protocol\n");
+		return;
+	}
+
+	if (reliability.unordered == false) {
+		fprintf(stderr, "Wrong DataChannel reliability\n");
+		return;
+	}
+
 	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);
+	peer->dc = dc;
+	peer->connected = true;
 
 	const char *message = peer == peer1 ? "Hello from 1" : "Hello from 2";
 	rtcSendMessage(peer->dc, message, -1); // negative size indicates a null-terminated string
@@ -173,7 +207,12 @@ int test_capi_connectivity_main() {
 		goto error;
 
 	// Peer 1: Create data channel
-	peer1->dc = rtcCreateDataChannel(peer1->pc, "test");
+	rtcDataChannelInit init;
+	memset(&init, 0, sizeof(init));
+	init.protocol = "protocol";
+	init.reliability.unordered = true;
+
+	peer1->dc = rtcCreateDataChannelEx(peer1->pc, "test", &init);
 	rtcSetOpenCallback(peer1->dc, openCallback);
 	rtcSetClosedCallback(peer1->dc, closedCallback);
 	rtcSetMessageCallback(peer1->dc, messageCallback);