Implementation of data channels on top of webrtc data channels

CMakeLists.txt

@@ -60,6 +60,8 @@ set(HEADER_FILES
 set(
 	SOURCE_FILES
 	src/Consumer.cpp
+	src/DataConsumer.cpp
+	src/DataProducer.cpp
 	src/Device.cpp
 	src/Handler.cpp
 	src/Logger.cpp

include/DataConsumer.hpp

@@ -0,0 +1,76 @@
+#ifndef MSC_DATACONSUMER_HPP
+#define MSC_DATACONSUMER_HPP
+
+#include <json.hpp>
+#include <api/media_stream_interface.h> // webrtc::MediaStreamTrackInterface
+#include <api/rtp_receiver_interface.h> // webrtc::RtpReceiverInterface
+#include <api/data_channel_interface.h>
+#include <string>
+
+namespace mediasoupclient
+{
+	class DataConsumer : public webrtc::DataChannelObserver {
+		public: 
+		class PrivateListener
+		{
+			public:
+				virtual void OnClose(DataConsumer* dataConsumer) = 0;
+		};
+
+		class Listener {
+			public:
+				// The data channel state have changed.
+				virtual void OnStateChange(DataConsumer* dataConsumer) { };
+				//  A data buffer was successfully received.
+				virtual void OnMessage(DataConsumer* dataConsumer, const webrtc::DataBuffer& buffer) = 0;
+				// The data channel's buffered_amount has changed.
+				// virtual void OnBufferedAmountChange(uint64_t sent_data_size) {}
+				//
+				virtual void OnTransportClosed(DataConsumer* dataConsumer) { };
+		};
+
+		public:
+			DataConsumer(
+				Listener* listener, 
+				PrivateListener* privateListener,
+				const std::string& id,
+				const std::string& dataProducerId,
+				rtc::scoped_refptr<webrtc::DataChannelInterface> webrtcDataChannel,
+				const nlohmann::json& sctpStreamParameters,
+				const nlohmann::json& appData);
+
+			const std::string& GetId();
+			std::string GetLocalId() const;
+			const std::string& GetDataProducerId();
+			const bool IsClosed();
+			const nlohmann::json& GetSctpStreamParameters();
+			webrtc::DataChannelInterface::DataState GetReadyState();
+			std::string GetLabel();
+			std::string GetProtocol();
+			const nlohmann::json& GetAppData();
+			void Close();
+			void TransportClosed();
+
+		private:
+			Listener* listener;
+			PrivateListener* privateListener;
+			std::string id;
+			std::string dataProducerId;
+			rtc::scoped_refptr<webrtc::DataChannelInterface> webrtcDataChannel;
+			bool closed;
+			nlohmann::json sctpParameters;
+			nlohmann::json appData;
+
+
+		public:
+			// From webrtc::DataChannelObsever
+			// The data channel state have changed.
+			void OnStateChange() override;
+			//  A data buffer was successfully received.
+			void OnMessage(const webrtc::DataBuffer& buffer) override;
+			// The data channel's buffered_amount has changed.
+			void OnBufferedAmountChange(uint64_t sent_data_size) override;
+	};
+} // namespace mediasoupclient
+
+#endif

include/DataProducer.hpp

@@ -0,0 +1,82 @@
+#ifndef MSC_DATAPRODUCER_HPP
+#define MSC_DATAPRODUCER_HPP
+
+#include <json.hpp>
+#include <string>
+#include <api/data_channel_interface.h>
+#include <api/stats/rtcstats_objects.h>
+#include <third_party/abseil-cpp/absl/types/optional.h>
+#include "Handler.hpp"
+
+namespace mediasoupclient
+{
+	// class Transport;
+	class SendTransport;
+
+	class DataProducer : public webrtc::DataChannelObserver {
+		public:
+			class PrivateListener
+			{
+			public:
+				virtual void OnClose(DataProducer* dataProducer) = 0;				
+			};
+
+		/* Public Listener API */
+		class Listener
+		{
+		public:
+			virtual void OnTransportClose(DataProducer* dataProducer) {};
+			virtual void OnBufferedAmountChange(DataProducer* dataProducer, uint64_t sent_data_size) {};
+			// virtual void OnOpen(DataProducer* dataProducer) = 0;
+			// virtual void OnError(DataProducer* dataProducer) = 0;
+			// virtual void OnClose(DataProducer* dataProducer) = 0;
+		};
+
+
+		private:
+			PrivateListener* privateListener;
+			Listener* listener;
+    		std::string id;
+			// std::string _localId;
+    		rtc::scoped_refptr<webrtc::DataChannelInterface> webrtcDataChannel;
+    		bool closed;
+    		nlohmann::json sctpStreamParameters;
+    		nlohmann::json appData;
+			void TransportClosed();
+
+			friend SendTransport;
+
+		public:
+			DataProducer(
+				DataProducer::PrivateListener* privateListener,
+				DataProducer::Listener* listener,
+				const std::string& id,
+				// const std::string& localId,
+				rtc::scoped_refptr<webrtc::DataChannelInterface> webrtcDataChannel, 
+				const nlohmann::json& sctpStreamParameters, 
+				const nlohmann::json& appData);
+			const std::string& GetId() const;
+			std::string GetLocalId() const;
+			// const std::string& GetLocalId() const;
+			bool IsClosed() const;
+			const nlohmann::json& GetSctpStreamParameters() const;
+			webrtc::DataChannelInterface::DataState GetReadyState(); // I would call this GetDataState()
+			std::string GetLabel();
+			std::string GetProtocol();
+			uint64_t GetBufferedAmount() const;
+			uint64_t BufferedAmountLowThreshold;
+			const nlohmann::json& GetAppData() const;
+			void Close();
+			void Send(const webrtc::DataBuffer& buffer);
+
+		// from DataChannelObserver
+		public:
+			void OnStateChange() override;
+  			//  A data buffer was successfully received.
+  			void OnMessage(const webrtc::DataBuffer& buffer) override;
+  			// The data channel's buffered_amount has changed.
+  			void OnBufferedAmountChange(uint64_t sent_data_size) override;
+	};
+} // namespace mediasoupclient
+
+#endif

include/Handler.hpp

@@ -10,6 +10,8 @@
 #include <api/rtp_receiver_interface.h>    // webrtc::RtpReceiverInterface
 #include <api/rtp_sender_interface.h>      // webrtc::RtpSenderInterface
 #include <api/rtp_transceiver_interface.h> // webrtc::RtpTransceiverInterface
+#include <third_party/abseil-cpp/absl/types/optional.h>
+
 #include <string>
 #include <unordered_map>
 
@@ -65,6 +67,8 @@ namespace mediasoupclient
 		std::unordered_map<std::string, webrtc::RtpTransceiverInterface*> mapMidTransceiver;
 		// PeerConnection instance.
 		std::unique_ptr<PeerConnection> pc;
+		bool hasDataChannelMediaSection = false;
+		uint32_t nextSendSctpStreamId = 0;
 	};
 
 	class SendHandler : public Handler
@@ -77,6 +81,15 @@ namespace mediasoupclient
 			nlohmann::json rtpParameters;
 		};
 
+		struct SendDataChannel
+		{
+			const std::string localId;
+			const rtc::scoped_refptr<webrtc::DataChannelInterface> dataChannel;
+			const nlohmann::json sctpStreamParameters;
+			SendDataChannel(std::string localId, rtc::scoped_refptr<webrtc::DataChannelInterface> dataChannel, nlohmann::json sctpStreamParameters):
+				localId(localId), dataChannel(dataChannel), sctpStreamParameters(sctpStreamParameters){}
+		};
+
 	public:
 		SendHandler(
 		  Handler::PrivateListener* privateListener,
@@ -94,10 +107,12 @@ namespace mediasoupclient
 		  std::vector<webrtc::RtpEncodingParameters>* encodings,
 		  const nlohmann::json* codecOptions);
 		void StopSending(const std::string& localId);
+		void StopSendingData(const std::string& localId);
 		void ReplaceTrack(const std::string& localId, webrtc::MediaStreamTrackInterface* track);
 		void SetMaxSpatialLayer(const std::string& localId, uint8_t spatialLayer);
 		nlohmann::json GetSenderStats(const std::string& localId);
 		void RestartIce(const nlohmann::json& iceParameters) override;
+		SendDataChannel CreateSendDataChannel(const std::string& label, webrtc::DataChannelInit dataChannelInit);
 
 	private:
 		// Generic sending RTP parameters for audio and video.
@@ -117,6 +132,16 @@ namespace mediasoupclient
 			webrtc::MediaStreamTrackInterface* track{ nullptr };
 		};
 
+		struct RecvDataChannel
+		{
+			const std::string localId;
+			const rtc::scoped_refptr<webrtc::DataChannelInterface> webrtcDataChannel;
+			const nlohmann::json sctpStreamParameters;
+			RecvDataChannel(std::string localId,
+				rtc::scoped_refptr<webrtc::DataChannelInterface> webrtcDataChannel,
+				nlohmann::json sctpStreamParameters): localId(localId), webrtcDataChannel(webrtcDataChannel), sctpStreamParameters(sctpStreamParameters){}
+		};
+
 	public:
 		RecvHandler(
 		  Handler::PrivateListener* privateListener,
@@ -128,8 +153,10 @@ namespace mediasoupclient
 
 		RecvData Receive(const std::string& id, const std::string& kind, const nlohmann::json* rtpParameters);
 		void StopReceiving(const std::string& localId);
+		void StopReceivingData(const std::string& localId);
 		nlohmann::json GetReceiverStats(const std::string& localId);
 		void RestartIce(const nlohmann::json& iceParameters) override;
+		RecvDataChannel CreateRecvDataChannel(const std::string& label, webrtc::DataChannelInit dataChannelInit);
 	};
 } // namespace mediasoupclient
 

include/PeerConnection.hpp

@@ -130,6 +130,7 @@ namespace mediasoupclient
 		nlohmann::json GetStats();
 		nlohmann::json GetStats(rtc::scoped_refptr<webrtc::RtpSenderInterface> selector);
 		nlohmann::json GetStats(rtc::scoped_refptr<webrtc::RtpReceiverInterface> selector);
+		rtc::scoped_refptr<webrtc::DataChannelInterface> CreateDataChannel(const std::string& label, const webrtc::DataChannelInit* config);
 
 	private:
 		// Signaling and worker threads.

include/Producer.hpp

@@ -64,7 +64,7 @@ namespace mediasoupclient
 	private:
 		void TransportClosed();
 
-		/* SendTransport will create instances and call private member TransporClosed */
+		/* SendTransport will create instances and call private member TransportClosed */
 		friend SendTransport;
 
 	private: