Add PacingHandler

MediaHandler that can be used to pace packet delivery

Author: Sean-Der

CMakeLists.txt

@@ -87,6 +87,7 @@ set(LIBDATACHANNEL_SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/src/rtp.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/capi.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/plihandler.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/pacinghandler.cpp
 )
 
 set(LIBDATACHANNEL_HEADERS
@@ -123,6 +124,7 @@ set(LIBDATACHANNEL_HEADERS
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtcpnackresponder.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/utils.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/plihandler.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/pacinghandler.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/version.h
 )
 

include/rtc/pacinghandler.hpp

@@ -0,0 +1,52 @@
+/**
+ * Copyright (c) 2020 Staz Modrzynski
+ * Copyright (c) 2020 Paul-Louis Ageneau
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+#ifndef RTC_PACING_HANDLER_H
+#define RTC_PACING_HANDLER_H
+
+#if RTC_ENABLE_MEDIA
+
+#include "mediahandler.hpp"
+#include "utils.hpp"
+
+#include <atomic>
+#include <deque>
+
+namespace rtc {
+
+// Paced sending of RTP packets. Takes a stream of RTP packets that can an
+// uneven bitrate. It then delivers these packets in a smoother manner by
+// sending a fixed size of them on an interval
+class RTC_CPP_EXPORT PacingHandler : public MediaHandler {
+public:
+	PacingHandler(size_t windowSize, std::chrono::milliseconds sendInterval);
+
+	void outgoing(message_vector &messages, const message_callback &send) override;
+
+private:
+	std::atomic<bool> mHaveScheduled = false;
+	size_t mWindowSize;
+	std::chrono::milliseconds mSendInterval;
+
+	class RTC_CPP_EXPORT Storage {
+	public:
+		std::mutex mMutex;
+		std::deque<message_ptr> mRtpBuffer = {};
+	};
+
+	const shared_ptr<Storage> mStorage;
+
+	void schedule(const message_callback &send);
+};
+
+} // namespace rtc
+
+#endif // RTC_ENABLE_MEDIA
+
+#endif // RTC_PACING_HANDLER_H

include/rtc/rtc.hpp

@@ -34,6 +34,7 @@
 #include "h265rtppacketizer.hpp"
 #include "mediahandler.hpp"
 #include "plihandler.hpp"
+#include "pacinghandler.hpp"
 #include "rtcpnackresponder.hpp"
 #include "rtcpreceivingsession.hpp"
 #include "rtcpsrreporter.hpp"

src/pacinghandler.cpp

@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+#if RTC_ENABLE_MEDIA
+
+#include <memory>
+
+#include "pacinghandler.hpp"
+
+#include "impl/internals.hpp"
+#include "impl/threadpool.hpp"
+
+namespace rtc {
+
+PacingHandler::PacingHandler(size_t windowSize, std::chrono::milliseconds sendInterval)
+    : mWindowSize(windowSize), mSendInterval(sendInterval), mStorage(std::make_shared<Storage>()){};
+
+void PacingHandler::schedule(const message_callback &send) {
+	if (!mHaveScheduled.exchange(true)) {
+		return;
+	}
+
+	impl::ThreadPool::Instance().schedule(mSendInterval, [weak_this = weak_from_this(), send]() {
+		if (auto locked = std::dynamic_pointer_cast<PacingHandler>(weak_this.lock())) {
+			const std::lock_guard<std::mutex> lock(locked->mStorage->mMutex);
+			locked->mHaveScheduled.store(false);
+
+			size_t amountSent = 0;
+			while (!locked->mStorage->mRtpBuffer.empty()) {
+				amountSent += locked->mStorage->mRtpBuffer.front()->size();
+				if (amountSent >= locked->mWindowSize) {
+					return;
+				}
+
+				send(std::move(locked->mStorage->mRtpBuffer.front()));
+				locked->mStorage->mRtpBuffer.pop_front();
+			}
+		}
+	});
+}
+
+void PacingHandler::outgoing(message_vector &messages, const message_callback &send) {
+
+	std::lock_guard<std::mutex> lock(mStorage->mMutex);
+
+	std::move(std::begin(messages), std::end(messages), std::back_inserter(mStorage->mRtpBuffer));
+	messages.clear();
+
+	schedule(send);
+}
+
+} // namespace rtc
+
+#endif /* RTC_ENABLE_MEDIA */