Added .travis.yml

Manually handle OpenSSL handshake timeout

.travis.yml

@@ -0,0 +1,4 @@
+os: osx
+osx_image: xcode11.3
+language: cpp
+script: cmake -B build -DUSE_JUICE=1 -DUSE_GNUTLS=1 && cd build && make && ./tests

CMakeLists.txt

@@ -17,6 +17,7 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 set(CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/cmake/Modules)
 
 if(WIN32)
+	add_definitions(-DWIN32_LEAN_AND_MEAN)
 	if (MSYS OR MINGW)
 		add_definitions(-DSCTP_STDINT_INCLUDE=<stdint.h>)
 	endif()

include/rtc/candidate.hpp

@@ -25,6 +25,15 @@
 
 namespace rtc {
 
+enum class CandidateType { Host = 0, ServerReflexive, PeerReflexive, Relayed };
+enum class CandidateTransportType { Udp = 0, TcpActive, TcpPassive, TcpSo };
+struct CandidateInfo {
+	string address;
+	int port;
+	CandidateType type;
+	CandidateTransportType transportType;
+};
+
 class Candidate {
 public:
 	Candidate(string candidate, string mid = "");
@@ -46,6 +55,8 @@ private:
 } // namespace rtc
 
 std::ostream &operator<<(std::ostream &out, const rtc::Candidate &candidate);
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateType &type);
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateTransportType &transportType);
 
 #endif
 

include/rtc/channel.hpp

@@ -60,6 +60,8 @@ protected:
 	virtual void triggerAvailable(size_t count);
 	virtual void triggerBufferedAmount(size_t amount);
 
+	void resetCallbacks();
+
 private:
 	synchronized_callback<> mOpenCallback;
 	synchronized_callback<> mClosedCallback;

include/rtc/configuration.hpp

@@ -51,8 +51,22 @@ struct IceServer {
 	RelayType relayType;
 };
 
+struct ProxyServer {
+	enum class Type { None = 0, Socks5, Http, Last = Http };
+
+	ProxyServer(Type type_, string ip_, uint16_t port_, string username_ = "",
+	            string password_ = "");
+
+	Type type;
+	string ip;
+	uint16_t port;
+	string username;
+	string password;
+};
+
 struct Configuration {
 	std::vector<IceServer> iceServers;
+	std::optional<ProxyServer> proxyServer;
 	bool enableIceTcp = false;
 	uint16_t portRangeBegin = 1024;
 	uint16_t portRangeEnd = 65535;

include/rtc/datachannel.hpp

@@ -40,7 +40,7 @@ class DataChannel : public std::enable_shared_from_this<DataChannel>, public Cha
 public:
 	DataChannel(std::weak_ptr<PeerConnection> pc, unsigned int stream, string label,
 	            string protocol, Reliability reliability);
-	DataChannel(std::weak_ptr<PeerConnection> pc, std::shared_ptr<SctpTransport> transport,
+	DataChannel(std::weak_ptr<PeerConnection> pc, std::weak_ptr<SctpTransport> transport,
 	            unsigned int stream);
 	~DataChannel();
 
@@ -65,13 +65,13 @@ public:
 
 private:
 	void remoteClose();
-	void open(std::shared_ptr<SctpTransport> sctpTransport);
+	void open(std::shared_ptr<SctpTransport> transport);
 	bool outgoing(mutable_message_ptr message);
 	void incoming(message_ptr message);
 	void processOpenMessage(message_ptr message);
 
 	const std::weak_ptr<PeerConnection> mPeerConnection;
-	std::shared_ptr<SctpTransport> mSctpTransport;
+	std::weak_ptr<SctpTransport> mSctpTransport;
 
 	unsigned int mStream;
 	string mLabel;

include/rtc/include.hpp

@@ -20,7 +20,6 @@
 #define RTC_INCLUDE_H
 
 #ifdef _WIN32
-#define WIN32_LEAN_AND_MEAN
 #ifndef _WIN32_WINNT
 #define _WIN32_WINNT 0x0602
 #endif

include/rtc/peerconnection.hpp

@@ -52,14 +52,13 @@ public:
 		Connected = RTC_CONNECTED,
 		Disconnected = RTC_DISCONNECTED,
 		Failed = RTC_FAILED,
-		Closed = RTC_CLOSED,
-		Destroying = RTC_DESTROYING
+		Closed = RTC_CLOSED
 	};
 
 	enum class GatheringState : int {
 		New = RTC_GATHERING_NEW,
 		InProgress = RTC_GATHERING_INPROGRESS,
-		Complete = RTC_GATHERING_COMPLETE,
+		Complete = RTC_GATHERING_COMPLETE
 	};
 
 	PeerConnection(void);
@@ -88,12 +87,21 @@ public:
 	void onStateChange(std::function<void(State state)> callback);
 	void onGatheringStateChange(std::function<void(GatheringState state)> callback);
 
+	bool getSelectedCandidatePair(CandidateInfo *local, CandidateInfo *remote);
+
+	// Stats
+	void clearStats();
+	size_t bytesSent();
+	size_t bytesReceived();
+	std::optional<std::chrono::milliseconds> rtt();
+
 private:
 	init_token mInitToken = Init::Token();
 
 	std::shared_ptr<IceTransport> initIceTransport(Description::Role role);
 	std::shared_ptr<DtlsTransport> initDtlsTransport();
 	std::shared_ptr<SctpTransport> initSctpTransport();
+	void closeTransports();
 
 	void endLocalCandidates();
 	bool checkFingerprint(const std::string &fingerprint) const;
@@ -112,8 +120,10 @@ private:
 	void processLocalDescription(Description description);
 	void processLocalCandidate(Candidate candidate);
 	void triggerDataChannel(std::weak_ptr<DataChannel> weakDataChannel);
-	void changeState(State state);
-	void changeGatheringState(GatheringState state);
+	bool changeState(State state);
+	bool changeGatheringState(GatheringState state);
+
+	void resetCallbacks();
 
 	const Configuration mConfig;
 	const std::shared_ptr<Certificate> mCertificate;
@@ -124,7 +134,6 @@ private:
 	std::shared_ptr<IceTransport> mIceTransport;
 	std::shared_ptr<DtlsTransport> mDtlsTransport;
 	std::shared_ptr<SctpTransport> mSctpTransport;
-	std::recursive_mutex mInitMutex;
 
 	std::unordered_map<unsigned int, std::weak_ptr<DataChannel>> mDataChannels;
 	std::shared_mutex mDataChannelsMutex;

include/rtc/rtc.h

@@ -23,6 +23,8 @@
 extern "C" {
 #endif
 
+#include <stdint.h>
+
 // libdatachannel C API
 
 typedef enum {
@@ -31,8 +33,7 @@ typedef enum {
 	RTC_CONNECTED = 2,
 	RTC_DISCONNECTED = 3,
 	RTC_FAILED = 4,
-	RTC_CLOSED = 5,
-	RTC_DESTROYING = 6 // internal
+	RTC_CLOSED = 5
 } rtcState;
 
 typedef enum {
@@ -55,6 +56,8 @@ typedef enum {
 typedef struct {
 	const char **iceServers;
 	int iceServersCount;
+	uint16_t portRangeBegin;
+	uint16_t portRangeEnd;
 } rtcConfiguration;
 
 typedef void (*dataChannelCallbackFunc)(int dc, void *ptr);

src/candidate.cpp

@@ -131,3 +131,33 @@ Candidate::operator string() const {
 std::ostream &operator<<(std::ostream &out, const rtc::Candidate &candidate) {
 	return out << std::string(candidate);
 }
+
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateType &type) {
+	switch (type) {
+	case rtc::CandidateType::Host:
+		return out << "Host";
+	case rtc::CandidateType::PeerReflexive:
+		return out << "PeerReflexive";
+	case rtc::CandidateType::Relayed:
+		return out << "Relayed";
+	case rtc::CandidateType::ServerReflexive:
+		return out << "ServerReflexive";
+	default:
+		return out << "Unknown";
+	}
+}
+
+std::ostream &operator<<(std::ostream &out, const rtc::CandidateTransportType &transportType) {
+	switch (transportType) {
+	case rtc::CandidateTransportType::TcpActive:
+		return out << "TcpActive";
+	case rtc::CandidateTransportType::TcpPassive:
+		return out << "TcpPassive";
+	case rtc::CandidateTransportType::TcpSo:
+		return out << "TcpSo";
+	case rtc::CandidateTransportType::Udp:
+		return out << "Udp";
+	default:
+		return out << "Unknown";
+	}
+}

src/channel.cpp

@@ -88,5 +88,14 @@ void Channel::triggerBufferedAmount(size_t amount) {
 		mBufferedAmountLowCallback();
 }
 
+void Channel::resetCallbacks() {
+	mOpenCallback = nullptr;
+	mClosedCallback = nullptr;
+	mErrorCallback = nullptr;
+	mMessageCallback = nullptr;
+	mAvailableCallback = nullptr;
+	mBufferedAmountLowCallback = nullptr;
+}
+
 } // namespace rtc
 

src/configuration.cpp

@@ -38,16 +38,17 @@ IceServer::IceServer(const string &url) {
 	});
 
 	string scheme = opt[2].value_or("stun");
+	relayType = RelayType::TurnUdp;
 	if (scheme == "stun" || scheme == "STUN")
 		type = Type::Stun;
 	else if (scheme == "turn" || scheme == "TURN")
 		type = Type::Turn;
-	else if (scheme == "turns" || scheme == "TURNS")
+	else if (scheme == "turns" || scheme == "TURNS") {
 		type = Type::Turn;
-	else
+		relayType = RelayType::TurnTls;
+	} else
 		throw std::invalid_argument("Unknown ICE server protocol: " + scheme);
 
-	relayType = RelayType::TurnUdp;
 	if (auto &query = opt[15]) {
 		if (query->find("transport=udp") != string::npos)
 			relayType = RelayType::TurnUdp;
@@ -84,4 +85,7 @@ IceServer::IceServer(string hostname_, string service_, string username_, string
     : hostname(std::move(hostname_)), service(std::move(service_)), type(Type::Turn),
       username(std::move(username_)), password(std::move(password_)), relayType(relayType_) {}
 
+ProxyServer::ProxyServer(Type type_, string ip_, uint16_t port_, string username_, string password_)
+    : type(type_), ip(ip_), port(port_), username(username_), password(password_) {}
+
 } // namespace rtc

src/datachannel.cpp

@@ -74,7 +74,7 @@ DataChannel::DataChannel(weak_ptr<PeerConnection> pc, unsigned int stream, strin
       mReliability(std::make_shared<Reliability>(std::move(reliability))),
       mRecvQueue(RECV_QUEUE_LIMIT, message_size_func) {}
 
-DataChannel::DataChannel(weak_ptr<PeerConnection> pc, shared_ptr<SctpTransport> transport,
+DataChannel::DataChannel(weak_ptr<PeerConnection> pc, weak_ptr<SctpTransport> transport,
                          unsigned int stream)
     : mPeerConnection(pc), mSctpTransport(transport), mStream(stream),
       mReliability(std::make_shared<Reliability>()),
@@ -93,10 +93,13 @@ string DataChannel::protocol() const { return mProtocol; }
 Reliability DataChannel::reliability() const { return *mReliability; }
 
 void DataChannel::close() {
-	if (mIsOpen.exchange(false) && mSctpTransport)
-		mSctpTransport->reset(mStream);
+	if (mIsOpen.exchange(false))
+		if (auto transport = mSctpTransport.lock())
+			transport->reset(mStream);
 	mIsClosed = true;
 	mSctpTransport.reset();
+
+	resetCallbacks();
 }
 
 void DataChannel::remoteClose() {
@@ -158,8 +161,8 @@ size_t DataChannel::maxMessageSize() const {
 
 size_t DataChannel::availableAmount() const { return mRecvQueue.amount(); }
 
-void DataChannel::open(shared_ptr<SctpTransport> sctpTransport) {
-	mSctpTransport = sctpTransport;
+void DataChannel::open(shared_ptr<SctpTransport> transport) {
+	mSctpTransport = transport;
 
 	uint8_t channelType = static_cast<uint8_t>(mReliability->type);
 	if (mReliability->unordered)
@@ -186,20 +189,24 @@ void DataChannel::open(shared_ptr<SctpTransport> sctpTransport) {
 	std::copy(mLabel.begin(), mLabel.end(), end);
 	std::copy(mProtocol.begin(), mProtocol.end(), end + mLabel.size());
 
-	mSctpTransport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
+	transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
 }
 
 bool DataChannel::outgoing(mutable_message_ptr message) {
-	if (mIsClosed || !mSctpTransport)
+	if (mIsClosed)
 		throw std::runtime_error("DataChannel is closed");
 
 	if (message->size() > maxMessageSize())
 		throw std::runtime_error("Message size exceeds limit");
 
+	auto transport = mSctpTransport.lock();
+	if (!transport)
+		throw std::runtime_error("DataChannel has no transport");
+
 	// Before the ACK has been received on a DataChannel, all messages must be sent ordered
 	message->reliability = mIsOpen ? mReliability : nullptr;
 	message->stream = mStream;
-	return mSctpTransport->send(message);
+	return transport->send(message);
 }
 
 void DataChannel::incoming(message_ptr message) {
@@ -238,6 +245,10 @@ void DataChannel::incoming(message_ptr message) {
 }
 
 void DataChannel::processOpenMessage(message_ptr message) {
+	auto transport = mSctpTransport.lock();
+	if (!transport)
+		throw std::runtime_error("DataChannel has no transport");
+
 	if (message->size() < sizeof(OpenMessage))
 		throw std::invalid_argument("DataChannel open message too small");
 
@@ -274,7 +285,7 @@ void DataChannel::processOpenMessage(message_ptr message) {
 	auto &ack = *reinterpret_cast<AckMessage *>(buffer.data());
 	ack.type = MESSAGE_ACK;
 
-	mSctpTransport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
+	transport->send(make_message(buffer.begin(), buffer.end(), Message::Control, mStream));
 
 	mIsOpen = true;
 	triggerOpen();

src/dtlstransport.cpp

@@ -72,8 +72,6 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 
 	PLOG_DEBUG << "Initializing DTLS transport (GnuTLS)";
 
-	gnutls_certificate_set_verify_function(mCertificate->credentials(), CertificateCallback);
-
 	bool active = lower->role() == Description::Role::Active;
 	unsigned int flags = GNUTLS_DATAGRAM | (active ? GNUTLS_CLIENT : GNUTLS_SERVER);
 	check_gnutls(gnutls_init(&mSession, flags));
@@ -86,12 +84,14 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 	check_gnutls(gnutls_priority_set_direct(mSession, priorities, &err_pos),
 	             "Unable to set TLS priorities");
 
+	gnutls_certificate_set_verify_function(mCertificate->credentials(), CertificateCallback);
 	check_gnutls(
 	    gnutls_credentials_set(mSession, GNUTLS_CRD_CERTIFICATE, mCertificate->credentials()));
 
-	gnutls_dtls_set_mtu(mSession, 1280 - 40 - 8); // min MTU over UDP/IPv6 (only for handshake)
-	gnutls_dtls_set_timeouts(mSession, 400, 60000);
-	gnutls_handshake_set_timeout(mSession, 60000);
+	gnutls_dtls_set_timeouts(mSession,
+	                         1000,   // 1s retransmission timeout recommended by RFC 6347
+	                         30000); // 30s total timeout
+	gnutls_handshake_set_timeout(mSession, 30000);
 
 	gnutls_session_set_ptr(mSession, this);
 	gnutls_transport_set_ptr(mSession, this);
@@ -110,15 +110,16 @@ DtlsTransport::~DtlsTransport() {
 
 DtlsTransport::State DtlsTransport::state() const { return mState; }
 
-void DtlsTransport::stop() {
-	Transport::stop();
+bool DtlsTransport::stop() {
+	if (!Transport::stop())
+		return false;
 
-	if (mRecvThread.joinable()) {
 	PLOG_DEBUG << "Stopping DTLS recv thread";
 	mIncomingQueue.stop();
 	gnutls_bye(mSession, GNUTLS_SHUT_RDWR);
 	mRecvThread.join();
-	}
+	onRecv(nullptr);
+	return true;
 }
 
 bool DtlsTransport::send(message_ptr message) {
@@ -159,6 +160,7 @@ void DtlsTransport::runRecvLoop() {
 	// Handshake loop
 	try {
 		changeState(State::Connecting);
+		gnutls_dtls_set_mtu(mSession, 1280 - 40 - 8); // min MTU over UDP/IPv6
 
 		int ret;
 		do {
@@ -182,6 +184,7 @@ void DtlsTransport::runRecvLoop() {
 
 	// Receive loop
 	try {
+		PLOG_INFO << "DTLS handshake done";
 		changeState(State::Connected);
 
 		const size_t bufferSize = maxMtu;
@@ -388,7 +391,6 @@ DtlsTransport::DtlsTransport(shared_ptr<IceTransport> lower, shared_ptr<Certific
 		throw std::runtime_error("Unable to create SSL instance");
 
 	SSL_set_ex_data(mSsl, TransportExIndex, this);
-	SSL_set_mtu(mSsl, 1280 - 40 - 8); // min MTU over UDP/IPv6
 
 	if (lower->role() == Description::Role::Active)
 		SSL_set_connect_state(mSsl);
@@ -417,16 +419,16 @@ DtlsTransport::~DtlsTransport() {
 	SSL_CTX_free(mCtx);
 }
 
-void DtlsTransport::stop() {
-	Transport::stop();
+bool DtlsTransport::stop() {
+	if (!Transport::stop())
+		return false;
 
-	if (mRecvThread.joinable()) {
 	PLOG_DEBUG << "Stopping DTLS recv thread";
 	mIncomingQueue.stop();
 	mRecvThread.join();
-
 	SSL_shutdown(mSsl);
-	}
+	onRecv(nullptr);
+	return true;
 }
 
 DtlsTransport::State DtlsTransport::state() const { return mState; }
@@ -462,53 +464,71 @@ void DtlsTransport::runRecvLoop() {
 	const size_t maxMtu = 4096;
 	try {
 		changeState(State::Connecting);
+		SSL_set_mtu(mSsl, 1280 - 40 - 8); // min MTU over UDP/IPv6
 
+		// Initiate the handshake
 		int ret = SSL_do_handshake(mSsl);
 		check_openssl_ret(mSsl, ret, "Handshake failed");
 
 		const size_t bufferSize = maxMtu;
 		byte buffer[bufferSize];
 		while (true) {
-			std::optional<milliseconds> duration;
-			struct timeval timeout = {};
-			if (DTLSv1_get_timeout(mSsl, &timeout))
-				duration = milliseconds(timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
-
-			if (!mIncomingQueue.wait(duration))
-				break; // queue is stopped
-
-			message_ptr decrypted;
-			if (!mIncomingQueue.empty()) {
+			// Process pending messages
+			while (!mIncomingQueue.empty()) {
 				auto message = *mIncomingQueue.pop();
 				BIO_write(mInBio, message->data(), message->size());
 
-				int ret = SSL_read(mSsl, buffer, bufferSize);
-				if (!check_openssl_ret(mSsl, ret))
-					break;
-
-				if (ret > 0)
-					decrypted = make_message(buffer, buffer + ret);
-			}
-
 				if (mState == State::Connecting) {
-				if (SSL_is_init_finished(mSsl)) {
-					changeState(State::Connected);
-
-					// RFC 8261: DTLS MUST support sending messages larger than the current path MTU
-					// See https://tools.ietf.org/html/rfc8261#section-5
-					SSL_set_mtu(mSsl, maxMtu + 1);
-				} else {
 					// Continue the handshake
 					int ret = SSL_do_handshake(mSsl);
 					if (!check_openssl_ret(mSsl, ret, "Handshake failed"))
 						break;
 
-					DTLSv1_handle_timeout(mSsl);
+					if (SSL_is_init_finished(mSsl)) {
+						// RFC 8261: DTLS MUST support sending messages larger than the current path
+						// MTU See https://tools.ietf.org/html/rfc8261#section-5
+						SSL_set_mtu(mSsl, maxMtu + 1);
+
+						PLOG_INFO << "DTLS handshake done";
+						changeState(State::Connected);
+					}
+				} else {
+					int ret = SSL_read(mSsl, buffer, bufferSize);
+					if (!check_openssl_ret(mSsl, ret))
+						break;
+					if (ret > 0)
+						recv(make_message(buffer, buffer + ret));
 				}
 			}
 
-			if (decrypted)
-				recv(decrypted);
+			// No more messages pending, retransmit and rearm timeout if connecting
+			std::optional<milliseconds> duration;
+			if (mState == State::Connecting) {
+				// Warning: This function breaks the usual return value convention
+				int ret = DTLSv1_handle_timeout(mSsl);
+				if (ret < 0) {
+					throw std::runtime_error("Handshake timeout"); // write BIO can't fail
+				} else if (ret > 0) {
+					LOG_VERBOSE << "OpenSSL did DTLS retransmit";
+				}
+
+				struct timeval timeout = {};
+				if (mState == State::Connecting && DTLSv1_get_timeout(mSsl, &timeout)) {
+					duration = milliseconds(timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
+					// Also handle handshake timeout manually because OpenSSL actually doesn't...
+					// OpenSSL backs off exponentially in base 2 starting from the recommended 1s
+					// so this allows for 5 retransmissions and fails after roughly 30s.
+					if (duration > 30s) {
+						throw std::runtime_error("Handshake timeout");
+					} else {
+						LOG_VERBOSE << "OpenSSL DTLS retransmit timeout is " << duration->count()
+						            << "ms";
+					}
+				}
+			}
+
+			if (!mIncomingQueue.wait(duration))
+				break; // queue is stopped
 		}
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "DTLS recv: " << e.what();
@@ -569,7 +589,8 @@ int DtlsTransport::BioMethodWrite(BIO *bio, const char *in, int inl) {
 	if (!transport)
 		return -1;
 	auto b = reinterpret_cast<const byte *>(in);
-	return transport->outgoing(make_message(b, b + inl)) ? inl : 0;
+	transport->outgoing(make_message(b, b + inl));
+	return inl; // can't fail
 }
 
 long DtlsTransport::BioMethodCtrl(BIO *bio, int cmd, long num, void *ptr) {

src/dtlstransport.hpp

@@ -57,7 +57,7 @@ public:
 
 	State state() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if dropped
 
 private:

src/icetransport.cpp

@@ -18,6 +18,7 @@
 
 #include "icetransport.hpp"
 #include "configuration.hpp"
+#include "transport.hpp"
 
 #include <iostream>
 #include <random>
@@ -72,7 +73,7 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 	unsigned seed = std::chrono::system_clock::now().time_since_epoch().count();
 	std::shuffle(servers.begin(), servers.end(), std::default_random_engine(seed));
 
-	// Pick a STUN server
+	// Pick a STUN server (TURN support is not implemented in libjuice yet)
 	for (auto &server : servers) {
 		if (!server.hostname.empty() && server.type == IceServer::Type::Stun) {
 			if (server.service.empty())
@@ -86,7 +87,12 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 		}
 	}
 
-	// TURN support is not implemented yet
+	// Port range
+	if (config.portRangeBegin > 1024 ||
+	    (config.portRangeEnd != 0 && config.portRangeEnd != 65535)) {
+		jconfig.local_port_range_begin = config.portRangeBegin;
+		jconfig.local_port_range_end = config.portRangeEnd;
+	}
 
 	// Create agent
 	mAgent = decltype(mAgent)(juice_create(&jconfig), juice_destroy);
@@ -96,8 +102,9 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 IceTransport::~IceTransport() { stop(); }
 
-void IceTransport::stop() {
-	// Nothing to do
+bool IceTransport::stop() {
+	onRecv(nullptr);
+	return Transport::stop();
 }
 
 Description::Role IceTransport::role() const { return mRole; }
@@ -312,6 +319,18 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 	g_object_set(G_OBJECT(mNiceAgent.get()), "upnp", FALSE, nullptr);
 	g_object_set(G_OBJECT(mNiceAgent.get()), "upnp-timeout", 200, nullptr);
 
+	// Proxy
+	if (config.proxyServer.has_value()) {
+		ProxyServer proxyServer = config.proxyServer.value();
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-type", proxyServer.type, nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-ip", proxyServer.ip.c_str(), nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-port", proxyServer.port, nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-username", proxyServer.username.c_str(),
+		             nullptr);
+		g_object_set(G_OBJECT(mNiceAgent.get()), "proxy-password", proxyServer.password.c_str(),
+		             nullptr);
+	}
+
 	// Randomize order
 	std::vector<IceServer> servers = config.iceServers;
 	unsigned seed = std::chrono::system_clock::now().time_since_epoch().count();
@@ -426,16 +445,19 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 
 IceTransport::~IceTransport() { stop(); }
 
-void IceTransport::stop() {
+bool IceTransport::stop() {
 	if (mTimeoutId) {
 		g_source_remove(mTimeoutId);
 		mTimeoutId = 0;
 	}
-	if (mMainLoopThread.joinable()) {
+
+	if (!Transport::stop())
+		return false;
+
 	PLOG_DEBUG << "Stopping ICE thread";
 	g_main_loop_quit(mMainLoop.get());
 	mMainLoopThread.join();
-	}
+	return true;
 }
 
 Description::Role IceTransport::role() const { return mRole; }
@@ -651,6 +673,58 @@ void IceTransport::LogCallback(const gchar *logDomain, GLogLevelFlags logLevel,
 	PLOG(severity) << "nice: " << message;
 }
 
+bool IceTransport::getSelectedCandidatePair(CandidateInfo *localInfo, CandidateInfo *remoteInfo) {
+	NiceCandidate *local, *remote;
+	gboolean result = nice_agent_get_selected_pair(mNiceAgent.get(), mStreamId, 1, &local, &remote);
+
+	if (!result)
+		return false;
+
+	char ipaddr[INET6_ADDRSTRLEN];
+	nice_address_to_string(&local->addr, ipaddr);
+	localInfo->address = std::string(ipaddr);
+	localInfo->port = nice_address_get_port(&local->addr);
+	localInfo->type = IceTransport::NiceTypeToCandidateType(local->type);
+	localInfo->transportType =
+	    IceTransport::NiceTransportTypeToCandidateTransportType(local->transport);
+
+	nice_address_to_string(&remote->addr, ipaddr);
+	remoteInfo->address = std::string(ipaddr);
+	remoteInfo->port = nice_address_get_port(&remote->addr);
+	remoteInfo->type = IceTransport::NiceTypeToCandidateType(remote->type);
+	remoteInfo->transportType =
+	    IceTransport::NiceTransportTypeToCandidateTransportType(remote->transport);
+
+	return true;
+}
+
+const CandidateType IceTransport::NiceTypeToCandidateType(NiceCandidateType type) {
+	switch (type) {
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_HOST:
+		return CandidateType::Host;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_PEER_REFLEXIVE:
+		return CandidateType::PeerReflexive;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_RELAYED:
+		return CandidateType::Relayed;
+	case NiceCandidateType::NICE_CANDIDATE_TYPE_SERVER_REFLEXIVE:
+		return CandidateType::ServerReflexive;
+	}
+}
+
+const CandidateTransportType
+IceTransport::NiceTransportTypeToCandidateTransportType(NiceCandidateTransport type) {
+	switch (type) {
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_TCP_ACTIVE:
+		return CandidateTransportType::TcpActive;
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_TCP_PASSIVE:
+		return CandidateTransportType::TcpPassive;
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_TCP_SO:
+		return CandidateTransportType::TcpSo;
+	case NiceCandidateTransport::NICE_CANDIDATE_TRANSPORT_UDP:
+		return CandidateTransportType::Udp;
+	}
+}
+
 } // namespace rtc
 
 #endif

src/icetransport.hpp

@@ -56,6 +56,8 @@ public:
 		Completed = NICE_COMPONENT_STATE_READY,
 		Failed = NICE_COMPONENT_STATE_FAILED,
 	};
+
+	bool getSelectedCandidatePair(CandidateInfo *local, CandidateInfo *remote);
 #endif
 	enum class GatheringState { New = 0, InProgress = 1, Complete = 2 };
 
@@ -79,7 +81,7 @@ public:
 	std::optional<string> getLocalAddress() const;
 	std::optional<string> getRemoteAddress() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if dropped
 
 private:
@@ -133,6 +135,8 @@ private:
 	static gboolean TimeoutCallback(gpointer userData);
 	static void LogCallback(const gchar *log_domain, GLogLevelFlags log_level, const gchar *message,
 	                        gpointer user_data);
+	static const CandidateType NiceTypeToCandidateType(NiceCandidateType type);
+	static const CandidateTransportType NiceTransportTypeToCandidateTransportType(NiceCandidateTransport type);
 #endif
 };
 

src/peerconnection.cpp

@@ -24,6 +24,7 @@
 #include "sctptransport.hpp"
 
 #include <iostream>
+#include <thread>
 
 namespace rtc {
 
@@ -32,34 +33,33 @@ using namespace std::placeholders;
 using std::shared_ptr;
 using std::weak_ptr;
 
-PeerConnection::PeerConnection() : PeerConnection(Configuration()) {
+template <typename F, typename T, typename... Args> auto weak_bind(F &&f, T *t, Args &&... _args) {
+	return [bound = std::bind(f, t, _args...), weak_this = t->weak_from_this()](auto &&... args) {
+		if (auto shared_this = weak_this.lock())
+			bound(args...);
+	};
 }
 
+template <typename F, typename T, typename... Args>
+auto weak_bind_verifier(F &&f, T *t, Args &&... _args) {
+	return [bound = std::bind(f, t, _args...), weak_this = t->weak_from_this()](auto &&... args) {
+		if (auto shared_this = weak_this.lock())
+			return bound(args...);
+		else
+			return false;
+	};
+}
+
+PeerConnection::PeerConnection() : PeerConnection(Configuration()) {}
+
 PeerConnection::PeerConnection(const Configuration &config)
     : mConfig(config), mCertificate(make_certificate("libdatachannel")), mState(State::New) {}
 
-PeerConnection::~PeerConnection() {
-	changeState(State::Destroying);
-	close();
-	mSctpTransport.reset();
-	mDtlsTransport.reset();
-	mIceTransport.reset();
-}
+PeerConnection::~PeerConnection() { close(); }
 
 void PeerConnection::close() {
-	// Close DataChannels
 	closeDataChannels();
-
-	// Close Transports
-	for (int i = 0; i < 2; ++i) { // Make sure a transport wasn't spawn behind our back
-		if (auto transport = std::atomic_load(&mSctpTransport))
-			transport->stop();
-		if (auto transport = std::atomic_load(&mDtlsTransport))
-			transport->stop();
-		if (auto transport = std::atomic_load(&mIceTransport))
-			transport->stop();
-	}
-	changeState(State::Closed);
+	closeTransports();
 }
 
 const Configuration *PeerConnection::config() const { return &mConfig; }
@@ -101,7 +101,7 @@ void PeerConnection::setRemoteDescription(Description description) {
 		if (!sctpTransport && 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);
+			std::unique_lock lock(mDataChannelsMutex); // we are going to swap the container
 			decltype(mDataChannels) newDataChannels;
 			auto it = mDataChannels.begin();
 			while (it != mDataChannels.end()) {
@@ -203,13 +203,15 @@ void PeerConnection::onGatheringStateChange(std::function<void(GatheringState st
 
 shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role) {
 	try {
-		std::lock_guard lock(mInitMutex);
 		if (auto transport = std::atomic_load(&mIceTransport))
 			return transport;
 
 		auto transport = std::make_shared<IceTransport>(
-		    mConfig, role, std::bind(&PeerConnection::processLocalCandidate, this, _1),
-		    [this](IceTransport::State state) {
+		    mConfig, role, weak_bind(&PeerConnection::processLocalCandidate, this, _1),
+		    [this, weak_this = weak_from_this()](IceTransport::State state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case IceTransport::State::Connecting:
 				    changeState(State::Connecting);
@@ -228,7 +230,10 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role
 				    break;
 			    }
 		    },
-		    [this](IceTransport::GatheringState state) {
+		    [this, weak_this = weak_from_this()](IceTransport::GatheringState state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case IceTransport::GatheringState::InProgress:
 				    changeGatheringState(GatheringState::InProgress);
@@ -242,8 +247,15 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role
 				    break;
 			    }
 		    });
+
 		std::atomic_store(&mIceTransport, transport);
+		if (mState == State::Closed) {
+			mIceTransport.reset();
+			transport->stop();
+			throw std::runtime_error("Connection is closed");
+		}
 		return transport;
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
 		changeState(State::Failed);
@@ -253,14 +265,16 @@ shared_ptr<IceTransport> PeerConnection::initIceTransport(Description::Role role
 
 shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 	try {
-		std::lock_guard lock(mInitMutex);
 		if (auto transport = std::atomic_load(&mDtlsTransport))
 			return transport;
 
 		auto lower = std::atomic_load(&mIceTransport);
 		auto transport = std::make_shared<DtlsTransport>(
-		    lower, mCertificate, std::bind(&PeerConnection::checkFingerprint, this, _1),
-		    [this](DtlsTransport::State state) {
+		    lower, mCertificate, weak_bind_verifier(&PeerConnection::checkFingerprint, this, _1),
+		    [this, weak_this = weak_from_this()](DtlsTransport::State state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case DtlsTransport::State::Connected:
 				    initSctpTransport();
@@ -276,8 +290,15 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 				    break;
 			    }
 		    });
+
 		std::atomic_store(&mDtlsTransport, transport);
+		if (mState == State::Closed) {
+			mDtlsTransport.reset();
+			transport->stop();
+			throw std::runtime_error("Connection is closed");
+		}
 		return transport;
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
 		changeState(State::Failed);
@@ -287,16 +308,18 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
 
 shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 	try {
-		std::lock_guard lock(mInitMutex);
 		if (auto transport = std::atomic_load(&mSctpTransport))
 			return transport;
 
 		uint16_t sctpPort = remoteDescription()->sctpPort().value_or(DEFAULT_SCTP_PORT);
 		auto lower = std::atomic_load(&mDtlsTransport);
 		auto transport = std::make_shared<SctpTransport>(
-		    lower, sctpPort, std::bind(&PeerConnection::forwardMessage, this, _1),
-		    std::bind(&PeerConnection::forwardBufferedAmount, this, _1, _2),
-		    [this](SctpTransport::State state) {
+		    lower, sctpPort, weak_bind(&PeerConnection::forwardMessage, this, _1),
+		    weak_bind(&PeerConnection::forwardBufferedAmount, this, _1, _2),
+		    [this, weak_this = weak_from_this()](SctpTransport::State state) {
+			    auto shared_this = weak_this.lock();
+			    if (!shared_this)
+				    return;
 			    switch (state) {
 			    case SctpTransport::State::Connected:
 				    changeState(State::Connected);
@@ -315,8 +338,15 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 				    break;
 			    }
 		    });
+
 		std::atomic_store(&mSctpTransport, transport);
+		if (mState == State::Closed) {
+			mSctpTransport.reset();
+			transport->stop();
+			throw std::runtime_error("Connection is closed");
+		}
 		return transport;
+
 	} catch (const std::exception &e) {
 		PLOG_ERROR << e.what();
 		changeState(State::Failed);
@@ -324,6 +354,34 @@ shared_ptr<SctpTransport> PeerConnection::initSctpTransport() {
 	}
 }
 
+void PeerConnection::closeTransports() {
+	// Change state to sink state Closed to block init methods
+	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
+	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));
+	if (sctp || dtls || ice) {
+		std::thread t([sctp, dtls, ice]() mutable {
+			if (sctp)
+				sctp->stop();
+			if (dtls)
+				dtls->stop();
+			if (ice)
+				ice->stop();
+
+			sctp.reset();
+			dtls.reset();
+			ice.reset();
+		});
+		t.detach();
+	}
+}
+
 void PeerConnection::endLocalCandidates() {
 	std::lock_guard lock(mLocalDescriptionMutex);
 	if (mLocalDescription)
@@ -359,7 +417,7 @@ void PeerConnection::forwardMessage(message_ptr message) {
 		    message->stream % 2 == remoteParity) {
 			channel =
 			    std::make_shared<DataChannel>(shared_from_this(), sctpTransport, message->stream);
-			channel->onOpen(std::bind(&PeerConnection::triggerDataChannel, this,
+			channel->onOpen(weak_bind(&PeerConnection::triggerDataChannel, this,
 			                          weak_ptr<DataChannel>{channel}));
 			mDataChannels.insert(std::make_pair(message->stream, channel));
 		} else {
@@ -384,7 +442,7 @@ shared_ptr<DataChannel> PeerConnection::emplaceDataChannel(Description::Role rol
 	// The active side must use streams with even identifiers, whereas the passive side must use
 	// streams with odd identifiers.
 	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-protocol-09#section-6
-	std::unique_lock lock(mDataChannelsMutex);
+	std::unique_lock lock(mDataChannelsMutex); // we are going to emplace
 	unsigned int stream = (role == Description::Role::Active) ? 0 : 1;
 	while (mDataChannels.find(stream) != mDataChannels.end()) {
 		stream += 2;
@@ -398,31 +456,42 @@ shared_ptr<DataChannel> PeerConnection::emplaceDataChannel(Description::Role rol
 }
 
 shared_ptr<DataChannel> PeerConnection::findDataChannel(uint16_t stream) {
-	std::shared_lock lock(mDataChannelsMutex);
-	shared_ptr<DataChannel> channel;
-	if (auto it = mDataChannels.find(stream); it != mDataChannels.end()) {
-		channel = it->second.lock();
-		if (!channel)
-			mDataChannels.erase(it);
-	}
+	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 nullptr;
 }
 
 void PeerConnection::iterateDataChannels(
     std::function<void(shared_ptr<DataChannel> channel)> func) {
-	std::shared_lock lock(mDataChannelsMutex);
+	// Iterate
+	{
+		std::shared_lock lock(mDataChannelsMutex); // read-only
 		auto it = mDataChannels.begin();
 		while (it != mDataChannels.end()) {
 			auto channel = it->second.lock();
-		if (!channel) {
+			if (channel && !channel->isClosed())
+				func(channel);
+
+			++it;
+		}
+	}
+
+	// Cleanup
+	{
+		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;
 			}
-		if (!channel->isClosed()) {
-			func(channel);
-		}
+
 			++it;
 		}
+	}
 }
 
 void PeerConnection::openDataChannels() {
@@ -470,21 +539,72 @@ void PeerConnection::triggerDataChannel(weak_ptr<DataChannel> weakDataChannel) {
 	mDataChannelCallback(dataChannel);
 }
 
-void PeerConnection::changeState(State state) {
+bool PeerConnection::changeState(State state) {
 	State current;
 	do {
 		current = mState.load();
-		if (current == state || current == State::Destroying)
-			return;
+		if (current == state)
+			return true;
+		if (current == State::Closed)
+			return false;
+
 	} while (!mState.compare_exchange_weak(current, state));
 
-	if (state != State::Destroying)
 	mStateChangeCallback(state);
+	return true;
 }
 
-void PeerConnection::changeGatheringState(GatheringState state) {
+bool PeerConnection::changeGatheringState(GatheringState state) {
 	if (mGatheringState.exchange(state) != state)
 		mGatheringStateChangeCallback(state);
+	return true;
+}
+
+void PeerConnection::resetCallbacks() {
+	// Unregister all callbacks
+	mDataChannelCallback = nullptr;
+	mLocalDescriptionCallback = nullptr;
+	mLocalCandidateCallback = nullptr;
+	mStateChangeCallback = nullptr;
+	mGatheringStateChangeCallback = nullptr;
+}
+
+bool PeerConnection::getSelectedCandidatePair(CandidateInfo *local, CandidateInfo *remote) {
+#if not USE_JUICE
+	auto iceTransport = std::atomic_load(&mIceTransport);
+	return iceTransport->getSelectedCandidatePair(local, remote);
+#else
+	PLOG_WARNING << "getSelectedCandidatePair is not implemented for libjuice";
+	return false;
+#endif
+}
+
+void PeerConnection::clearStats() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->clearStats();
+}
+
+size_t PeerConnection::bytesSent() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->bytesSent();
+	return 0;
+}
+
+size_t PeerConnection::bytesReceived() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->bytesReceived();
+	return 0;
+}
+
+std::optional<std::chrono::milliseconds> PeerConnection::rtt() {
+	auto sctpTransport = std::atomic_load(&mSctpTransport);
+	if (sctpTransport)
+		return sctpTransport->rtt();
+	PLOG_WARNING << "Could not load sctpTransport";
+	return std::nullopt;
 }
 
 } // namespace rtc
@@ -511,9 +631,6 @@ std::ostream &operator<<(std::ostream &out, const rtc::PeerConnection::State &st
 	case State::Closed:
 		str = "closed";
 		break;
-	case State::Destroying:
-		str = "destroying";
-		break;
 	default:
 		str = "unknown";
 		break;
@@ -540,4 +657,3 @@ std::ostream &operator<<(std::ostream &out, const rtc::PeerConnection::Gathering
 	}
 	return out << str;
 }
-

src/rtc.cpp

@@ -53,6 +53,14 @@ void *getUserPointer(int id) {
 	return it != userPointerMap.end() ? it->second : nullptr;
 }
 
+void setUserPointer(int i, void *ptr) {
+	std::lock_guard lock(mutex);
+	if (ptr)
+		userPointerMap.insert(std::make_pair(i, ptr));
+	else
+		userPointerMap.erase(i);
+}
+
 shared_ptr<PeerConnection> getPeerConnection(int id) {
 	std::lock_guard lock(mutex);
 	auto it = peerConnectionMap.find(id);
@@ -99,18 +107,18 @@ bool eraseDataChannel(int dc) {
 
 void rtcInitLogger(rtcLogLevel level) { InitLogger(static_cast<LogLevel>(level)); }
 
-void rtcSetUserPointer(int i, void *ptr) {
-	if (ptr)
-		userPointerMap.insert(std::make_pair(i, ptr));
-	else
-		userPointerMap.erase(i);
-}
+void rtcSetUserPointer(int i, void *ptr) { setUserPointer(i, ptr); }
 
 int rtcCreatePeerConnection(const rtcConfiguration *config) {
 	Configuration c;
 	for (int i = 0; i < config->iceServersCount; ++i)
 		c.iceServers.emplace_back(string(config->iceServers[i]));
 
+	if (config->portRangeBegin || config->portRangeEnd) {
+		c.portRangeBegin = config->portRangeBegin;
+		c.portRangeEnd = config->portRangeEnd;
+	}
+
 	return emplacePeerConnection(std::make_shared<PeerConnection>(c));
 }
 

src/sctptransport.cpp

@@ -144,6 +144,14 @@ SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
 		throw std::runtime_error("Could not set socket option SCTP_INITMSG, errno=" +
 		                         std::to_string(errno));
 
+	// Prevent fragmented interleave of messages (i.e. level 0), see RFC 6458 8.1.20.
+	// Unless the user has set the fragmentation interleave level to 0, notifications
+	// may also be interleaved with partially delivered messages.
+	int level = 0;
+	if (usrsctp_setsockopt(mSock, IPPROTO_SCTP, SCTP_FRAGMENT_INTERLEAVE, &level, sizeof(level)))
+		throw std::runtime_error("Could not disable SCTP fragmented interleave, errno=" +
+		                         std::to_string(errno));
+
 	// The default send and receive window size of usrsctp is 256KiB, which is too small for
 	// realistic RTTs, therefore we increase it to 1MiB for better performance.
 	// See https://bugzilla.mozilla.org/show_bug.cgi?id=1051685
@@ -167,15 +175,15 @@ SctpTransport::~SctpTransport() {
 
 SctpTransport::State SctpTransport::state() const { return mState; }
 
-void SctpTransport::stop() {
-	Transport::stop();
-	onRecv(nullptr);
+bool SctpTransport::stop() {
+	if (!Transport::stop())
+		return false;
 
-	if (!mShutdown.exchange(true)) {
 	mSendQueue.stop();
 	safeFlush();
 	shutdown();
-	}
+	onRecv(nullptr);
+	return true;
 }
 
 void SctpTransport::connect() {
@@ -356,6 +364,8 @@ bool SctpTransport::trySendMessage(message_ptr message) {
 
 	if (ret >= 0) {
 		PLOG_VERBOSE << "SCTP sent size=" << message->size();
+		if (message->type == Message::Type::Binary || message->type == Message::Type::String)
+			mBytesSent += message->size();
 		return true;
 	} else if (errno == EWOULDBLOCK || errno == EAGAIN) {
 		PLOG_VERBOSE << "SCTP sending not possible";
@@ -396,13 +406,15 @@ int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, co
 		if (!len)
 			return -1;
 
+		// This is valid because SCTP_FRAGMENT_INTERLEAVE is set to level 0
+		// so partial messages and notifications may not be interleaved.
 		if (flags & MSG_EOR) {
 			if (!mPartialRecv.empty()) {
 				mPartialRecv.insert(mPartialRecv.end(), data, data + len);
 				data = mPartialRecv.data();
 				len = mPartialRecv.size();
 			}
-			// Message is complete, process it
+			// Message/Notification is complete, process it
 			if (flags & MSG_NOTIFICATION)
 				processNotification(reinterpret_cast<const union sctp_notification *>(data), len);
 			else
@@ -410,7 +422,7 @@ int SctpTransport::handleRecv(struct socket *sock, union sctp_sockstore addr, co
 
 			mPartialRecv.clear();
 		} else {
-			// Message is not complete
+			// Message/Notification is not complete
 			mPartialRecv.insert(mPartialRecv.end(), data, data + len);
 		}
 	} catch (const std::exception &e) {
@@ -460,9 +472,11 @@ void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, Payl
 
 	case PPID_STRING:
 		if (mPartialStringData.empty()) {
+			mBytesReceived += len;
 			recv(make_message(data, data + len, Message::String, sid));
 		} else {
 			mPartialStringData.insert(mPartialStringData.end(), data, data + len);
+			mBytesReceived += mPartialStringData.size();
 			recv(make_message(mPartialStringData.begin(), mPartialStringData.end(), Message::String,
 			                  sid));
 			mPartialStringData.clear();
@@ -482,9 +496,11 @@ void SctpTransport::processData(const byte *data, size_t len, uint16_t sid, Payl
 
 	case PPID_BINARY:
 		if (mPartialBinaryData.empty()) {
+			mBytesReceived += len;
 			recv(make_message(data, data + len, 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.clear();
@@ -568,6 +584,26 @@ void SctpTransport::processNotification(const union sctp_notification *notify, s
 	}
 }
 
+void SctpTransport::clearStats() {
+	mBytesReceived = 0;
+	mBytesSent = 0;
+}
+
+size_t SctpTransport::bytesSent() { return mBytesSent; }
+
+size_t SctpTransport::bytesReceived() { return mBytesReceived; }
+
+std::optional<std::chrono::milliseconds> SctpTransport::rtt() {
+	struct sctp_status status = {};
+	socklen_t len = sizeof(status);
+
+	if (usrsctp_getsockopt(this->mSock, IPPROTO_SCTP, SCTP_STATUS, &status, &len)) {
+		PLOG_WARNING << "Could not read SCTP_STATUS";
+		return std::nullopt;
+	}
+	return std::chrono::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 *ptr) {
 	int ret = static_cast<SctpTransport *>(ptr)->handleRecv(

src/sctptransport.hpp

@@ -49,11 +49,17 @@ public:
 
 	State state() const;
 
-	void stop() override;
+	bool stop() override;
 	bool send(message_ptr message) override; // false if buffered
 	void flush();
 	void reset(unsigned int stream);
 
+	// Stats
+	void clearStats();
+	size_t bytesSent();
+	size_t bytesReceived();
+	std::optional<std::chrono::milliseconds> rtt();
+
 private:
 	// Order seems wrong but these are the actual values
 	// See https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13#section-8
@@ -98,11 +104,12 @@ private:
 	bool mWritten = false;
 	bool mWrittenOnce = false;
 
-	std::atomic<bool> mShutdown = false;
-
 	state_callback mStateChangeCallback;
 	std::atomic<State> mState;
 
+	// Stats
+	std::atomic<size_t> mBytesSent = 0, mBytesReceived = 0;
+
 	binary mPartialRecv, mPartialStringData, mPartialBinaryData;
 
 	static int RecvCallback(struct socket *sock, union sctp_sockstore addr, void *data, size_t len,

src/transport.hpp

@@ -38,9 +38,10 @@ public:
 	}
 	virtual ~Transport() { stop(); }
 
-	virtual void stop() {
+	virtual bool stop() {
 		if (mLower)
 			mLower->onRecv(nullptr);
+		return !mShutdown.exchange(true);
 	}
 
 	virtual bool send(message_ptr message) = 0;
@@ -61,6 +62,7 @@ protected:
 private:
 	std::shared_ptr<Transport> mLower;
 	synchronized_callback<message_ptr> mRecvCallback;
+	std::atomic<bool> mShutdown = false;
 };
 
 } // namespace rtc

test/capi.cpp

@@ -131,21 +131,33 @@ static void deletePeer(Peer *peer) {
 }
 
 int test_capi_main() {
+	int attempts;
+
 	rtcInitLogger(RTC_LOG_DEBUG);
 
-	rtcConfiguration config;
-	memset(&config, 0, sizeof(config));
-	// const char *iceServers[1] = {"stun:stun.l.google.com:19302"};
-	// config.iceServers = iceServers;
-	// config.iceServersCount = 1;
-
 	// Create peer 1
-	peer1 = createPeer(&config);
+	rtcConfiguration config1;
+	memset(&config1, 0, sizeof(config1));
+	// STUN server example
+	// const char *iceServers[1] = {"stun:stun.l.google.com:19302"};
+	// config1.iceServers = iceServers;
+	// config1.iceServersCount = 1;
+
+	peer1 = createPeer(&config1);
 	if (!peer1)
 		goto error;
 
 	// Create peer 2
-	peer2 = createPeer(&config);
+	rtcConfiguration config2;
+	memset(&config2, 0, sizeof(config2));
+	// STUN server example
+	// config2.iceServers = iceServers;
+	// config2.iceServersCount = 1;
+	// Port range example
+	config2.portRangeBegin = 5000;
+	config2.portRangeEnd = 6000;
+
+	peer2 = createPeer(&config2);
 	if (!peer2)
 		goto error;
 
@@ -155,7 +167,19 @@ int test_capi_main() {
 	rtcSetClosedCallback(peer1->dc, closedCallback);
 	rtcSetMessageCallback(peer1->dc, messageCallback);
 
-	sleep(3);
+	attempts = 10;
+	while (!peer2->connected && !peer1->connected && attempts--)
+		sleep(1);
+
+	if (peer1->state != RTC_CONNECTED || peer2->state != RTC_CONNECTED) {
+		fprintf(stderr, "PeerConnection is not connected\n");
+		goto error;
+	}
+
+	if (!peer1->connected || !peer2->connected) {
+		fprintf(stderr, "DataChannel is not connected\n");
+		goto error;
+	}
 
 	char buffer[256];
 	if (rtcGetLocalAddress(peer1->pc, buffer, 256) >= 0)
@@ -167,13 +191,13 @@ int test_capi_main() {
 	if (rtcGetRemoteAddress(peer2->pc, buffer, 256) >= 0)
 		printf("Remote address 2: %s\n", buffer);
 
-	if (peer1->connected && peer2->connected) {
 	deletePeer(peer1);
+	sleep(1);
 	deletePeer(peer2);
 	sleep(1);
+
 	printf("Success\n");
 	return 0;
-	}
 
 error:
 	deletePeer(peer1);

test/connectivity.cpp

@@ -31,12 +31,20 @@ template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
 void test_connectivity() {
 	InitLogger(LogLevel::Debug);
 
-	Configuration config;
-	// config.iceServers.emplace_back("stun:stun.l.google.com:19302");
+	Configuration config1;
+	// STUN server example
+	// config1.iceServers.emplace_back("stun:stun.l.google.com:19302");
 
-	auto pc1 = std::make_shared<PeerConnection>(config);
+	auto pc1 = std::make_shared<PeerConnection>(config1);
 
-	auto pc2 = std::make_shared<PeerConnection>(config);
+	Configuration config2;
+	// STUN server example
+	// config2.iceServers.emplace_back("stun:stun.l.google.com:19302");
+	// Port range example
+	config2.portRangeBegin = 5000;
+	config2.portRangeEnd = 6000;
+
+	auto pc2 = std::make_shared<PeerConnection>(config2);
 
 	pc1->onLocalDescription([wpc2 = make_weak_ptr(pc2)](const Description &sdp) {
 		auto pc2 = wpc2.lock();
@@ -106,7 +114,16 @@ void test_connectivity() {
 		}
 	});
 
-	this_thread::sleep_for(3s);
+	int attempts = 10;
+	while ((!dc2 || !dc2->isOpen() || !dc1->isOpen()) && attempts--)
+		this_thread::sleep_for(1s);
+
+	if (pc1->state() != PeerConnection::State::Connected &&
+	    pc2->state() != PeerConnection::State::Connected)
+		throw runtime_error("PeerConnection is not connected");
+
+	if (!dc1->isOpen() || !dc2->isOpen())
+		throw runtime_error("DataChannel is not open");
 
 	if (auto addr = pc1->localAddress())
 		cout << "Local address 1:  " << *addr << endl;
@@ -117,12 +134,10 @@ void test_connectivity() {
 	if (auto addr = pc2->remoteAddress())
 		cout << "Remote address 2: " << *addr << endl;
 
-	if (!dc1->isOpen() || !dc2->isOpen())
-		throw runtime_error("DataChannel is not open");
-
+	// Delay close of peer 2 to check closing works properly
 	pc1->close();
+	this_thread::sleep_for(1s);
 	pc2->close();
-
 	this_thread::sleep_for(1s);
 
 	cout << "Success" << endl;

test/p2p/answerer.cpp

@@ -76,7 +76,8 @@ int main(int argc, char **argv) {
 		     << "* 0: Exit /"
 		     << " 1: Enter remote description /"
 		     << " 2: Enter remote candidate /"
-		     << " 3: Send message *" << endl
+		     << " 3: Send message /"
+		     << " 4: Print Connection Info *" << endl
 		     << "[Command]: ";
 
 		int command = -1;
@@ -120,6 +121,30 @@ int main(int argc, char **argv) {
 			dc->send(message);
 			break;
 		}
+		case 4: {
+			// Connection Info
+			if (!dc || !dc->isOpen()) {
+				cout << "** Channel is not Open ** ";
+				break;
+			}
+			CandidateInfo local, remote;
+			std::optional<std::chrono::milliseconds> rtt = pc->rtt();
+			if (pc->getSelectedCandidatePair(&local, &remote)) {
+				cout << "Local: " << local.address << ":" << local.port << " " << local.type << " "
+				     << local.transportType << endl;
+				cout << "Remote: " << remote.address << ":" << remote.port << " " << remote.type
+				     << " " << remote.transportType << endl;
+				cout << "Bytes Sent:" << pc->bytesSent()
+				     << " / Bytes Received:" << pc->bytesReceived() << " / Round-Trip Time:";
+				if (rtt.has_value())
+					cout << rtt.value().count();
+				else
+					cout << "null";
+				cout << " ms";
+			} else
+				cout << "Could not get Candidate Pair Info" << endl;
+			break;
+		}
 		default: {
 			cout << "** Invalid Command ** ";
 			break;

test/p2p/offerer.cpp

@@ -77,7 +77,8 @@ int main(int argc, char **argv) {
 		     << "* 0: Exit /"
 		     << " 1: Enter remote description /"
 		     << " 2: Enter remote candidate /"
-		     << " 3: Send message *" << endl
+		     << " 3: Send message /"
+		     << " 4: Print Connection Info *" << endl
 		     << "[Command]: ";
 
 		int command = -1;
@@ -120,6 +121,30 @@ int main(int argc, char **argv) {
 			dc->send(message);
 			break;
 		}
+		case 4: {
+			// Connection Info
+			if (!dc || !dc->isOpen()) {
+				cout << "** Channel is not Open ** ";
+				break;
+			}
+			CandidateInfo local, remote;
+			std::optional<std::chrono::milliseconds> rtt = pc->rtt();
+			if (pc->getSelectedCandidatePair(&local, &remote)) {
+				cout << "Local: " << local.address << ":" << local.port << " " << local.type << " "
+				     << local.transportType << endl;
+				cout << "Remote: " << remote.address << ":" << remote.port << " " << remote.type
+				     << " " << remote.transportType << endl;
+				cout << "Bytes Sent:" << pc->bytesSent()
+				     << " / Bytes Received:" << pc->bytesReceived() << " / Round-Trip Time:";
+				if (rtt.has_value())
+					cout << rtt.value().count();
+				else
+					cout << "null";
+				cout << " ms";
+			} else
+				cout << "Could not get Candidate Pair Info" << endl;
+			break;
+		}
 		default: {
 			cout << "** Invalid Command ** ";
 			break;