Plugins prototype

worker/include/RTC/DataProducer.hpp

@@ -8,6 +8,7 @@
 #include "RTC/RTCP/Packet.hpp"
 #include "RTC/SctpDictionaries.hpp"
 #include "RTC/Shared.hpp"
+#include "Utils.hpp"
 #include <nlohmann/json.hpp>
 #include <string>
 
@@ -69,6 +70,8 @@ namespace RTC
 		// Passed by argument.
 		const std::string id;
 
+		Utils::Event<uint32_t, const uint8_t*, size_t> Data;
+
 	private:
 		// Passed by argument.
 		RTC::Shared* shared{ nullptr };

worker/include/RTC/Plugin.hpp

@@ -0,0 +1,15 @@
+#ifndef MS_RTC_PLUGIN_HPP
+#define MS_RTC_PLUGIN_HPP
+
+class Worker;
+
+namespace RTC
+{
+	class Plugin
+    {
+    public:
+        virtual ~Plugin() = default;
+    };
+} // namespace RTC
+
+#endif

worker/include/RTC/Router.hpp

@@ -15,6 +15,7 @@
 #include "RTC/Shared.hpp"
 #include "RTC/Transport.hpp"
 #include "RTC/WebRtcServer.hpp"
+#include "Utils.hpp"
 #include <absl/container/flat_hash_map.h>
 #include <nlohmann/json.hpp>
 #include <string>
@@ -115,6 +116,8 @@ namespace RTC
 	public:
 		// Passed by argument.
 		const std::string id;
+		Utils::Event<RTC::Transport*> TransportCreated;
+		Utils::Event<RTC::Transport*> TransportClosed;
 
 	private:
 		// Passed by argument.
@@ -124,6 +127,9 @@ namespace RTC
 		absl::flat_hash_map<std::string, RTC::Transport*> mapTransports;
 		absl::flat_hash_map<std::string, RTC::RtpObserver*> mapRtpObservers;
 		// Others.
+
+		// TODO: Replace with smart pointers
+
 		absl::flat_hash_map<RTC::Producer*, absl::flat_hash_set<RTC::Consumer*>> mapProducerConsumers;
 		absl::flat_hash_map<RTC::Consumer*, RTC::Producer*> mapConsumerProducer;
 		absl::flat_hash_map<RTC::Producer*, absl::flat_hash_set<RTC::RtpObserver*>> mapProducerRtpObservers;

worker/include/RTC/Transport.hpp

@@ -297,6 +297,15 @@ namespace RTC
 		// Passed by argument.
 		const std::string id;
 
+		Utils::Event<RTC::Producer*> ProducerCreated;
+		Utils::Event<RTC::Producer*> ProducerClosed;
+		Utils::Event<RTC::Consumer*> ConsumerCreated;
+		Utils::Event<RTC::Consumer*> ConsumerClosed;
+		Utils::Event<RTC::DataProducer*> DataProducerCreated;
+		Utils::Event<RTC::DataProducer*> DataProducerClosed;
+		Utils::Event<RTC::DataConsumer*> DataConsumerCreated;
+		Utils::Event<RTC::DataConsumer*> DataConsumerClosed;
+
 	protected:
 		RTC::Shared* shared{ nullptr };
 		size_t maxMessageSize{ 262144u };

worker/include/Utils.hpp

@@ -375,6 +375,32 @@ namespace Utils
 			}
 		}
 	};
+
+	template <typename... Args>
+	class Event
+	{
+	public:
+		// Typedef for the function signature.
+		typedef std::function<void(Args...)> Listener;
+		void operator()(Args... args) const
+		{
+			for (auto& listener : this->listeners)
+			{
+				listener(args...);
+			}
+		}
+		void operator+=(const Listener& listener)
+		{
+			this->listeners.push_back(listener);
+		}
+		void operator-=(const Listener& listener)
+		{
+			std::remove(listeners.begin(), listeners.end(), listener);
+		}
+	private:
+		std::vector<Listener> listeners;
+	};
+
 } // namespace Utils
 
 #endif

worker/include/Worker.hpp

@@ -7,13 +7,16 @@
 #include "PayloadChannel/PayloadChannelNotification.hpp"
 #include "PayloadChannel/PayloadChannelRequest.hpp"
 #include "PayloadChannel/PayloadChannelSocket.hpp"
+#include "RTC/Plugin.hpp"
 #include "RTC/Router.hpp"
 #include "RTC/Shared.hpp"
 #include "RTC/WebRtcServer.hpp"
+#include "Utils.hpp"
 #include "handles/SignalsHandler.hpp"
 #include <absl/container/flat_hash_map.h>
 #include <nlohmann/json.hpp>
 #include <string>
+#include <vector>
 
 using json = nlohmann::json;
 
@@ -25,6 +28,8 @@ class Worker : public Channel::ChannelSocket::Listener,
 public:
 	explicit Worker(Channel::ChannelSocket* channel, PayloadChannel::PayloadChannelSocket* payloadChannel);
 	~Worker();
+
+	void RunInContext(std::function<void()> func);
 
 private:
 	void Close();
@@ -63,6 +68,10 @@ class Worker : public Channel::ChannelSocket::Listener,
 public:
 	RTC::WebRtcServer* OnRouterNeedWebRtcServer(RTC::Router* router, std::string& webRtcServerId) override;
 
+public:
+	Utils::Event<RTC::Router*> RouterCreated;
+	Utils::Event<RTC::Router*> RouterClosed;
+
 private:
 	// Passed by argument.
 	Channel::ChannelSocket* channel{ nullptr };
@@ -72,6 +81,7 @@ class Worker : public Channel::ChannelSocket::Listener,
 	RTC::Shared* shared{ nullptr };
 	absl::flat_hash_map<std::string, RTC::WebRtcServer*> mapWebRtcServers;
 	absl::flat_hash_map<std::string, RTC::Router*> mapRouters;
+	std::vector<RTC::Plugin*> plugins;
 	// Others.
 	bool closed{ false };
 };

worker/meson.build

@@ -165,6 +165,7 @@ common_sources = [
   'src/RTC/RTCP/XR.cpp',
   'src/RTC/RTCP/XrDelaySinceLastRr.cpp',
   'src/RTC/RTCP/XrReceiverReferenceTime.cpp',
+  'plugins/DataPing/Plugin.cpp'
 ]
 
 cpp = meson.get_compiler('cpp')
@@ -264,7 +265,7 @@ libmediasoup_worker = library(
   install_tag: 'libmediasoup-worker',
   dependencies: dependencies,
   sources: common_sources,
-  include_directories: include_directories('include'),
+  include_directories: include_directories('include', 'plugins'),
   cpp_args: cpp_args,
   link_whole: link_whole,
 )
@@ -276,7 +277,7 @@ executable(
   install_tag: 'mediasoup-worker',
   dependencies: dependencies,
   sources: common_sources + ['src/main.cpp'],
-  include_directories: include_directories('include'),
+  include_directories: include_directories('include', 'plugins'),
   cpp_args: cpp_args + ['-DMS_EXECUTABLE'],
 )
 

worker/plugins/DataPing/Plugin.cpp

@@ -0,0 +1,61 @@
+#include "Plugin.hpp"
+
+#include <iostream>
+
+#include "Worker.hpp"
+#include "RTC/Router.hpp"
+#include "RTC/DataProducer.hpp"
+#include "RTC/DataConsumer.hpp"
+
+DataPingPlugin::DataPingPlugin(Worker* worker)
+{
+    worker->RouterCreated += [&](RTC::Router* router)
+    {
+        std::cerr << "RouterCreated" << std::endl;
+
+        router->TransportCreated += [&](RTC::Transport* transport)
+        {
+            std::cerr << "TransportCreated" << std::endl;
+
+            transport->DataProducerCreated += [&](RTC::DataProducer* producer)
+            {
+                producer->Data += [&](uint32_t id, const uint8_t* data, size_t len)
+                {
+                    auto isPing = len == 4 && std::string((const char*)data, len) == "ping";
+                    if (!isPing)
+                    {
+                        return;
+                    }
+                    for (auto [ consumerTransport, consumer ] : this->consumers)
+                    {
+                        if (consumerTransport == transport &&
+                            consumer->GetSctpStreamParameters().streamId == producer->GetSctpStreamParameters().streamId)
+                        {
+                            consumer->SendMessage(id, data, len);
+                        }
+                    }
+                    // TBD: Implement support to stop processing of this message.
+                    // return false;
+                };
+            };
+
+            transport->DataConsumerCreated += [&](RTC::DataConsumer* dataConsumer)
+            {
+                this->consumers.push_back({ transport, dataConsumer });
+            };
+
+            transport->DataConsumerClosed += [&](RTC::DataConsumer* dataConsumer)
+            {
+                for (auto it = this->consumers.begin(); it != this->consumers.end(); ++it)
+                {
+                    if (it->dataConsumer == dataConsumer)
+                    {
+                        this->consumers.erase(it);
+                        break;
+                    }
+                }
+            };
+
+        };
+    };
+}

worker/plugins/DataPing/Plugin.hpp

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "RTC/Plugin.hpp"
+#include "RTC/Transport.hpp"
+#include "RTC/DataConsumer.hpp"
+
+#include <vector>
+
+class DataPingPlugin: public RTC::Plugin
+{
+public:
+    DataPingPlugin(Worker* worker);
+
+private:
+    struct ConsumerInfo
+    {
+        RTC::Transport* transport;
+        RTC::DataConsumer* dataConsumer;
+    };
+    std::vector<ConsumerInfo> consumers;
+};
+
+// RTC::Plugin* CreatePlugin(Worker* worker);

worker/src/RTC/DataProducer.cpp

@@ -216,5 +216,7 @@ namespace RTC
 		this->bytesReceived += len;
 
 		this->listener->OnDataProducerMessageReceived(this, ppid, msg, len);
+
+		this->Data(ppid, msg, len);
 	}
 } // namespace RTC

worker/src/RTC/Router.cpp

@@ -41,6 +41,8 @@ namespace RTC
 			auto* transport = kv.second;
 
 			delete transport;
+
+			TransportClosed(transport);
 		}
 		this->mapTransports.clear();
 
@@ -297,6 +299,8 @@ namespace RTC
 
 				request->Accept(data);
 
+				TransportCreated(webRtcTransport);
+
 				break;
 			}
 
@@ -321,6 +325,8 @@ namespace RTC
 
 				request->Accept(data);
 
+				TransportCreated(plainTransport);
+
 				break;
 			}
 
@@ -343,6 +349,8 @@ namespace RTC
 				pipeTransport->FillJson(data);
 
 				request->Accept(data);
+
+				TransportCreated(pipeTransport);
 
 				break;
 			}
@@ -367,6 +375,8 @@ namespace RTC
 				directTransport->FillJson(data);
 
 				request->Accept(data);
+
+				TransportCreated(directTransport);
 
 				break;
 			}
@@ -430,6 +440,8 @@ namespace RTC
 
 				request->Accept();
 
+				TransportClosed(transport);
+
 				break;
 			}
 
@@ -718,17 +730,22 @@ namespace RTC
 			// Cloned ref-counted packet that RtpStreamSend will store for as long as
 			// needed avoiding multiple allocations unless absolutely necessary.
 			// Clone only happens if needed.
-			std::shared_ptr<RTC::RtpPacket> sharedPacket;
+			std::shared_ptr<RTC::RtpPacket> sharedPacket(packet->Clone());
 
 			for (auto* consumer : consumers)
 			{
-				// Update MID RTP extension value.
-				const auto& mid = consumer->GetRtpParameters().mid;
-
-				if (!mid.empty())
-					packet->UpdateMid(mid);
-
-				consumer->SendRtpPacket(packet, sharedPacket);
+				// Possible optimizations
+				// if consumer.worker == producer.worker
+				// group consumers by worker and call run per worker
+				consumer->RunInContext([consumer, sharedPacket]() mutable {
+					// Update MID RTP extension value.
+					const auto& mid = consumer->GetRtpParameters().mid;
+
+					if (!mid.empty())
+						sharedPacket->UpdateMid(mid);
+
+					consumer->SendRtpPacket(sharedPacket.get(), sharedPacket);
+				});
 			}
 		}