Bumped version to 0.9.3

Fix OpenSSL recv loop quitting too early

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required(VERSION 3.7)
 project(libdatachannel
 	DESCRIPTION "WebRTC Data Channels Library"
-	VERSION 0.9.1
+	VERSION 0.9.3
 	LANGUAGES CXX)
 
 # Options

examples/client/README-getopt-for-windows.md

@@ -1,4 +0,0 @@
-# getopt-for-windows
-getopt.h and getopt.c is very often used in linux, to make it easy for windows user, two files were extracted from glibc. In order to make it works properly in windows, some modification was done and you may compare the change using original source files. Enjoy it!
-
-Source: https://github.com/Chunde/getopt-for-windows.  IMPORTANT: getopt.[ch] are likely not safe for Linux due to conflict with existing getopt.[ch].  They are thus NOT in CMakeFiles.txt and instead both files are #include only on Windows.

examples/client/main.cpp

@@ -66,12 +66,16 @@ int main(int argc, char **argv) {
 
 	Configuration config;
 	string stunServer = "";
+	if (params->noStun()) {
+		cout << "No stun server is configured.  Only local hosts and public IP addresses suported." << endl;
+	} else {
 		if (params->stunServer().substr(0,5).compare("stun:") != 0) {
 			stunServer = "stun:";
 		}
 		stunServer += params->stunServer() + ":" + to_string(params->stunPort());
 		cout << "Stun server is " << stunServer << endl;
 		config.iceServers.emplace_back(stunServer);
+	}
 
 	localId = randomId(4);
 	cout << "The local ID is: " << localId << endl;

examples/client/parse_cl.cpp

@@ -43,6 +43,8 @@ Cmdline::Cmdline (int argc, char *argv[]) // ISO C++17 not allowed: throw (std::
 
   static struct option long_options[] =
   {
+    {"echo", no_argument, NULL, 'e'},
+    {"noStun", no_argument, NULL, 'n'},
     {"stunServer", required_argument, NULL, 's'},
     {"stunPort", required_argument, NULL, 't'},
     {"webSocketServer", required_argument, NULL, 'w'},
@@ -55,19 +57,28 @@ Cmdline::Cmdline (int argc, char *argv[]) // ISO C++17 not allowed: throw (std::
   _program_name += argv[0];
 
   /* default values */
+  _e = false;
+  _n = false;
   _s = "stun.l.google.com";
   _t = 19302;
   _w = "localhost";
   _x = 8000;
-  _e = false;
   _h = false;
   _v = false;
 
   optind = 0;
-  while ((c = getopt_long (argc, argv, "s:t:w:x:ehv", long_options, &optind)) != - 1)
+  while ((c = getopt_long (argc, argv, "s:t:w:x:enhv", long_options, &optind)) != - 1)
     {
       switch (c)
         {
+        case 'e': 
+          _e = true;
+          break;
+
+        case 'n': 
+          _n = true;
+          break;
+
         case 's': 
           _s = optarg;
           break;
@@ -108,10 +119,6 @@ Cmdline::Cmdline (int argc, char *argv[]) // ISO C++17 not allowed: throw (std::
             }
           break;
 
-        case 'e': 
-          _e = true;
-          break;
-
         case 'h': 
           _h = true;
           this->usage (EXIT_SUCCESS);
@@ -146,10 +153,12 @@ void Cmdline::usage (int status)
   else
     {
       std::cout << "\
-usage: " << _program_name << " [ -estwxhv ] \n\
+usage: " << _program_name << " [ -enstwxhv ] \n\
 libdatachannel client implementing WebRTC Data Channels with WebSocket signaling\n\
    [ -e ] [ --echo ] (type=FLAG)\n\
           Echo data channel messages back to sender rather than putting to stdout.\n\
+   [ -n ] [ --noStun ] (type=FLAG)\n\
+          Do NOT use a stun server (overrides -s and -t).\n\
    [ -s ] [ --stunServer ] (type=STRING, default=stun.l.google.com)\n\
           Stun server URL or IP address.\n\
    [ -t ] [ --stunPort ] (type=INTEGER, range=0...65535, default=19302)\n\

examples/client/parse_cl.h

@@ -34,11 +34,12 @@ class Cmdline
 {
 private:
   /* parameters */
+  bool _e;
+  bool _n;
   std::string _s;
   int _t;
   std::string _w;
   int _x;
-  bool _e;
   bool _h;
   bool _v;
 
@@ -60,11 +61,12 @@ public:
   /* return next (non-option) parameter */
   int next_param () { return _optind; }
 
+  bool echoDataChannelMessages () const { return _e; }
+  bool noStun () const { return _n; }
   std::string stunServer () const { return _s; }
   int stunPort () const { return _t; }
   std::string webSocketServer () const { return _w; }
   int webSocketPort () const { return _x; }
-  bool echoDataChannelMessages () const { return _e; }
   bool h () const { return _h; }
   bool v () const { return _v; }
 };

include/rtc/queue.hpp

@@ -38,6 +38,7 @@ public:
 	~Queue();
 
 	void stop();
+	bool running() const;
 	bool empty() const;
 	bool full() const;
 	size_t size() const;   // elements
@@ -80,6 +81,11 @@ template <typename T> void Queue<T>::stop() {
 	mPushCondition.notify_all();
 }
 
+template <typename T> bool Queue<T>::running() const {
+	std::lock_guard lock(mMutex);
+	return !mQueue.empty() || !mStopping;
+}
+
 template <typename T> bool Queue<T>::empty() const {
 	std::lock_guard lock(mMutex);
 	return mQueue.empty();

src/candidate.cpp

@@ -94,7 +94,7 @@ bool Candidate::resolve(ResolveMode mode) {
 
 		struct addrinfo *result = nullptr;
 		if (getaddrinfo(node.c_str(), service.c_str(), &hints, &result) == 0) {
-			for (auto p = result; p; p = p->ai_next)
+			for (auto p = result; p; p = p->ai_next) {
 				if (p->ai_family == AF_INET || p->ai_family == AF_INET6) {
 					// Rewrite the candidate
 					char nodebuffer[MAX_NUMERICNODE_LEN];
@@ -117,6 +117,7 @@ bool Candidate::resolve(ResolveMode mode) {
 
 			freeaddrinfo(result);
 		}
+	}
 
 	return mIsResolved;
 }

src/datachannel.cpp

@@ -218,6 +218,8 @@ void DataChannel::incoming(message_ptr message) {
 
 	switch (message->type) {
 	case Message::Control: {
+		if (message->size() == 0)
+			break; // Ignore
 		auto raw = reinterpret_cast<const uint8_t *>(message->data());
 		switch (raw[0]) {
 		case MESSAGE_OPEN:

src/dtlstransport.cpp

@@ -435,7 +435,7 @@ void DtlsTransport::runRecvLoop() {
 
 		const size_t bufferSize = maxMtu;
 		byte buffer[bufferSize];
-		while (true) {
+		while (mIncomingQueue.running()) {
 			// Process pending messages
 			while (auto next = mIncomingQueue.tryPop()) {
 				message_ptr message = std::move(*next);
@@ -492,8 +492,7 @@ void DtlsTransport::runRecvLoop() {
 				}
 			}
 
-			if (!mIncomingQueue.wait(duration))
+			mIncomingQueue.wait(duration);
-				break; // queue is stopped
 		}
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "DTLS recv: " << e.what();

src/icetransport.cpp

@@ -380,8 +380,11 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 		hints.ai_protocol = IPPROTO_UDP;
 		hints.ai_flags = AI_ADDRCONFIG;
 		struct addrinfo *result = nullptr;
-		if (getaddrinfo(server.hostname.c_str(), server.service.c_str(), &hints, &result) != 0)
+		if (getaddrinfo(server.hostname.c_str(), server.service.c_str(), &hints, &result) != 0) {
+			PLOG_WARNING << "Unable to resolve STUN server address: " << server.hostname << ':'
+			             << server.service;
 			continue;
+		}
 
 		for (auto p = result; p; p = p->ai_next) {
 			if (p->ai_family == AF_INET) {
@@ -423,8 +426,11 @@ IceTransport::IceTransport(const Configuration &config, Description::Role role,
 		    server.relayType == IceServer::RelayType::TurnUdp ? IPPROTO_UDP : IPPROTO_TCP;
 		hints.ai_flags = AI_ADDRCONFIG;
 		struct addrinfo *result = nullptr;
-		if (getaddrinfo(server.hostname.c_str(), server.service.c_str(), &hints, &result) != 0)
+		if (getaddrinfo(server.hostname.c_str(), server.service.c_str(), &hints, &result) != 0) {
+			PLOG_WARNING << "Unable to resolve TURN server address: " << server.hostname << ':'
+			             << server.service;
 			continue;
+		}
 
 		for (auto p = result; p; p = p->ai_next) {
 			if (p->ai_family == AF_INET || p->ai_family == AF_INET6) {

src/peerconnection.cpp

@@ -102,6 +102,7 @@ void PeerConnection::setLocalDescription() {
 
 	if (std::atomic_load(&mIceTransport)) {
 		PLOG_DEBUG << "Local description is already set, ignoring";
+		return;
 	}
 
 	// RFC 5763: The endpoint that is the offerer MUST use the setup attribute value of
@@ -546,17 +547,17 @@ void PeerConnection::forwardMessage(message_ptr message) {
 	}
 
 	auto channel = findDataChannel(uint16_t(message->stream));
-
+	if (!channel) {
 		auto iceTransport = std::atomic_load(&mIceTransport);
 		auto sctpTransport = std::atomic_load(&mSctpTransport);
 		if (!iceTransport || !sctpTransport)
 			return;
 
-	if (!channel) {
 		const byte dataChannelOpenMessage{0x03};
 		unsigned int remoteParity = (iceTransport->role() == Description::Role::Active) ? 1 : 0;
 		if (message->type == Message::Control && *message->data() == dataChannelOpenMessage &&
 		    message->stream % 2 == remoteParity) {
+
 			channel =
 			    std::make_shared<DataChannel>(shared_from_this(), sctpTransport, message->stream);
 			channel->onOpen(weak_bind(&PeerConnection::triggerDataChannel, this,

src/rtcp.cpp

@@ -71,24 +71,24 @@ private:
 	uint32_t _delaySinceLastReport;
 
 public:
-	inline void preparePacket(SSRC ssrc, [[maybe_unused]] unsigned int packetsLost,
+	inline void preparePacket(SSRC ssrc_, [[maybe_unused]] unsigned int packetsLost,
 	                          [[maybe_unused]] unsigned int totalPackets, uint16_t highestSeqNo,
 	                          uint16_t seqNoCycles, uint32_t jitter, uint64_t lastSR_NTP,
 	                          uint64_t lastSR_DELAY) {
 		setSeqNo(highestSeqNo, seqNoCycles);
 		setJitter(jitter);
-		setSSRC(ssrc);
+		setSSRC(ssrc_);
 
 		// Middle 32 bits of NTP Timestamp
-		//		  this->lastReport = lastSR_NTP >> 16u;
+		// _lastReport = lastSR_NTP >> 16u;
 		setNTPOfSR(uint32_t(lastSR_NTP));
 		setDelaySinceSR(uint32_t(lastSR_DELAY));
 
 		// The delay, expressed in units of 1/65536 seconds
-		//		  this->delaySinceLastReport = lastSR_DELAY;
+		// _delaySinceLastReport = lastSR_DELAY;
 	}
 
-	inline void setSSRC(SSRC ssrc) { this->ssrc = htonl(ssrc); }
+	inline void setSSRC(SSRC ssrc_) { ssrc = htonl(ssrc_); }
 	inline SSRC getSSRC() const { return ntohl(ssrc); }
 
 	inline void setPacketsLost([[maybe_unused]] unsigned int packetsLost,
@@ -172,7 +172,7 @@ public:
 
 struct RTCP_SR {
 	RTCP_HEADER header;
-	SSRC senderSSRC;
+	SSRC senderSsrc;
 
 private:
 	uint64_t _ntpTimestamp;
@@ -183,11 +183,11 @@ private:
 	RTCP_ReportBlock _reportBlocks;
 
 public:
-	inline void preparePacket(SSRC senderSSRC, uint8_t reportCount) {
+	inline void preparePacket(SSRC senderSsrc_, uint8_t reportCount) {
 		unsigned int length =
 		    ((sizeof(header) + 24 + reportCount * sizeof(RTCP_ReportBlock)) / 4) - 1;
 		header.prepareHeader(200, reportCount, uint16_t(length));
-		this->senderSSRC = htonl(senderSSRC);
+		senderSsrc = htonl(senderSsrc_);
 	}
 
 	inline RTCP_ReportBlock *getReportBlock(int num) { return &_reportBlocks + num; }
@@ -209,7 +209,7 @@ public:
 	inline void log() const {
 		header.log();
 		PLOG_DEBUG << "RTCP SR: "
-		           << " SSRC=" << ntohl(senderSSRC) << ", NTP_TS=" << ntpTimestamp()
+		           << " SSRC=" << ntohl(senderSsrc) << ", NTP_TS=" << ntpTimestamp()
 		           << ", RTP_TS=" << rtpTimestamp() << ", packetCount=" << packetCount()
 		           << ", octetCount=" << octetCount();
 
@@ -221,7 +221,7 @@ public:
 
 struct RTCP_RR {
 	RTCP_HEADER header;
-	SSRC senderSSRC;
+	SSRC senderSsrc;
 
 private:
 	RTCP_ReportBlock _reportBlocks;
@@ -230,19 +230,19 @@ public:
 	inline RTCP_ReportBlock *getReportBlock(int num) { return &_reportBlocks + num; }
 	inline const RTCP_ReportBlock *getReportBlock(int num) const { return &_reportBlocks + num; }
 
-	inline SSRC getSenderSSRC() const { return ntohl(senderSSRC); }
+	inline SSRC getSenderSSRC() const { return ntohl(senderSsrc); }
-	inline void setSenderSSRC(SSRC ssrc) { this->senderSSRC = htonl(ssrc); }
+	inline void setSenderSSRC(SSRC ssrc) { senderSsrc = htonl(ssrc); }
 
 	[[nodiscard]] inline size_t getSize() const {
 		// "length" in packet is one less than the number of 32 bit words in the packet.
 		return sizeof(uint32_t) * (1 + size_t(header.length()));
 	}
 
-	inline void preparePacket(SSRC senderSSRC, uint8_t reportCount) {
+	inline void preparePacket(SSRC ssrc, uint8_t reportCount) {
 		// "length" in packet is one less than the number of 32 bit words in the packet.
 		size_t length = (sizeWithReportBlocks(reportCount) / 4) - 1;
 		header.prepareHeader(201, reportCount, uint16_t(length));
-		this->senderSSRC = htonl(senderSSRC);
+		senderSsrc = htonl(ssrc);
 	}
 
 	inline static size_t sizeWithReportBlocks(uint8_t reportCount) {
@@ -252,7 +252,7 @@ public:
 	inline void log() const {
 		header.log();
 		PLOG_DEBUG << "RTCP RR: "
-		           << " SSRC=" << ntohl(senderSSRC);
+		           << " SSRC=" << ntohl(senderSsrc);
 
 		for (unsigned i = 0; i < unsigned(header.reportCount()); i++) {
 			getReportBlock(i)->log();
@@ -262,7 +262,7 @@ public:
 
 struct RTCP_REMB {
 	RTCP_HEADER header;
-	SSRC senderSSRC;
+	SSRC senderSsrc;
 	SSRC mediaSourceSSRC;
 
 	// Unique identifier
@@ -278,48 +278,48 @@ struct RTCP_REMB {
 		return sizeof(uint32_t) * (1 + size_t(header.length()));
 	}
 
-	inline void preparePacket(SSRC senderSSRC, unsigned int numSSRC, unsigned int bitrate) {
+	inline void preparePacket(SSRC senderSsrc_, unsigned int numSSRC, unsigned int br) {
 		// Report Count becomes the format here.
 		header.prepareHeader(206, 15, 0);
 
 		// Always zero.
 		mediaSourceSSRC = 0;
 
-		this->senderSSRC = htonl(senderSSRC);
+		senderSsrc = htonl(senderSsrc_);
-		setBitrate(numSSRC, bitrate);
+		setBitrate(numSSRC, br);
 	}
 
-	inline void setBitrate(unsigned int numSSRC, unsigned int bitrate) {
+	inline void setBitrate(unsigned int numSSRC, unsigned int br) {
 		unsigned int exp = 0;
-		while (bitrate > pow(2, 18) - 1) {
+		while (br > pow(2, 18) - 1) {
 			exp++;
-			bitrate /= 2;
+			br /= 2;
 		}
 
 		// "length" in packet is one less than the number of 32 bit words in the packet.
 		header.setLength(uint16_t(((sizeof(header) + 4 * 2 + 4 + 4) / 4) - 1 + numSSRC));
 
-		this->bitrate = htonl((numSSRC << (32u - 8u)) | (exp << (32u - 8u - 6u)) | bitrate);
+		bitrate = htonl((numSSRC << (32u - 8u)) | (exp << (32u - 8u - 6u)) | br);
 	}
 
 	// TODO Make this work
 	//	  uint64_t getBitrate() const{
-	//		  uint32_t ntohed = ntohl(this->bitrate);
+	//		  uint32_t ntohed = ntohl(bitrate);
 	//		  uint64_t bitrate = ntohed & (unsigned int)(pow(2, 18)-1);
 	//		  unsigned int exp = ntohed & ((unsigned int)( (pow(2, 6)-1)) << (32u-8u-6u));
 	//		  return bitrate * pow(2,exp);
 	//	  }
 	//
 	//	  uint8_t getNumSSRCS() const {
-	//		  return ntohl(this->bitrate) & (((unsigned int) pow(2,8)-1) << (32u-8u));
+	//		  return ntohl(bitrate) & (((unsigned int) pow(2,8)-1) << (32u-8u));
 	//	  }
 
-	inline void setSSRC(uint8_t iterator, SSRC ssrc) { this->ssrc[iterator] = htonl(ssrc); }
+	inline void setSSRC(uint8_t iterator, SSRC ssrc_) { ssrc[iterator] = htonl(ssrc_); }
 
 	inline void log() const {
 		header.log();
 		PLOG_DEBUG << "RTCP REMB: "
-		           << " SSRC=" << ntohl(senderSSRC);
+		           << " SSRC=" << ntohl(senderSsrc);
 	}
 
 	static unsigned int sizeWithSSRCs(int numSSRC) {
@@ -407,7 +407,7 @@ void RtcpSession::pushRR(unsigned int lastSR_delay) {
 	auto msg = rtc::make_message(RTCP_RR::sizeWithReportBlocks(1), rtc::Message::Type::Control);
 	auto rr = reinterpret_cast<RTCP_RR *>(msg->data());
 	rr->preparePacket(mSsrc, 1);
-	rr->getReportBlock(0)->preparePacket(mSsrc, 0, 0, mGreatestSeqNo, 0, 0, mSyncNTPTS,
+	rr->getReportBlock(0)->preparePacket(mSsrc, 0, 0, uint16_t(mGreatestSeqNo), 0, 0, mSyncNTPTS,
 	                                     lastSR_delay);
 	rr->log();
 

src/sctptransport.cpp

@@ -476,8 +476,6 @@ int SctpTransport::handleRecv(struct socket * /*sock*/, union sctp_sockstore /*a
                               const byte *data, size_t len, struct sctp_rcvinfo info, int flags) {
 	try {
 		PLOG_VERBOSE << "Handle recv, len=" << len;
-		if (!len)
-			return 0; // Ignore
 
 		// SCTP_FRAGMENT_INTERLEAVE does not seem to work as expected for messages > 64KB,
 		// therefore partial notifications and messages need to be handled separately.
@@ -497,7 +495,7 @@ int SctpTransport::handleRecv(struct socket * /*sock*/, union sctp_sockstore /*a
 			if (flags & MSG_EOR) {
 				// Message is complete, process it
 				processData(std::move(mPartialMessage), info.rcv_sid,
-				            PayloadId(htonl(info.rcv_ppid)));
+				            PayloadId(ntohl(info.rcv_ppid)));
 				mPartialMessage.clear();
 			}
 		}

src/websocket.cpp

@@ -133,6 +133,11 @@ bool WebSocket::outgoing(message_ptr message) {
 }
 
 void WebSocket::incoming(message_ptr message) {
+	if (!message) {
+		remoteClose();
+		return;
+	}
+
 	if (message->type == Message::String || message->type == Message::Binary) {
 		mRecvQueue.push(message);
 		triggerAvailable(mRecvQueue.size());

test/connectivity.cpp

@@ -125,6 +125,7 @@ void test_connectivity() {
 		}
 	});
 
+	// Wait a bit
 	int attempts = 10;
 	shared_ptr<DataChannel> adc2;
 	while ((!(adc2 = std::atomic_load(&dc2)) || !adc2->isOpen() || !dc1->isOpen()) && attempts--)
@@ -146,6 +147,49 @@ void test_connectivity() {
 	if (auto addr = pc2->remoteAddress())
 		cout << "Remote address 2: " << *addr << 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);
+
 	// Delay close of peer 2 to check closing works properly
 	pc1->close();
 	this_thread::sleep_for(1s);