mii/libdatachannel
1
0
Fork 0
mirror of https://github.com/mii443/libdatachannel.git synced 2025-08-22 23:25:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into feature/stream-h264-opus

Browse Source
This commit is contained in:
Filip Klembara
2021-01-08 10:11:29 +01:00
parent 28a2868edc 3210814648
commit 5da67f6ca3
24 changed files with 570 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -57,6 +57,7 @@ set(LIBDATACHANNEL_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/src/log.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/message.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/peerconnection.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/logcounter.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/rtcp.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/sctptransport.cpp
${CMAKE_CURRENT_SOURCE_DIR}/src/threadpool.cpp
@ -116,6 +117,7 @@ set(LIBDATACHANNEL_HEADERS
set(TESTS_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/test/main.cpp
${CMAKE_CURRENT_SOURCE_DIR}/test/connectivity.cpp
${CMAKE_CURRENT_SOURCE_DIR}/test/turn_connectivity.cpp
${CMAKE_CURRENT_SOURCE_DIR}/test/track.cpp
${CMAKE_CURRENT_SOURCE_DIR}/test/capi_connectivity.cpp
${CMAKE_CURRENT_SOURCE_DIR}/test/capi_track.cpp

5
README.md
View File

@ -21,7 +21,7 @@ Protocol stack:
- SCTP-based Data Channels ([draft-ietf-rtcweb-data-channel-13](https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13))
- SRTP-based Media Transport ([draft-ietf-rtcweb-rtp-usage-26](https://tools.ietf.org/html/draft-ietf-rtcweb-rtp-usage-26))
- DTLS/UDP ([RFC7350](https://tools.ietf.org/html/rfc7350) and [RFC8261](https://tools.ietf.org/html/rfc8261))
- ICE ([RFC8445](https://tools.ietf.org/html/rfc8445)) with STUN ([RFC5389](https://tools.ietf.org/html/rfc5389))
- ICE ([RFC8445](https://tools.ietf.org/html/rfc8445)) with STUN ([RFC8489](https://tools.ietf.org/html/rfc8489)) and its extension TURN ([RFC8656](https://tools.ietf.org/html/rfc8656))
Features:
- Full IPv6 support
@ -30,7 +30,6 @@ Features:
- Multicast DNS candidates ([draft-ietf-rtcweb-mdns-ice-candidates-04](https://tools.ietf.org/html/draft-ietf-rtcweb-mdns-ice-candidates-04))
- SRTP and SRTCP key derivation from DTLS ([RFC5764](https://tools.ietf.org/html/rfc5764))
- Differentiated Services QoS ([draft-ietf-tsvwg-rtcweb-qos-18](https://tools.ietf.org/html/draft-ietf-tsvwg-rtcweb-qos-18))
- TURN relaying ([RFC5766](https://tools.ietf.org/html/rfc5766)) with [libnice](https://github.com/libnice/libnice) as ICE backend
Note only SDP BUNDLE mode is supported for media multiplexing ([draft-ietf-mmusic-sdp-bundle-negotiation-54](https://tools.ietf.org/html/draft-ietf-mmusic-sdp-bundle-negotiation-54)). The behavior is equivalent to the JSEP bundle-only policy: the library always negociates one unique network component, where SRTP media streams are multiplexed with SRTCP control packets ([RFC5761](https://tools.ietf.org/html/rfc5761)) and SCTP/DTLS data traffic ([RFC5764](https://tools.ietf.org/html/rfc5764)).
@ -74,8 +73,6 @@ $ git submodule update --init --recursive
The CMake library targets `libdatachannel` and `libdatachannel-static` respectively correspond to the shared and static libraries. The default target will build tests and examples. The option `USE_GNUTLS` allows to switch between OpenSSL (default) and GnuTLS, and the option `USE_NICE` allows to switch between libjuice as submodule (default) and libnice.
On Windows, the DLL resulting from the shared library build only exposes the C API, use the static library for the C++ API.
#### POSIX-compliant operating systems (including Linux and Apple macOS)
```bash
$ cmake -B build -DUSE_GNUTLS=1 -DUSE_NICE=0

2
deps/libjuice vendored

Submodule deps/libjuice updated: 421c650d9d...d691a7778d

28
examples/web/package-lock.json generated
View File

@ -5,11 +5,11 @@
"requires": true,
"dependencies": {
"bufferutil": {
"version": "4.0.1",
"resolved": "https://registry.npmjs.org/bufferutil/-/bufferutil-4.0.1.tgz",
"integrity": "sha512-xowrxvpxojqkagPcWRQVXZl0YXhRhAtBEIq3VoER1NH5Mw1n1o0ojdspp+GS2J//2gCVyrzQDApQ4unGF+QOoA==",
"version": "4.0.2",
"resolved": "https://registry.npmjs.org/bufferutil/-/bufferutil-4.0.2.tgz",
"integrity": "sha512-AtnG3W6M8B2n4xDQ5R+70EXvOpnXsFYg/AK2yTZd+HQ/oxAdz+GI+DvjmhBw3L0ole+LJ0ngqY4JMbDzkfNzhA==",
"requires": {
"node-gyp-build": "~3.7.0"
"node-gyp-build": "^4.2.0"
}
},
"d": {
@ -89,9 +89,9 @@
"integrity": "sha1-yobR/ogoFpsBICCOPchCS524NCw="
},
"node-gyp-build": {
"version": "3.7.0",
"resolved": "https://registry.npmjs.org/node-gyp-build/-/node-gyp-build-3.7.0.tgz",
"integrity": "sha512-L/Eg02Epx6Si2NXmedx+Okg+4UHqmaf3TNcxd50SF9NQGcJaON3AtU++kax69XV7YWz4tUspqZSAsVofhFKG2w=="
"version": "4.2.3",
"resolved": "https://registry.npmjs.org/node-gyp-build/-/node-gyp-build-4.2.3.tgz",
"integrity": "sha512-MN6ZpzmfNCRM+3t57PTJHgHyw/h4OWnZ6mR8P5j/uZtqQr46RRuDE/P+g3n0YR/AiYXeWixZZzaip77gdICfRg=="
},
"type": {
"version": "1.2.0",
@ -107,17 +107,17 @@
}
},
"utf-8-validate": {
"version": "5.0.2",
"resolved": "https://registry.npmjs.org/utf-8-validate/-/utf-8-validate-5.0.2.tgz",
"integrity": "sha512-SwV++i2gTD5qh2XqaPzBnNX88N6HdyhQrNNRykvcS0QKvItV9u3vPEJr+X5Hhfb1JC0r0e1alL0iB09rY8+nmw==",
"version": "5.0.3",
"resolved": "https://registry.npmjs.org/utf-8-validate/-/utf-8-validate-5.0.3.tgz",
"integrity": "sha512-jtJM6fpGv8C1SoH4PtG22pGto6x+Y8uPprW0tw3//gGFhDDTiuksgradgFN6yRayDP4SyZZa6ZMGHLIa17+M8A==",
"requires": {
"node-gyp-build": "~3.7.0"
"node-gyp-build": "^4.2.0"
}
},
"websocket": {
"version": "1.0.32",
"resolved": "https://registry.npmjs.org/websocket/-/websocket-1.0.32.tgz",
"integrity": "sha512-i4yhcllSP4wrpoPMU2N0TQ/q0O94LRG/eUQjEAamRltjQ1oT1PFFKOG4i877OlJgCG8rw6LrrowJp+TYCEWF7Q==",
"version": "1.0.33",
"resolved": "https://registry.npmjs.org/websocket/-/websocket-1.0.33.tgz",
"integrity": "sha512-XwNqM2rN5eh3G2CUQE3OHZj+0xfdH42+OFK6LdC2yqiC0YU8e5UK0nYre220T0IyyN031V/XOvtHvXozvJYFWA==",
"requires": {
"bufferutil": "^4.0.1",
"debug": "^2.2.0",

2
examples/web/package.json
View File

@ -18,6 +18,6 @@
},
"homepage": "https://github.com/paullouisageneau/libdatachannel#readme",
"dependencies": {
"websocket": "^1.0.32"
"websocket": "^1.0.33"
}
}

1
include/rtc/rtc.h
View File

@ -86,6 +86,7 @@ typedef enum { // Don't change, it must match plog severity
typedef struct {
const char **iceServers;
int iceServersCount;
bool enableIceTcp;
uint16_t portRangeBegin;
uint16_t portRangeEnd;
} rtcConfiguration;

1
src/capi.cpp
View File

@ -287,6 +287,7 @@ int rtcCreatePeerConnection(const rtcConfiguration *config) {
c.portRangeEnd = config->portRangeEnd;
}
c.enableIceTcp = config->enableIceTcp;
return emplacePeerConnection(std::make_shared<PeerConnection>(c));
});
}

5
src/datachannel.cpp
View File

@ -20,6 +20,7 @@
#include "include.hpp"
#include "peerconnection.hpp"
#include "sctptransport.hpp"
#include "logcounter.hpp"
#ifdef _WIN32
#include <winsock2.h>
@ -27,6 +28,7 @@
#include <arpa/inet.h>
#endif
rtc::LogCounter COUNTER_USERNEG_OPEN_MESSAGE(plog::warning, "Number of open messages for a user-negotiated DataChannel received");
namespace rtc {
using std::shared_ptr;
@ -170,7 +172,8 @@ void DataChannel::open(shared_ptr<SctpTransport> transport) {
}
void DataChannel::processOpenMessage(message_ptr) {
PLOG_WARNING << "Received an open message for a user-negotiated DataChannel, ignoring";
PLOG_DEBUG << "Received an open message for a user-negotiated DataChannel, ignoring";
COUNTER_USERNEG_OPEN_MESSAGE++;
}
bool DataChannel::outgoing(message_ptr message) {

121
src/dtlssrtptransport.cpp
View File

@ -18,6 +18,7 @@
#include "dtlssrtptransport.hpp"
#include "tls.hpp"
#include "logcounter.hpp"
#if RTC_ENABLE_MEDIA
@ -28,6 +29,15 @@ using std::shared_ptr;
using std::to_integer;
using std::to_string;
static rtc::LogCounter COUNTER_MEDIA_TRUNCATED(plog::warning, "Number of truncated SRT(C)P packets received");
static rtc::LogCounter COUNTER_UNKNOWN_PACKET_TYPE(plog::warning, "Number of RTP packets received with an unknown packet type");
static rtc::LogCounter COUNTER_SRTCP_REPLAY(plog::warning, "Number of SRTCP replay packets received");
static rtc::LogCounter COUNTER_SRTCP_AUTH_FAIL(plog::warning, "Number of SRTCP packets received that failed authentication checks");
static rtc::LogCounter COUNTER_SRTCP_FAIL(plog::warning, "Number of SRTCP packets received that had an unknown libSRTP failure");
static rtc::LogCounter COUNTER_SRTP_REPLAY(plog::warning, "Number of SRTP replay packets received");
static rtc::LogCounter COUNTER_SRTP_AUTH_FAIL(plog::warning, "Number of SRTP packets received that failed authentication checks");
static rtc::LogCounter COUNTER_SRTP_FAIL(plog::warning, "Number of SRTP packets received that had an unknown libSRTP failure");
namespace rtc {
void DtlsSrtpTransport::Init() { srtp_init(); }
@ -73,13 +83,14 @@ DtlsSrtpTransport::~DtlsSrtpTransport() {
srtp_dealloc(mSrtpOut);
}
bool DtlsSrtpTransport::sendMedia(message_ptr message) {
std::lock_guard lock(sendMutex);
if (!message)
return false;
if (!mInitDone) {
PLOG_WARNING << "SRTP media sent before keys are derived";
PLOG_ERROR << "SRTP media sent before keys are derived";
return false;
}
@ -109,32 +120,17 @@ bool DtlsSrtpTransport::sendMedia(message_ptr message) {
if (value2 >= 64 && value2 <= 95) { // Range 64-95 (inclusive) MUST be RTCP
if (srtp_err_status_t err = srtp_protect_rtcp(mSrtpOut, message->data(), &size)) {
if (err == srtp_err_status_replay_fail)
throw std::runtime_error("SRTCP packet is a replay");
else if (err == srtp_err_status_no_ctx) {
auto ssrc = reinterpret_cast<RTCP_SR *>(message->data())->senderSSRC();
PLOG_INFO << "Adding SSRC to SRTCP: " << ssrc;
addSSRC(ssrc);
if ((err = srtp_protect_rtcp(mSrtpOut, message->data(), &size)))
throw std::runtime_error("SRTCP protect error, status=" +
to_string(static_cast<int>(err)));
} else {
throw std::runtime_error("Outgoing SRTCP packet is a replay");
else
throw std::runtime_error("SRTCP protect error, status=" +
to_string(static_cast<int>(err)));
}
}
PLOG_VERBOSE << "Protected SRTCP packet, size=" << size;
} else {
if (srtp_err_status_t err = srtp_protect(mSrtpOut, message->data(), &size)) {
if (err == srtp_err_status_replay_fail)
throw std::runtime_error("Outgoing SRTP packet is a replay");
else if (err == srtp_err_status_no_ctx) {
auto ssrc = reinterpret_cast<RTP *>(message->data())->ssrc();
PLOG_INFO << "Adding SSRC to RTP: " << ssrc;
addSSRC(ssrc);
if ((err = srtp_protect(mSrtpOut, message->data(), &size)))
throw std::runtime_error("SRTP protect error, status=" +
to_string(static_cast<int>(err)));
} else
else
throw std::runtime_error("SRTP protect error, status=" +
to_string(static_cast<int>(err)));
}
@ -152,6 +148,7 @@ bool DtlsSrtpTransport::sendMedia(message_ptr message) {
return Transport::outgoing(message); // bypass DTLS DSCP marking
}
void DtlsSrtpTransport::incoming(message_ptr message) {
if (!mInitDone) {
// Bypas
@ -180,7 +177,8 @@ void DtlsSrtpTransport::incoming(message_ptr message) {
// The RTP header has a minimum size of 12 bytes
// An RTCP packet can have a minimum size of 8 bytes
if (size < 8) {
PLOG_WARNING << "Incoming SRTP/SRTCP packet too short, size=" << size;
COUNTER_MEDIA_TRUNCATED++;
PLOG_VERBOSE << "Incoming SRTP/SRTCP packet too short, size=" << size;
return;
}
@ -192,38 +190,36 @@ void DtlsSrtpTransport::incoming(message_ptr message) {
if (value2 >= 64 && value2 <= 95) { // Range 64-95 (inclusive) MUST be RTCP
PLOG_VERBOSE << "Incoming SRTCP packet, size=" << size;
if (srtp_err_status_t err = srtp_unprotect_rtcp(mSrtpIn, message->data(), &size)) {
if (err == srtp_err_status_replay_fail)
PLOG_WARNING << "Incoming SRTCP packet is a replay";
else if (err == srtp_err_status_auth_fail)
PLOG_WARNING << "Incoming SRTCP packet failed authentication check";
else if (err == srtp_err_status_no_ctx) {
auto ssrc = reinterpret_cast<RTCP_SR *>(message->data())->senderSSRC();
PLOG_INFO << "Adding SSRC to RTCP: " << ssrc;
addSSRC(ssrc);
} else {
PLOG_WARNING << "SRTCP unprotect error, status=" << err
<< " SSRC=" << ((RTCP_SR *)message->data())->senderSSRC();
}
if (err == srtp_err_status_replay_fail) {
PLOG_VERBOSE << "Incoming SRTCP packet is a replay";
COUNTER_SRTCP_REPLAY++;
}else if (err == srtp_err_status_auth_fail) {
PLOG_VERBOSE << "Incoming SRTCP packet failed authentication check";
COUNTER_SRTCP_AUTH_FAIL++;
}else {
PLOG_VERBOSE << "SRTCP unprotect error, status=" << err;
COUNTER_SRTCP_FAIL++;
}
return;
}
PLOG_VERBOSE << "Unprotected SRTCP packet, size=" << size;
message->type = Message::Type::Control;
message->stream = reinterpret_cast<RTCP_SR *>(message->data())->senderSSRC();
} else {
PLOG_VERBOSE << "Incoming SRTP packet, size=" << size;
if (srtp_err_status_t err = srtp_unprotect(mSrtpIn, message->data(), &size)) {
if (err == srtp_err_status_replay_fail)
PLOG_WARNING << "Incoming SRTP packet is a replay";
else if (err == srtp_err_status_auth_fail)
PLOG_WARNING << "Incoming SRTP packet failed authentication check";
else if (err == srtp_err_status_no_ctx) {
auto ssrc = reinterpret_cast<RTP *>(message->data())->ssrc();
PLOG_INFO << "Adding SSRC to RTP: " << ssrc;
addSSRC(ssrc);
} else {
PLOG_WARNING << "SRTP unprotect error, status=" << err
<< " SSRC=" << reinterpret_cast<RTP *>(message->data())->ssrc();
}
if (err == srtp_err_status_replay_fail) {
PLOG_VERBOSE << "Incoming SRTP packet is a replay";
COUNTER_SRTP_REPLAY++;
} else if (err == srtp_err_status_auth_fail) {
PLOG_VERBOSE << "Incoming SRTP packet failed authentication check";
COUNTER_SRTP_AUTH_FAIL++;
} else {
PLOG_VERBOSE << "SRTP unprotect error, status=" << err;
COUNTER_SRTP_FAIL++;
}
return;
}
PLOG_VERBOSE << "Unprotected SRTP packet, size=" << size;
@ -235,7 +231,8 @@ void DtlsSrtpTransport::incoming(message_ptr message) {
mSrtpRecvCallback(message);
} else {
PLOG_WARNING << "Unknown packet type, value=" << unsigned(value1) << ", size=" << size;
COUNTER_UNKNOWN_PACKET_TYPE++;
PLOG_VERBOSE << "Unknown packet type, value=" << unsigned(value1) << ", size=" << size;
}
}
@ -299,37 +296,14 @@ void DtlsSrtpTransport::postHandshake() {
std::memcpy(mServerSessionKey, serverKey, SRTP_AES_128_KEY_LEN);
std::memcpy(mServerSessionKey + SRTP_AES_128_KEY_LEN, serverSalt, SRTP_SALT_LEN);
// Add SSRC=1 as an inbound because that is what Chrome does.
srtp_policy_t inbound = {};
srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtp);
srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtcp);
inbound.ssrc.type = ssrc_specific;
inbound.ssrc.value = 1;
inbound.ssrc.type = ssrc_any_inbound;
inbound.key = mIsClient ? mServerSessionKey : mClientSessionKey;
inbound.window_size = 1024;
inbound.next = nullptr;
if (srtp_err_status_t err = srtp_add_stream(mSrtpIn, &inbound)) {
throw std::runtime_error("SRTP add inbound stream failed, status=" +
to_string(static_cast<int>(err)));
}
mInitDone = true;
}
void DtlsSrtpTransport::addSSRC(uint32_t ssrc) {
if (!mInitDone)
throw std::logic_error("Attempted to add SSRC before SRTP keying material is derived");
srtp_policy_t inbound = {};
srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtp);
srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&inbound.rtcp);
inbound.ssrc.type = ssrc_specific;
inbound.ssrc.value = ssrc;
inbound.key = mIsClient ? mServerSessionKey : mClientSessionKey;
inbound.window_size = 1024;
inbound.next = nullptr;
inbound.allow_repeat_tx = true;
inbound.next = nullptr;
if (srtp_err_status_t err = srtp_add_stream(mSrtpIn, &inbound))
throw std::runtime_error("SRTP add inbound stream failed, status=" +
@ -338,16 +312,17 @@ void DtlsSrtpTransport::addSSRC(uint32_t ssrc) {
srtp_policy_t outbound = {};
srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&outbound.rtp);
srtp_crypto_policy_set_aes_cm_128_hmac_sha1_80(&outbound.rtcp);
outbound.ssrc.type = ssrc_specific;
outbound.ssrc.value = ssrc;
outbound.ssrc.type = ssrc_any_outbound;
outbound.key = mIsClient ? mClientSessionKey : mServerSessionKey;
outbound.window_size = 1024;
outbound.next = nullptr;
outbound.allow_repeat_tx = true;
outbound.next = nullptr;
if (srtp_err_status_t err = srtp_add_stream(mSrtpOut, &outbound))
throw std::runtime_error("SRTP add outbound stream failed, status=" +
to_string(static_cast<int>(err)));
mInitDone = true;
}
} // namespace rtc

1
src/dtlssrtptransport.hpp
View File

@ -45,7 +45,6 @@ public:
~DtlsSrtpTransport();
bool sendMedia(message_ptr message);
void addSSRC(uint32_t ssrc);
private:
void incoming(message_ptr message) override;

41
src/icetransport.cpp
View File

@ -44,6 +44,8 @@ using std::chrono::system_clock;
#if !USE_NICE
#define MAX_TURN_SERVERS_COUNT 2
namespace rtc {
IceTransport::IceTransport(const Configuration &config, candidate_callback candidateCallback,
@ -98,21 +100,39 @@ IceTransport::IceTransport(const Configuration &config, candidate_callback candi
auto seed = static_cast<unsigned int>(system_clock::now().time_since_epoch().count());
std::shuffle(servers.begin(), servers.end(), std::default_random_engine(seed));
// Pick a STUN server (TURN support is not implemented in libjuice yet)
// Pick a STUN server
for (auto &server : servers) {
if (!server.hostname.empty() && server.type == IceServer::Type::Stun) {
if (server.service.empty())
server.service = "3478"; // STUN UDP port
PLOG_DEBUG << "Using STUN server \"" << server.hostname << ":" << server.service
<< "\"";
mStunHostname = server.hostname;
mStunService = server.service;
jconfig.stun_server_host = mStunHostname.c_str();
jconfig.stun_server_port = uint16_t(std::stoul(mStunService));
PLOG_INFO << "Using STUN server \"" << server.hostname << ":" << server.service << "\"";
jconfig.stun_server_host = server.hostname.c_str();
jconfig.stun_server_port = uint16_t(std::stoul(server.service));
break;
}
}
juice_turn_server_t turn_servers[MAX_TURN_SERVERS_COUNT];
std::memset(turn_servers, 0, sizeof(turn_servers));
// Add TURN servers
int k = 0;
for (auto &server : servers) {
if (!server.hostname.empty() && server.type == IceServer::Type::Turn) {
if (server.service.empty())
server.service = "3478"; // TURN UDP port
PLOG_INFO << "Using TURN server \"" << server.hostname << ":" << server.service << "\"";
turn_servers[k].host = server.hostname.c_str();
turn_servers[k].username = server.username.c_str();
turn_servers[k].password = server.password.c_str();
turn_servers[k].port = uint16_t(std::stoul(server.service));
if (++k >= MAX_TURN_SERVERS_COUNT)
break;
}
}
jconfig.turn_servers = k > 0 ? turn_servers : nullptr;
jconfig.turn_servers_count = k;
// Port range
if (config.portRangeBegin > 1024 ||
(config.portRangeEnd != 0 && config.portRangeEnd != 65535)) {
@ -424,8 +444,8 @@ IceTransport::IceTransport(const Configuration &config, candidate_callback candi
if (getnameinfo(p->ai_addr, p->ai_addrlen, nodebuffer, MAX_NUMERICNODE_LEN,
servbuffer, MAX_NUMERICNODE_LEN,
NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
PLOG_DEBUG << "Using STUN server \"" << server.hostname << ":" << server.service
<< "\"";
PLOG_INFO << "Using STUN server \"" << server.hostname << ":" << server.service
<< "\"";
g_object_set(G_OBJECT(mNiceAgent.get()), "stun-server", nodebuffer, nullptr);
g_object_set(G_OBJECT(mNiceAgent.get()), "stun-server-port",
std::stoul(servbuffer), nullptr);
@ -470,7 +490,8 @@ IceTransport::IceTransport(const Configuration &config, candidate_callback candi
if (getnameinfo(p->ai_addr, p->ai_addrlen, nodebuffer, MAX_NUMERICNODE_LEN,
servbuffer, MAX_NUMERICNODE_LEN,
NI_NUMERICHOST | NI_NUMERICSERV) == 0) {
PLOG_INFO << "Using TURN server \"" << server.hostname << ":" << server.service
<< "\"";
NiceRelayType niceRelayType;
switch (server.relayType) {
case IceServer::RelayType::TurnTcp:

2
src/icetransport.hpp
View File

@ -86,8 +86,6 @@ private:
#if !USE_NICE
std::unique_ptr<juice_agent_t, void (*)(juice_agent_t *)> mAgent;
string mStunHostname;
string mStunService;
static void StateChangeCallback(juice_agent_t *agent, juice_state_t state, void *user_ptr);
static void CandidateCallback(juice_agent_t *agent, const char *sdp, void *user_ptr);

40
src/logcounter.cpp Normal file
View File

@ -0,0 +1,40 @@
/**
* Copyright (c) 2021 Staz Modrzynski
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "logcounter.hpp"
rtc::LogCounter::LogCounter(plog::Severity severity, const std::string &text, std::chrono::seconds duration) {
mData = std::make_shared<LogData>();
mData->mDuration =duration;
mData->mSeverity = severity;
mData->mText = text;
}
rtc::LogCounter& rtc::LogCounter::operator++(int) {
if (mData->mCount++ == 0) {
ThreadPool::Instance().schedule(mData->mDuration, [](std::weak_ptr<LogData> data) {
if (auto ptr = data.lock()) {
int countCopy;
countCopy = ptr->mCount.exchange(0);
PLOG(ptr->mSeverity) << ptr->mText << ": " << countCopy << " (over "
<< std::chrono::duration_cast<std::chrono::seconds>(ptr->mDuration).count() << " seconds)";
}
}, mData);
}
return *this;
}

46
src/logcounter.hpp Normal file
View File

@ -0,0 +1,46 @@
/**
* Copyright (c) 2021 Staz Modrzynski
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef WEBRTC_SERVER_LOGCOUNTER_HPP
#define WEBRTC_SERVER_LOGCOUNTER_HPP
#include "threadpool.hpp"
#include "include.hpp"
namespace rtc {
class LogCounter {
private:
struct LogData {
plog::Severity mSeverity;
std::string mText;
std::chrono::steady_clock::duration mDuration;
std::atomic<int> mCount = 0;
};
std::shared_ptr<LogData> mData;
public:
LogCounter(plog::Severity severity, const std::string& text, std::chrono::seconds duration=std::chrono::seconds(1));
LogCounter& operator++(int);
};
}
#endif //WEBRTC_SERVER_LOGCOUNTER_HPP

15
src/peerconnection.cpp
View File

@ -22,6 +22,7 @@
#include "include.hpp"
#include "processor.hpp"
#include "threadpool.hpp"
#include "logcounter.hpp"
#include "dtlstransport.hpp"
#include "icetransport.hpp"
@ -46,6 +47,11 @@ inline std::shared_ptr<To> reinterpret_pointer_cast(std::shared_ptr<From> const
using std::reinterpret_pointer_cast;
#endif
static rtc::LogCounter COUNTER_MEDIA_TRUNCATED(plog::warning, "Number of RTP packets truncated over past second");
static rtc::LogCounter COUNTER_SRTP_DECRYPT_ERROR(plog::warning, "Number of SRTP decryption errors over past second");
static rtc::LogCounter COUNTER_SRTP_ENCRYPT_ERROR(plog::warning, "Number of SRTP encryption errors over past second");
static rtc::LogCounter COUNTER_UNKNOWN_PACKET_TYPE(plog::warning, "Number of unknown RTCP packet types over past second");
namespace rtc {
using namespace std::placeholders;
@ -504,7 +510,7 @@ shared_ptr<DtlsTransport> PeerConnection::initDtlsTransport() {
// DTLS-SRTP
transport = std::make_shared<DtlsSrtpTransport>(
lower, certificate, verifierCallback,
std::bind(&PeerConnection::forwardMedia, this, _1), stateChangeCallback);
weak_bind(&PeerConnection::forwardMedia, this, _1), stateChangeCallback);
#else
PLOG_WARNING << "Ignoring media support (not compiled with media support)";
#endif
@ -678,7 +684,7 @@ void PeerConnection::forwardMedia(message_ptr message) {
while ((sizeof(rtc::RTCP_HEADER) + offset) <= message->size()) {
auto header = reinterpret_cast<rtc::RTCP_HEADER *>(message->data() + offset);
if (header->lengthInBytes() > message->size() - offset) {
PLOG_WARNING << "RTCP packet is truncated";
COUNTER_MEDIA_TRUNCATED++;
break;
}
offset += header->lengthInBytes();
@ -705,8 +711,7 @@ void PeerConnection::forwardMedia(message_ptr message) {
} else {
// PT=207 == Extended Report
if (header->payloadType() != 207) {
PLOG_WARNING << "Unknown packet type: " << (int)header->version() << " "
<< header->payloadType() << "";
COUNTER_UNKNOWN_PACKET_TYPE++;
}
}
}
@ -740,7 +745,7 @@ void PeerConnection::forwardMedia(message_ptr message) {
// PLOG_WARNING << "Track not found for SSRC " << ssrc << ", dropping";
return;
}
} // namespace rtc
}
std::optional<std::string> PeerConnection::getMidFromSsrc(uint32_t ssrc) {
if (auto it = mMidFromSsrc.find(ssrc); it != mMidFromSsrc.end())

17
src/rtcp.cpp
View File

@ -22,6 +22,7 @@
#include "track.hpp"
#include <cmath>
#include <utility>
#include "logcounter.hpp"
#ifdef _WIN32
#include <winsock2.h>
@ -29,6 +30,11 @@
#include <arpa/inet.h>
#endif
static rtc::LogCounter COUNTER_BAD_RTP_HEADER(plog::warning, "Number of malformed RTP headers");
static rtc::LogCounter COUNTER_UNKNOWN_PPID(plog::warning, "Number of Unknown PPID messages");
static rtc::LogCounter COUNTER_BAD_NOTIF_LEN(plog::warning, "Number of Bad-Lengthed notifications");
static rtc::LogCounter COUNTER_BAD_SCTP_STATUS(plog::warning, "Number of unknown SCTP_STATUS errors");
namespace rtc {
rtc::message_ptr RtcpReceivingSession::outgoing(rtc::message_ptr ptr) { return ptr; }
@ -39,20 +45,19 @@ rtc::message_ptr RtcpReceivingSession::incoming(rtc::message_ptr ptr) {
// https://tools.ietf.org/html/rfc3550#appendix-A.1
if (rtp->version() != 2) {
PLOG_WARNING << "RTP packet is not version 2";
COUNTER_BAD_RTP_HEADER++;
PLOG_VERBOSE << "RTP packet is not version 2";
return nullptr;
}
if (rtp->payloadType() == 201 || rtp->payloadType() == 200) {
PLOG_WARNING << "RTP packet has a payload type indicating RR/SR";
COUNTER_BAD_RTP_HEADER++;
PLOG_VERBOSE << "RTP packet has a payload type indicating RR/SR";
return nullptr;
}
// TODO Implement the padding bit
if (rtp->padding()) {
PLOG_WARNING << "Padding processing not implemented";
}
// Padding-processing is a user-level thing
mSsrc = rtp->ssrc();

21
src/sctptransport.cpp
View File

@ -17,6 +17,7 @@
*/
#include "sctptransport.hpp"
#include "logcounter.hpp"
#include <chrono>
#include <exception>
@ -48,6 +49,10 @@ using namespace std::chrono;
using std::shared_ptr;
static rtc::LogCounter COUNTER_UNKNOWN_PPID(plog::warning, "Number of SCTP packets received with an unknown PPID");
static rtc::LogCounter COUNTER_BAD_NOTIF_LEN(plog::warning, "Number of SCTP packets received with an bad notification length");
static rtc::LogCounter COUNTER_BAD_SCTP_STATUS(plog::warning, "Number of SCTP packets received with a bad status");
namespace rtc {
std::unordered_set<SctpTransport *> SctpTransport::Instances;
@ -283,12 +288,12 @@ bool SctpTransport::send(message_ptr message) {
std::lock_guard lock(mSendMutex);
if (!message)
return mSendQueue.empty();
return trySendQueue();
PLOG_VERBOSE << "Send size=" << message->size();
// If nothing is pending, try to send directly
if (mSendQueue.empty() && trySendMessage(message))
// Flush the queue, and if nothing is pending, try to send directly
if (trySendQueue() && trySendMessage(message))
return true;
mSendQueue.push(message);
@ -336,7 +341,7 @@ void SctpTransport::doRecv() {
std::lock_guard lock(mRecvMutex);
--mPendingRecvCount;
try {
while (true) {
while (state() != State::Disconnected && state() != State::Failed) {
const size_t bufferSize = 65536;
byte buffer[bufferSize];
socklen_t fromlen = 0;
@ -631,14 +636,15 @@ void SctpTransport::processData(binary &&data, uint16_t sid, PayloadId ppid) {
default:
// Unknown
PLOG_WARNING << "Unknown PPID: " << uint32_t(ppid);
COUNTER_UNKNOWN_PPID++;
PLOG_VERBOSE << "Unknown PPID: " << uint32_t(ppid);
return;
}
}
void SctpTransport::processNotification(const union sctp_notification *notify, size_t len) {
if (len != size_t(notify->sn_header.sn_length)) {
PLOG_WARNING << "Invalid notification length";
COUNTER_BAD_NOTIF_LEN++;
return;
}
@ -738,7 +744,8 @@ std::optional<milliseconds> SctpTransport::rtt() {
struct sctp_status status = {};
socklen_t len = sizeof(status);
if (usrsctp_getsockopt(mSock, IPPROTO_SCTP, SCTP_STATUS, &status, &len)) {
PLOG_WARNING << "Could not read SCTP_STATUS";
COUNTER_BAD_SCTP_STATUS++;
return nullopt;
}
return milliseconds(status.sstat_primary.spinfo_srtt);

7
src/tcptransport.cpp
View File

@ -113,7 +113,7 @@ bool TcpTransport::send(message_ptr message) {
return false;
if (!message)
return mSendQueue.empty();
return trySendQueue();
PLOG_VERBOSE << "Send size=" << (message ? message->size() : 0);
return outgoing(message);
@ -129,9 +129,8 @@ void TcpTransport::incoming(message_ptr message) {
bool TcpTransport::outgoing(message_ptr message) {
// mSockMutex must be locked
// If nothing is pending, try to send directly
// It's safe because if the queue is empty, the thread is not sending
if (mSendQueue.empty() && trySendMessage(message))
// Flush the queue, and if nothing is pending, try to send directly
if (trySendQueue() && trySendMessage(message))
return true;
mSendQueue.push(message);

26
src/threadpool.cpp
View File

@ -66,12 +66,26 @@ bool ThreadPool::runOne() {
std::function<void()> ThreadPool::dequeue() {
std::unique_lock lock(mMutex);
mCondition.wait(lock, [this]() { return !mTasks.empty() || mJoining; });
if (mTasks.empty())
return nullptr;
auto task = std::move(mTasks.front());
mTasks.pop();
return task;
while (true) {
if (!mTasks.empty()) {
if (mTasks.top().time <= clock::now()) {
auto func = std::move(mTasks.top().func);
mTasks.pop();
return func;
}
if (mJoining)
break;
mCondition.wait_until(lock, mTasks.top().time);
} else {
if (mJoining)
break;
mCondition.wait(lock);
}
}
return nullptr;
}
} // namespace rtc

37
src/threadpool.hpp
View File

@ -22,7 +22,9 @@
#include "include.hpp"
#include "init.hpp"
#include <chrono>
#include <condition_variable>
#include <deque>
#include <functional>
#include <future>
#include <memory>
@ -39,6 +41,8 @@ using invoke_future_t = std::future<std::invoke_result_t<std::decay_t<F>, std::d
class ThreadPool final {
public:
using clock = std::chrono::steady_clock;
static ThreadPool &Instance();
ThreadPool(const ThreadPool &) = delete;
@ -53,7 +57,13 @@ public:
bool runOne();
template <class F, class... Args>
auto enqueue(F &&f, Args &&... args) -> invoke_future_t<F, Args...>;
auto enqueue(F &&f, Args &&...args) -> invoke_future_t<F, Args...>;
template <class F, class... Args>
auto schedule(clock::duration delay, F &&f, Args &&...args) -> invoke_future_t<F, Args...>;
template <class F, class... Args>
auto schedule(clock::time_point time, F &&f, Args &&...args) -> invoke_future_t<F, Args...>;
protected:
ThreadPool() = default;
@ -62,15 +72,34 @@ protected:
std::function<void()> dequeue(); // returns null function if joining
std::vector<std::thread> mWorkers;
std::queue<std::function<void()>> mTasks;
std::atomic<bool> mJoining = false;
struct Task {
clock::time_point time;
std::function<void()> func;
bool operator>(const Task &other) const { return time > other.time; }
bool operator<(const Task &other) const { return time < other.time; }
};
std::priority_queue<Task, std::deque<Task>, std::greater<Task>> mTasks;
mutable std::mutex mMutex, mWorkersMutex;
std::condition_variable mCondition;
};
template <class F, class... Args>
auto ThreadPool::enqueue(F &&f, Args &&... args) -> invoke_future_t<F, Args...> {
auto ThreadPool::enqueue(F &&f, Args &&...args) -> invoke_future_t<F, Args...> {
return schedule(clock::now(), std::forward<F>(f), std::forward<Args>(args)...);
}
template <class F, class... Args>
auto ThreadPool::schedule(clock::duration delay, F &&f, Args &&...args)
-> invoke_future_t<F, Args...> {
return schedule(clock::now() + delay, std::forward<F>(f), std::forward<Args>(args)...);
}
template <class F, class... Args>
auto ThreadPool::schedule(clock::time_point time, F &&f, Args &&...args)
-> invoke_future_t<F, Args...> {
std::unique_lock lock(mMutex);
using R = std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>;
auto bound = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
@ -84,7 +113,7 @@ auto ThreadPool::enqueue(F &&f, Args &&... args) -> invoke_future_t<F, Args...>
});
std::future<R> result = task->get_future();
mTasks.emplace([task = std::move(task), token = Init::Token()]() { return (*task)(); });
mTasks.push({time, [task = std::move(task), token = Init::Token()]() { return (*task)(); }});
mCondition.notify_one();
return result;
}

10
src/track.cpp
View File

@ -19,6 +19,10 @@
#include "track.hpp"
#include "dtlssrtptransport.hpp"
#include "include.hpp"
#include "logcounter.hpp"
static rtc::LogCounter COUNTER_MEDIA_BAD_DIRECTION(plog::warning, "Number of media packets sent in invalid directions");
static rtc::LogCounter COUNTER_QUEUE_FULL(plog::warning, "Number of media packets dropped due to a full queue");
namespace rtc {
@ -52,7 +56,7 @@ bool Track::send(message_variant data) {
auto direction = mMediaDescription.direction();
if ((direction == Description::Direction::RecvOnly ||
direction == Description::Direction::Inactive)) {
PLOG_WARNING << "Track media direction does not allow transmission, dropping";
COUNTER_MEDIA_BAD_DIRECTION++;
return false;
}
@ -114,7 +118,7 @@ void Track::incoming(message_ptr message) {
if ((direction == Description::Direction::SendOnly ||
direction == Description::Direction::Inactive) &&
message->type != Message::Control) {
PLOG_WARNING << "Track media direction does not allow reception, dropping";
COUNTER_MEDIA_BAD_DIRECTION++;
return;
}
@ -126,7 +130,7 @@ void Track::incoming(message_ptr message) {
// Tail drop if queue is full
if (mRecvQueue.full()) {
PLOG_WARNING << "Track incoming queue is full, dropping";
COUNTER_QUEUE_FULL++;
return;
}

10
test/connectivity.cpp
View File

@ -33,14 +33,16 @@ void test_connectivity() {
InitLogger(LogLevel::Debug);
Configuration config1;
// STUN server example
// config1.iceServers.emplace_back("stun:stun.l.google.com:19302");
// STUN server example (not necessary to connect locally)
// Please do not use outside of libdatachannel tests
config1.iceServers.emplace_back("stun:stun.ageneau.net:3478");
auto pc1 = std::make_shared<PeerConnection>(config1);
Configuration config2;
// STUN server example
// config2.iceServers.emplace_back("stun:stun.l.google.com:19302");
// STUN server example (not necessary to connect locally)
// Please do not use outside of libdatachannel tests
config2.iceServers.emplace_back("stun:stun.ageneau.net:3478");
// Port range example
config2.portRangeBegin = 5000;
config2.portRangeEnd = 6000;

10
test/main.cpp
View File

@ -24,6 +24,7 @@ using namespace std;
using namespace chrono_literals;
void test_connectivity();
void test_turn_connectivity();
void test_track();
void test_capi_connectivity();
void test_capi_track();
@ -51,6 +52,15 @@ int main(int argc, char **argv) {
return -1;
}
this_thread::sleep_for(1s);
try {
cout << endl << "*** Running WebRTC TURN connectivity test..." << endl;
test_turn_connectivity();
cout << "*** Finished WebRTC TURN connectivity test" << endl;
} catch (const exception &e) {
cerr << "WebRTC TURN connectivity test failed: " << e.what() << endl;
return -1;
}
this_thread::sleep_for(1s);
try {
cout << endl << "*** Running WebRTC C API connectivity test..." << endl;
test_capi_connectivity();

266
test/turn_connectivity.cpp Normal file
View File

@ -0,0 +1,266 @@
/**
* Copyright (c) 2019 Paul-Louis Ageneau
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "rtc/rtc.hpp"
#include <atomic>
#include <chrono>
#include <iostream>
#include <memory>
#include <thread>
using namespace rtc;
using namespace std;
template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
void test_turn_connectivity() {
InitLogger(LogLevel::Debug);
Configuration config1;
// STUN server example (not necessary, just here for testing)
// Please do not use outside of libdatachannel tests
config1.iceServers.emplace_back("stun:stun.ageneau.net:3478");
// TURN server example
// Please do not use outside of libdatachannel tests
config1.iceServers.emplace_back("turn:datachannel_test:14018314739877@stun.ageneau.net:3478");
auto pc1 = std::make_shared<PeerConnection>(config1);
Configuration config2;
// STUN server example (not necessary, just here for testing)
// Please do not use outside of libdatachannel tests
config1.iceServers.emplace_back("stun:stun.ageneau.net:3478");
// TURN server example
// Please do not use outside of libdatachannel tests
config2.iceServers.emplace_back("turn:datachannel_test:14018314739877@stun.ageneau.net:3478");
auto pc2 = std::make_shared<PeerConnection>(config2);
pc1->onLocalDescription([wpc2 = make_weak_ptr(pc2)](Description sdp) {
auto pc2 = wpc2.lock();
if (!pc2)
return;
cout << "Description 1: " << sdp << endl;
pc2->setRemoteDescription(string(sdp));
});
pc1->onLocalCandidate([wpc2 = make_weak_ptr(pc2)](Candidate candidate) {
auto pc2 = wpc2.lock();
if (!pc2)
return;
// For this test, filter out non-relay candidates to force TURN
string str(candidate);
if(str.find("relay") != string::npos) {
cout << "Candidate 1: " << str << endl;
pc2->addRemoteCandidate(str);
}
});
pc1->onStateChange([](PeerConnection::State state) { cout << "State 1: " << state << endl; });
pc1->onGatheringStateChange([](PeerConnection::GatheringState state) {
cout << "Gathering state 1: " << state << endl;
});
pc1->onSignalingStateChange([](PeerConnection::SignalingState state) {
cout << "Signaling state 1: " << state << endl;
});
pc2->onLocalDescription([wpc1 = make_weak_ptr(pc1)](Description sdp) {
auto pc1 = wpc1.lock();
if (!pc1)
return;
cout << "Description 2: " << sdp << endl;
pc1->setRemoteDescription(string(sdp));
});
pc2->onLocalCandidate([wpc1 = make_weak_ptr(pc1)](Candidate candidate) {
auto pc1 = wpc1.lock();
if (!pc1)
return;
// For this test, filter out non-relay candidates to force TURN
string str(candidate);
if(str.find("relay") != string::npos) {
cout << "Candidate 1: " << str << endl;
pc1->addRemoteCandidate(str);
}
});
pc2->onStateChange([](PeerConnection::State state) { cout << "State 2: " << state << endl; });
pc2->onGatheringStateChange([](PeerConnection::GatheringState state) {
cout << "Gathering state 2: " << state << endl;
});
pc2->onSignalingStateChange([](PeerConnection::SignalingState state) {
cout << "Signaling state 2: " << state << endl;
});
shared_ptr<DataChannel> dc2;
pc2->onDataChannel([&dc2](shared_ptr<DataChannel> dc) {
cout << "DataChannel 2: Received with label \"" << dc->label() << "\"" << endl;
if (dc->label() != "test") {
cerr << "Wrong DataChannel label" << endl;
return;
}
dc->onMessage([](variant<binary, string> message) {
if (holds_alternative<string>(message)) {
cout << "Message 2: " << get<string>(message) << endl;
}
});
dc->send("Hello from 2");
std::atomic_store(&dc2, dc);
});
auto dc1 = pc1->createDataChannel("test");
dc1->onOpen([wdc1 = make_weak_ptr(dc1)]() {
auto dc1 = wdc1.lock();
if (!dc1)
return;
cout << "DataChannel 1: Open" << endl;
dc1->send("Hello from 1");
});
dc1->onMessage([](const variant<binary, string> &message) {
if (holds_alternative<string>(message)) {
cout << "Message 1: " << get<string>(message) << endl;
}
});
// Wait a bit
int attempts = 10;
shared_ptr<DataChannel> adc2;
while ((!(adc2 = std::atomic_load(&dc2)) || !adc2->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 (!adc2 || !adc2->isOpen() || !dc1->isOpen())
throw runtime_error("DataChannel is not open");
if (auto addr = pc1->localAddress())
cout << "Local address 1: " << *addr << endl;
if (auto addr = pc1->remoteAddress())
cout << "Remote address 1: " << *addr << endl;
if (auto addr = pc2->localAddress())
cout << "Local address 2: " << *addr << endl;
if (auto addr = pc2->remoteAddress())
cout << "Remote address 2: " << *addr << endl;
Candidate local, remote;
if (pc1->getSelectedCandidatePair(&local, &remote)) {
cout << "Local candidate 1: " << local << endl;
cout << "Remote candidate 1: " << remote << endl;
}
if (pc2->getSelectedCandidatePair(&local, &remote)) {
cout << "Local candidate 2: " << local << endl;
cout << "Remote candidate 2: " << remote << endl;
}
// Try to open a second data channel with another label
shared_ptr<DataChannel> second2;
pc2->onDataChannel([&second2](shared_ptr<DataChannel> dc) {
cout << "Second DataChannel 2: Received with label \"" << dc->label() << "\"" << endl;
if (dc->label() != "second") {
cerr << "Wrong second DataChannel label" << endl;
return;
}
dc->onMessage([](variant<binary, string> message) {
if (holds_alternative<string>(message)) {
cout << "Second Message 2: " << get<string>(message) << endl;
}
});
dc->send("Send hello from 2");
std::atomic_store(&second2, dc);
});
auto second1 = pc1->createDataChannel("second");
second1->onOpen([wsecond1 = make_weak_ptr(dc1)]() {
auto second1 = wsecond1.lock();
if (!second1)
return;
cout << "Second DataChannel 1: Open" << endl;
second1->send("Second hello from 1");
});
dc1->onMessage([](const variant<binary, string> &message) {
if (holds_alternative<string>(message)) {
cout << "Second Message 1: " << get<string>(message) << endl;
}
});
// Wait a bit
attempts = 10;
shared_ptr<DataChannel> asecond2;
while (
(!(asecond2 = std::atomic_load(&second2)) || !asecond2->isOpen() || !second1->isOpen()) &&
attempts--)
this_thread::sleep_for(1s);
if (!asecond2 || !asecond2->isOpen() || !second1->isOpen())
throw runtime_error("Second DataChannel is not open");
// Try to open a negotiated channel
DataChannelInit init;
init.negotiated = true;
init.id = 42;
auto negotiated1 = pc1->createDataChannel("negotiated", init);
auto negotiated2 = pc2->createDataChannel("negoctated", init);
if (!negotiated1->isOpen() || !negotiated2->isOpen())
throw runtime_error("Negociated DataChannel is not open");
std::atomic<bool> received = false;
negotiated2->onMessage([&received](const variant<binary, string> &message) {
if (holds_alternative<string>(message)) {
cout << "Second Message 2: " << get<string>(message) << endl;
received = true;
}
});
negotiated1->send("Hello from negotiated channel");
// Wait a bit
attempts = 5;
while (!received && attempts--)
this_thread::sleep_for(1s);
if (!received)
throw runtime_error("Negociated DataChannel failed");
// Delay close of peer 2 to check closing works properly
pc1->close();
this_thread::sleep_for(1s);
pc2->close();
this_thread::sleep_for(1s);
// You may call rtc::Cleanup() when finished to free static resources
rtc::Cleanup();
this_thread::sleep_for(1s);
cout << "Success" << endl;
}