Merge pull request #291 from in2core/feature/stream-h264-opus

Streaming H264 and opus samples

.gitignore

@@ -7,4 +7,5 @@ node_modules/
 compile_commands.json
 tests
 .DS_Store
+.idea
 

CMakeLists.txt

@@ -65,6 +65,14 @@ set(LIBDATACHANNEL_SOURCES
 	${CMAKE_CURRENT_SOURCE_DIR}/src/track.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/processor.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/src/capi.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/rtppacketizationconfig.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/rtcpsenderreportable.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/rtppacketizer.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/opusrtppacketizer.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/opuspacketizationhandler.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/h264rtppacketizer.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/nalunit.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/src/h264packetizationhandler.cpp
 )
 
 set(LIBDATACHANNEL_WEBSOCKET_SOURCES
@@ -96,6 +104,14 @@ set(LIBDATACHANNEL_HEADERS
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtp.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/track.hpp
 	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/websocket.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtppacketizationconfig.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtcpsenderreportable.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/rtppacketizer.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/opusrtppacketizer.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/opuspacketizationhandler.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/h264rtppacketizer.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/nalunit.hpp
+	${CMAKE_CURRENT_SOURCE_DIR}/include/rtc/h264packetizationhandler.hpp
 )
 
 set(TESTS_SOURCES
@@ -296,6 +312,9 @@ if(NOT NO_EXAMPLES AND NOT CMAKE_SYSTEM_NAME STREQUAL "WindowsStore")
 	add_subdirectory(examples/client)
 	add_subdirectory(examples/media)
 	add_subdirectory(examples/sfu-media)
+if(NOT NO_MEDIA)
+    add_subdirectory(examples/streamer)
+endif()
 	add_subdirectory(examples/copy-paste)
 	add_subdirectory(examples/copy-paste-capi)
 endif()

examples/README.md

@@ -8,6 +8,7 @@ This directory contains different WebRTC clients and compatible WebSocket + JSON
 - [signaling-server-rust](signaling-server-rust) contains a similar signaling server in Rust (see [lerouxrgd/datachannel-rs](https://github.com/lerouxrgd/datachannel-rs) for Rust wrappers)
 
 - [media](media) is a copy/paste demo to send the webcam from your browser into gstreamer.
+- [streamer](streamer) streams h264 and opus samples to web browsers (signaling-server-python is required).
 
 Additionally, it contains two debugging tools for libdatachannel with copy-pasting as signaling:
 - [copy-paste](copy-paste) using the C++ API

examples/streamer/ArgParser.cpp

@@ -0,0 +1,75 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "ArgParser.hpp"
+#include <iostream>
+
+ArgParser::ArgParser(std::vector<std::pair<std::string, std::string>> options, std::vector<std::pair<std::string, std::string>> flags) {
+    for(auto option: options) {
+        this->options.insert(option.first);
+        this->options.insert(option.second);
+        shortToLongMap.emplace(option.first, option.second);
+        shortToLongMap.emplace(option.second, option.second);
+    }
+    for(auto flag: flags) {
+        this->flags.insert(flag.first);
+        this->flags.insert(flag.second);
+        shortToLongMap.emplace(flag.first, flag.second);
+        shortToLongMap.emplace(flag.second, flag.second);
+    }
+}
+
+std::optional<std::string> ArgParser::toKey(std::string prefixedKey) {
+    if (prefixedKey.find("--") == 0) {
+        return prefixedKey.substr(2, prefixedKey.length());
+    } else if (prefixedKey.find("-") == 0) {
+        return prefixedKey.substr(1, prefixedKey.length());
+    } else {
+        return std::nullopt;
+    }
+}
+
+bool ArgParser::parse(int argc, char **argv, std::function<bool (std::string, std::string)> onOption, std::function<bool (std::string)> onFlag) {
+    std::optional<std::string> currentOption = std::nullopt;
+    for(int i = 1; i < argc; i++) {
+        std::string current = argv[i];
+        auto optKey = toKey(current);
+        if (!currentOption.has_value() && optKey.has_value() && flags.find(optKey.value()) != flags.end()) {
+            auto check = onFlag(shortToLongMap.at(optKey.value()));
+            if (!check) {
+                return false;
+            }
+        } else if (!currentOption.has_value() && optKey.has_value() && options.find(optKey.value()) != options.end()) {
+            currentOption = optKey.value();
+        } else if (currentOption.has_value()) {
+            auto check = onOption(shortToLongMap.at(currentOption.value()), current);
+            if (!check) {
+                return false;
+            }
+            currentOption = std::nullopt;
+        } else {
+            std::cerr << "Unrecognized option " << current << std::endl;
+            return false;
+        }
+    }
+    if (currentOption.has_value()) {
+        std::cerr << "Missing value for " << currentOption.value() << std::endl;
+        return false;
+    }
+    return true;
+}

examples/streamer/ArgParser.hpp

@@ -0,0 +1,41 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef ArgParser_hpp
+#define ArgParser_hpp
+
+#include <functional>
+#include <vector>
+#include <utility>
+#include <string>
+#include <set>
+#include <unordered_map>
+#include <optional>
+
+struct ArgParser {
+private:
+    std::set<std::string> options{};
+    std::set<std::string> flags{};
+    std::unordered_map<std::string, std::string> shortToLongMap{};
+public:
+    ArgParser(std::vector<std::pair<std::string, std::string>> options, std::vector<std::pair<std::string, std::string>> flags);
+    std::optional<std::string> toKey(std::string prefixedKey);
+    bool parse(int argc, char **argv, std::function<bool (std::string, std::string)> onOption, std::function<bool (std::string)> onFlag);
+};
+
+#endif /* ArgParser_hpp */

examples/streamer/CMakeLists.txt

@@ -0,0 +1,19 @@
+cmake_minimum_required(VERSION 3.7)
+if(POLICY CMP0079)
+	cmake_policy(SET CMP0079 NEW)
+endif()
+
+if(WIN32)
+add_executable(streamer main.cpp dispatchqueue.cpp dispatchqueue.hpp h264fileparser.cpp h264fileparser.hpp helpers.cpp helpers.hpp opusfileparser.cpp opusfileparser.hpp fileparser.cpp fileparser.hpp stream.cpp stream.hpp ArgParser.hpp ArgParser.cpp)
+target_compile_definitions(streamer PUBLIC STATIC_GETOPT)
+else()
+add_executable(streamer main.cpp dispatchqueue.cpp dispatchqueue.hpp h264fileparser.cpp h264fileparser.hpp helpers.cpp helpers.hpp opusfileparser.cpp opusfileparser.hpp fileparser.cpp fileparser.hpp stream.cpp stream.hpp ArgParser.hpp ArgParser.cpp)
+endif()
+set_target_properties(streamer PROPERTIES
+	CXX_STANDARD 17
+	OUTPUT_NAME streamer)
+
+target_link_libraries(streamer datachannel)
+
+target_link_libraries(streamer datachannel nlohmann_json)
+

examples/streamer/README.md

@@ -0,0 +1,32 @@
+# Streaming H264 and opus
+
+This example streams H264 and opus<sup id="a1">[1](#f1)</sup> samples to the connected browser client.
+
+## Starting signaling server
+
+```sh
+$ python3 ../signaling-server-python/signaling-server.py
+```
+
+## Starting php
+
+```sh
+$ php -S 127.0.0.1:8080
+```
+
+Now you can open demo at [127.0.0.1:8080](127.0.0.1:8080).
+
+## Arguments
+
+- `-a` Directory with OPUS samples (default: *../../../../examples/streamer/samples/opus/*).
+- `-b` Directory with H264 samples (default: *../../../../examples/streamer/samples/h264/*).
+- `-d` Signaling server IP address (default: 127.0.0.1).
+- `-p` Signaling server port (default: 8000).
+- `-v` Enable debug logs.
+- `-h` Print this help and exit.
+
+## Generating H264 and Opus samples
+
+You can generate H264 and Opus sample with *samples/generate_h264.py* and *samples/generate_opus.py* respectively. This require ffmpeg, python3 and kaitaistruct library to be installed. Use `-h`/`--help` to learn more about arguments.
+
+<b id="f1">1</b> Opus samples are generated from music downloaded at [bensound](https://www.bensound.com). [↩](#a1)

examples/streamer/client.js

@@ -0,0 +1,209 @@
+/** @type {RTCPeerConnection} */
+let rtc;
+const iceConnectionLog = document.getElementById('ice-connection-state'),
+    iceGatheringLog = document.getElementById('ice-gathering-state'),
+    signalingLog = document.getElementById('signaling-state'),
+    dataChannelLog = document.getElementById('data-channel');
+
+function randomString(len) {
+    const charSet = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789';
+    let randomString = '';
+    for (let i = 0; i < len; i++) {
+        const randomPoz = Math.floor(Math.random() * charSet.length);
+        randomString += charSet.substring(randomPoz, randomPoz + 1);
+    }
+    return randomString;
+}
+
+const receiveID = randomString(10);
+const websocket = new WebSocket('ws://127.0.0.1:8000/' + receiveID);
+websocket.onopen = function () {
+    document.getElementById('start').disabled = false;
+}
+
+// data channel
+let dc = null, dcTimeout = null;
+
+function createPeerConnection() {
+    const config = {
+        sdpSemantics: 'unified-plan',
+        bundlePolicy: "max-bundle",
+    };
+
+    if (document.getElementById('use-stun').checked) {
+        config.iceServers = [{urls: ['stun:stun.l.google.com:19302']}];
+    }
+
+    let pc = new RTCPeerConnection(config);
+
+    // register some listeners to help debugging
+    pc.addEventListener('icegatheringstatechange', function () {
+        iceGatheringLog.textContent += ' -> ' + pc.iceGatheringState;
+    }, false);
+    iceGatheringLog.textContent = pc.iceGatheringState;
+
+    pc.addEventListener('iceconnectionstatechange', function () {
+        iceConnectionLog.textContent += ' -> ' + pc.iceConnectionState;
+    }, false);
+    iceConnectionLog.textContent = pc.iceConnectionState;
+
+    pc.addEventListener('signalingstatechange', function () {
+        signalingLog.textContent += ' -> ' + pc.signalingState;
+    }, false);
+    signalingLog.textContent = pc.signalingState;
+
+    // connect audio / video
+    pc.addEventListener('track', function (evt) {
+        if (evt.track.kind == 'video') {
+            document.getElementById('media').style.display = 'block';
+            document.getElementById('video').srcObject = evt.streams[0];
+        } else {
+            document.getElementById('audio').srcObject = evt.streams[0];
+        }
+    });
+
+    let time_start = null;
+
+    function current_stamp() {
+        if (time_start === null) {
+            time_start = new Date().getTime();
+            return 0;
+        } else {
+            return new Date().getTime() - time_start;
+        }
+    }
+
+    pc.ondatachannel = function (event) {
+        dc = event.channel;
+        dc.onopen = function () {
+            dataChannelLog.textContent += '- open\n';
+            dataChannelLog.scrollTop = dataChannelLog.scrollHeight;
+        };
+        dc.onmessage = function (evt) {
+
+            dataChannelLog.textContent += '< ' + evt.data + '\n';
+            dataChannelLog.scrollTop = dataChannelLog.scrollHeight;
+
+            dcTimeout = setTimeout(function () {
+                if (dc == null && dcTimeout != null) {
+                    dcTimeout = null;
+                    return
+                }
+                const message = 'Pong ' + current_stamp();
+                dataChannelLog.textContent += '> ' + message + '\n';
+                dataChannelLog.scrollTop = dataChannelLog.scrollHeight;
+                dc.send(message);
+            }, 1000);
+        }
+        dc.onclose = function () {
+            clearTimeout(dcTimeout);
+            dcTimeout = null;
+            dataChannelLog.textContent += '- close\n';
+            dataChannelLog.scrollTop = dataChannelLog.scrollHeight;
+        };
+    }
+
+    return pc;
+}
+
+function sendAnswer(pc) {
+    return pc.createAnswer()
+        .then((answer) => rtc.setLocalDescription(answer))
+        .then(function () {
+            // wait for ICE gathering to complete
+            return new Promise(function (resolve) {
+                if (pc.iceGatheringState === 'complete') {
+                    resolve();
+                } else {
+                    function checkState() {
+                        if (pc.iceGatheringState === 'complete') {
+                            pc.removeEventListener('icegatheringstatechange', checkState);
+                            resolve();
+                        }
+                    }
+
+                    pc.addEventListener('icegatheringstatechange', checkState);
+                }
+            });
+        }).then(function () {
+            const answer = pc.localDescription;
+
+            document.getElementById('answer-sdp').textContent = answer.sdp;
+
+            return websocket.send(JSON.stringify(
+                {
+                    id: "server",
+                    type: answer.type,
+                    sdp: answer.sdp,
+                }));
+        }).catch(function (e) {
+            alert(e);
+        });
+}
+
+function handleOffer(offer) {
+    rtc = createPeerConnection();
+    return rtc.setRemoteDescription(offer)
+        .then(() => sendAnswer(rtc));
+}
+
+function sendStreamRequest() {
+    websocket.send(JSON.stringify(
+        {
+            id: "server",
+            type: "streamRequest",
+            receiver: receiveID,
+        }));
+}
+
+async function start() {
+    document.getElementById('start').style.display = 'none';
+    document.getElementById('stop').style.display = 'inline-block';
+    document.getElementById('media').style.display = 'block';
+    sendStreamRequest();
+}
+
+function stop() {
+    document.getElementById('stop').style.display = 'none';
+    document.getElementById('media').style.display = 'none';
+    document.getElementById('start').style.display = 'inline-block';
+
+    // close data channel
+    if (dc) {
+        dc.close();
+        dc = null;
+    }
+
+    // close transceivers
+    if (rtc.getTransceivers) {
+        rtc.getTransceivers().forEach(function (transceiver) {
+            if (transceiver.stop) {
+                transceiver.stop();
+            }
+        });
+    }
+
+    // close local audio / video
+    rtc.getSenders().forEach(function (sender) {
+        const track = sender.track;
+        if (track !== null) {
+            sender.track.stop();
+        }
+    });
+
+    // close peer connection
+    setTimeout(function () {
+        rtc.close();
+        rtc = null;
+    }, 500);
+}
+
+
+websocket.onmessage = async function (evt) {
+    const received_msg = evt.data;
+    const object = JSON.parse(received_msg);
+    if (object.type == "offer") {
+        document.getElementById('offer-sdp').textContent = object.sdp;
+        await handleOffer(object)
+    }
+}

examples/streamer/dispatchqueue.cpp

@@ -0,0 +1,94 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+
+#include "dispatchqueue.hpp"
+
+DispatchQueue::DispatchQueue(std::string name, size_t threadCount) :
+    name{std::move(name)}, threads(threadCount) {
+    for(size_t i = 0; i < threads.size(); i++)
+    {
+        threads[i] = std::thread(&DispatchQueue::dispatchThreadHandler, this);
+    }
+}
+
+DispatchQueue::~DispatchQueue() {
+    // Signal to dispatch threads that it's time to wrap up
+    std::unique_lock<std::mutex> lock(lockMutex);
+    quit = true;
+    lock.unlock();
+    condition.notify_all();
+
+    // Wait for threads to finish before we exit
+    for(size_t i = 0; i < threads.size(); i++)
+    {
+        if(threads[i].joinable())
+        {
+            threads[i].join();
+        }
+    }
+}
+
+void DispatchQueue::removePending() {
+    std::unique_lock<std::mutex> lock(lockMutex);
+    queue = {};
+}
+
+void DispatchQueue::dispatch(const fp_t& op) {
+    std::unique_lock<std::mutex> lock(lockMutex);
+    queue.push(op);
+
+    // Manual unlocking is done before notifying, to avoid waking up
+    // the waiting thread only to block again (see notify_one for details)
+    lock.unlock();
+    condition.notify_one();
+}
+
+void DispatchQueue::dispatch(fp_t&& op) {
+    std::unique_lock<std::mutex> lock(lockMutex);
+    queue.push(std::move(op));
+
+    // Manual unlocking is done before notifying, to avoid waking up
+    // the waiting thread only to block again (see notify_one for details)
+    lock.unlock();
+    condition.notify_one();
+}
+
+void DispatchQueue::dispatchThreadHandler(void) {
+    std::unique_lock<std::mutex> lock(lockMutex);
+    do {
+        //Wait until we have data or a quit signal
+        condition.wait(lock, [this]{
+            return (queue.size() || quit);
+        });
+
+        //after wait, we own the lock
+        if(!quit && queue.size())
+        {
+            auto op = std::move(queue.front());
+            queue.pop();
+
+            //unlock now that we're done messing with the queue
+            lock.unlock();
+
+            op();
+
+            lock.lock();
+        }
+    } while (!quit);
+}

examples/streamer/dispatchqueue.hpp

@@ -0,0 +1,59 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef dispatchqueue_hpp
+#define dispatchqueue_hpp
+
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+#include <queue>
+#include <functional>
+
+class DispatchQueue {
+    typedef std::function<void(void)> fp_t;
+
+public:
+    DispatchQueue(std::string name, size_t threadCount = 1);
+    ~DispatchQueue();
+
+    // dispatch and copy
+    void dispatch(const fp_t& op);
+    // dispatch and move
+    void dispatch(fp_t&& op);
+
+    void removePending();
+
+    // Deleted operations
+    DispatchQueue(const DispatchQueue& rhs) = delete;
+    DispatchQueue& operator=(const DispatchQueue& rhs) = delete;
+    DispatchQueue(DispatchQueue&& rhs) = delete;
+    DispatchQueue& operator=(DispatchQueue&& rhs) = delete;
+
+private:
+    std::string name;
+    std::mutex lockMutex;
+    std::vector<std::thread> threads;
+    std::queue<fp_t> queue;
+    std::condition_variable condition;
+    bool quit = false;
+
+    void dispatchThreadHandler(void);
+};
+
+#endif /* dispatchqueue_hpp */

examples/streamer/fileparser.cpp

@@ -0,0 +1,59 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "fileparser.hpp"
+#include <fstream>
+
+using namespace std;
+
+FileParser::FileParser(string directory, string extension, uint32_t samplesPerSecond, bool loop): sampleDuration_us(1000 * 1000 / samplesPerSecond), StreamSource() {
+    this->directory = directory;
+    this->extension = extension;
+    this->loop = loop;
+}
+
+void FileParser::start() {
+    sampleTime_us = -sampleDuration_us;
+    loadNextSample();
+}
+
+void FileParser::stop() {
+    StreamSource::stop();
+    counter = -1;
+}
+
+void FileParser::loadNextSample() {
+    string frame_id = to_string(++counter);
+
+    string url = directory + "/sample-" + frame_id + extension;
+    ifstream source(url, ios_base::binary);
+    if (!source) {
+        if (loop && counter > 0) {
+            loopTimestampOffset = sampleTime_us;
+            counter = -1;
+            loadNextSample();
+            return;
+        }
+        sample = {};
+        return;
+    }
+
+    vector<uint8_t> fileContents((std::istreambuf_iterator<char>(source)), std::istreambuf_iterator<char>());
+    sample = *reinterpret_cast<vector<byte> *>(&fileContents);
+    sampleTime_us += sampleDuration_us;
+}

examples/streamer/fileparser.hpp

@@ -0,0 +1,40 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef fileparser_hpp
+#define fileparser_hpp
+
+#include <string>
+#include <vector>
+#include "stream.hpp"
+
+class FileParser: public StreamSource {
+    std::string directory;
+    std::string extension;
+    uint32_t counter = -1;
+    bool loop;
+    uint64_t loopTimestampOffset = 0;
+public:
+    const uint64_t sampleDuration_us;
+    virtual void start();
+    virtual void stop();
+    FileParser(std::string directory, std::string extension, uint32_t samplesPerSecond, bool loop);
+    virtual void loadNextSample();
+};
+
+#endif /* fileparser_hpp */

examples/streamer/h264fileparser.cpp

@@ -0,0 +1,70 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "h264fileparser.hpp"
+#include <fstream>
+#include "rtc/rtc.hpp"
+
+using namespace std;
+
+H264FileParser::H264FileParser(string directory, uint32_t fps, bool loop): FileParser(directory, ".h264", fps, loop) { }
+
+void H264FileParser::loadNextSample() {
+    FileParser::loadNextSample();
+
+    unsigned long long i = 0;
+    while (i < sample.size()) {
+        assert(i + 4 < sample.size());
+        auto lengthPtr = (uint32_t *) (sample.data() + i);
+        uint32_t length = ntohl(*lengthPtr);
+        auto naluStartIndex = i + 4;
+        auto naluEndIndex = naluStartIndex + length;
+        assert(naluEndIndex <= sample.size());
+        auto header = reinterpret_cast<rtc::NalUnitHeader *>(sample.data() + naluStartIndex);
+        auto type = header->unitType();
+        switch (type) {
+            case 7:
+                previousUnitType7 = {sample.begin() + i, sample.begin() + naluEndIndex};
+                break;
+            case 8:
+                previousUnitType8 = {sample.begin() + i, sample.begin() + naluEndIndex};;
+                break;
+            case 5:
+                previousUnitType5 = {sample.begin() + i, sample.begin() + naluEndIndex};;
+                break;
+        }
+        i = naluEndIndex;
+    }
+}
+
+vector<byte> H264FileParser::initialNALUS() {
+    vector<byte> units{};
+    if (previousUnitType7.has_value()) {
+        auto nalu = previousUnitType7.value();
+        units.insert(units.end(), nalu.begin(), nalu.end());
+    }
+    if (previousUnitType8.has_value()) {
+        auto nalu = previousUnitType8.value();
+        units.insert(units.end(), nalu.begin(), nalu.end());
+    }
+    if (previousUnitType5.has_value()) {
+        auto nalu = previousUnitType5.value();
+        units.insert(units.end(), nalu.begin(), nalu.end());
+    }
+    return units;
+}

examples/streamer/h264fileparser.hpp

@@ -0,0 +1,36 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef h264fileparser_hpp
+#define h264fileparser_hpp
+
+#include "fileparser.hpp"
+#include <optional>
+
+class H264FileParser: public FileParser {
+    std::optional<std::vector<std::byte>> previousUnitType5 = std::nullopt;
+    std::optional<std::vector<std::byte>> previousUnitType7 = std::nullopt;
+    std::optional<std::vector<std::byte>> previousUnitType8 = std::nullopt;
+
+public:
+    H264FileParser(std::string directory, uint32_t fps, bool loop);
+    void loadNextSample() override;
+    std::vector<std::byte> initialNALUS();
+};
+
+#endif /* h264fileparser_hpp */

examples/streamer/helpers.cpp

@@ -0,0 +1,87 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "helpers.hpp"
+#include <ctime>
+
+#if _WIN32
+// taken from https://stackoverflow.com/questions/10905892/equivalent-of-gettimeday-for-windows
+
+#include <Windows.h>
+
+struct timezone {
+    int tz_minuteswest;
+    int tz_dsttime;
+};
+int gettimeofday(struct timeval *tv, struct timezone *tz)
+{
+    if (tv) {
+        FILETIME               filetime; /* 64-bit value representing the number of 100-nanosecond intervals since January 1, 1601 00:00 UTC */
+        ULARGE_INTEGER         x;
+        ULONGLONG              usec;
+        static const ULONGLONG epoch_offset_us = 11644473600000000ULL; /* microseconds betweeen Jan 1,1601 and Jan 1,1970 */
+
+#if _WIN32_WINNT >= _WIN32_WINNT_WIN8
+        GetSystemTimePreciseAsFileTime(&filetime);
+#else
+        GetSystemTimeAsFileTime(&filetime);
+#endif
+        x.LowPart =  filetime.dwLowDateTime;
+        x.HighPart = filetime.dwHighDateTime;
+        usec = x.QuadPart / 10  -  epoch_offset_us;
+        tv->tv_sec  = (time_t)(usec / 1000000ULL);
+        tv->tv_usec = (long)(usec % 1000000ULL);
+    }
+    if (tz) {
+        TIME_ZONE_INFORMATION timezone;
+        GetTimeZoneInformation(&timezone);
+        tz->tz_minuteswest = timezone.Bias;
+        tz->tz_dsttime = 0;
+    }
+    return 0;
+}
+#endif
+
+using namespace std;
+using namespace rtc;
+
+ClientTrackData::ClientTrackData(shared_ptr<Track> track, shared_ptr<RTCPSenderReportable> sender) {
+    this->track = track;
+    this->sender = sender;
+}
+
+void Client::setState(State state) {
+    std::unique_lock lock(_mutex);
+    this->state = state;
+}
+
+Client::State Client::getState() {
+    std::shared_lock lock(_mutex);
+    return state;
+}
+
+ClientTrack::ClientTrack(string id, shared_ptr<ClientTrackData> trackData) {
+    this->id = id;
+    this->trackData = trackData;
+}
+
+uint64_t currentTimeInMicroSeconds() {
+    struct timeval time;
+    gettimeofday(&time, NULL);
+    return uint64_t(time.tv_sec) * 1000 * 1000 + time.tv_usec;
+}

examples/streamer/helpers.hpp

@@ -0,0 +1,63 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef helpers_hpp
+#define helpers_hpp
+
+#include "rtc/rtc.hpp"
+
+struct ClientTrackData {
+    std::shared_ptr<rtc::Track> track;
+    std::shared_ptr<rtc::RTCPSenderReportable> sender;
+
+    ClientTrackData(std::shared_ptr<rtc::Track> track, std::shared_ptr<rtc::RTCPSenderReportable> sender);
+};
+
+struct Client {
+    enum class State {
+        Waiting,
+        WaitingForVideo,
+        WaitingForAudio,
+        Ready
+    };
+    const std::shared_ptr<rtc::PeerConnection> & peerConnection = _peerConnection;
+    Client(std::shared_ptr<rtc::PeerConnection> pc) {
+        _peerConnection = pc;
+    }
+    std::optional<std::shared_ptr<ClientTrackData>> video;
+    std::optional<std::shared_ptr<ClientTrackData>> audio;
+    std::optional<std::shared_ptr<rtc::DataChannel>> dataChannel{};
+    void setState(State state);
+    State getState();
+
+private:
+    std::shared_mutex _mutex;
+    State state = State::Waiting;
+    std::string id;
+    std::shared_ptr<rtc::PeerConnection> _peerConnection;
+};
+
+struct ClientTrack {
+    std::string id;
+    std::shared_ptr<ClientTrackData> trackData;
+    ClientTrack(std::string id, std::shared_ptr<ClientTrackData> trackData);
+};
+
+uint64_t currentTimeInMicroSeconds();
+
+#endif /* helpers_hpp */

examples/streamer/index.html

@@ -0,0 +1,73 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title>libdatachannel media example</title>
+    <style>
+        button {
+            padding: 8px 16px;
+        }
+
+        pre {
+            overflow-x: hidden;
+            overflow-y: auto;
+        }
+
+        video {
+            width: 100%;
+        }
+
+        .option {
+            margin-bottom: 8px;
+        }
+
+        #media {
+            max-width: 1280px;
+        }
+    </style>
+</head>
+<body>
+<script src="https://webrtc.github.io/adapter/adapter-latest.js"></script>
+
+<h2>Options</h2>
+
+<div class="option">
+    <input id="use-stun" type="checkbox"/>
+    <label for="use-stun">Use STUN server</label>
+</div>
+
+<button id="start" onclick="start()" disabled>Start</button>
+<button id="stop" style="display: none" onclick="stop()">Stop</button>
+
+<h2>State</h2>
+<p>
+    ICE gathering state: <span id="ice-gathering-state"></span>
+</p>
+<p>
+    ICE connection state: <span id="ice-connection-state"></span>
+</p>
+<p>
+    Signaling state: <span id="signaling-state"></span>
+</p>
+
+<div id="media" style="display: none">
+    <h2>Media</h2>
+    <audio id="audio" autoplay></audio>
+    <video id="video" autoplay playsinline></video>
+</div>
+
+<h2>Data channel</h2>
+<pre id="data-channel" style="height: 200px;"></pre>
+
+<h2>SDP</h2>
+
+<h3>Offer</h3>
+<pre id="offer-sdp"></pre>
+
+<h3>Answer</h3>
+<pre id="answer-sdp"></pre>
+
+<script src="client.js"></script>
+
+</body>
+</html>

examples/streamer/main.cpp

@@ -0,0 +1,468 @@
+/*
+ * libdatachannel client example
+ * Copyright (c) 2019-2020 Paul-Louis Ageneau
+ * Copyright (c) 2019 Murat Dogan
+ * Copyright (c) 2020 Will Munn
+ * Copyright (c) 2020 Nico Chatzi
+ * Copyright (c) 2020 Lara Mackey
+ * Copyright (c) 2020 Erik Cota-Robles
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "nlohmann/json.hpp"
+
+#include "h264fileparser.hpp"
+#include "opusfileparser.hpp"
+#include "helpers.hpp"
+#include "ArgParser.hpp"
+
+using namespace rtc;
+using namespace std;
+using namespace std::chrono_literals;
+
+using json = nlohmann::json;
+
+template <class T> weak_ptr<T> make_weak_ptr(shared_ptr<T> ptr) { return ptr; }
+
+/// all connected clients
+unordered_map<string, shared_ptr<Client>> clients{};
+
+/// Creates peer connection and client representation
+/// @param config Configuration
+/// @param wws Websocket for signaling
+/// @param id Client ID
+/// @returns Client
+shared_ptr<Client> createPeerConnection(const Configuration &config,
+                                        weak_ptr<WebSocket> wws,
+                                        string id);
+
+/// Creates stream
+/// @param h264Samples Directory with H264 samples
+/// @param fps Video FPS
+/// @param opusSamples Directory with opus samples
+/// @returns Stream object
+shared_ptr<Stream> createStream(const string h264Samples, const unsigned fps, const string opusSamples);
+
+/// Add client to stream
+/// @param client Client
+/// @param adding_video True if adding video
+void addToStream(shared_ptr<Client> client, bool isAddingVideo);
+
+/// Start stream
+void startStream();
+
+/// Main dispatch queue
+DispatchQueue MainThread("Main");
+
+/// Audio and video stream
+optional<shared_ptr<Stream>> avStream = nullopt;
+
+const string defaultRootDirectory = "../../../../examples/streamer/samples/";
+const string defaultH264SamplesDirectory = defaultRootDirectory + "h264/";
+string h264SamplesDirectory = defaultH264SamplesDirectory;
+const string defaultOpusSamplesDirectory = defaultRootDirectory + "opus/";
+string opusSamplesDirectory = defaultOpusSamplesDirectory;
+const string defaultIPAddress = "127.0.0.1";
+const uint16_t defaultPort = 8000;
+string ip_address = defaultIPAddress;
+uint16_t port = defaultPort;
+
+/// Incomming message handler for websocket
+/// @param message Incommint message
+/// @param config Configuration
+/// @param ws Websocket
+void wsOnMessage(json message, Configuration config, shared_ptr<WebSocket> ws) {
+    auto it = message.find("id");
+    if (it == message.end())
+        return;
+    string id = it->get<string>();
+    it = message.find("type");
+    if (it == message.end())
+        return;
+    string type = it->get<string>();
+
+    if (type == "streamRequest") {
+        shared_ptr<Client> c = createPeerConnection(config, make_weak_ptr(ws), id);
+        clients.emplace(id, c);
+    } else if (type == "answer") {
+        shared_ptr<Client> c;
+        if (auto jt = clients.find(id); jt != clients.end()) {
+            auto pc = clients.at(id)->peerConnection;
+            auto sdp = message["sdp"].get<string>();
+            auto description = Description(sdp, type);
+            pc->setRemoteDescription(description);
+        }
+    }
+}
+
+int main(int argc, char **argv) try {
+    bool enableDebugLogs = false;
+    bool printHelp = false;
+    int c = 0;
+    auto parser = ArgParser({{"a", "audio"}, {"b", "video"}, {"d", "ip"}, {"p","port"}}, {{"h", "help"}, {"v", "verbose"}});
+    auto parsingResult = parser.parse(argc, argv, [](string key, string value) {
+        if (key == "audio") {
+            opusSamplesDirectory = value + "/";
+        } else if (key == "video") {
+            h264SamplesDirectory = value + "/";
+        } else if (key == "ip") {
+            ip_address = value;
+        } else if (key == "port") {
+            port = atoi(value.data());
+        } else {
+            cerr << "Invalid option --" << key << " with value " << value << endl;
+            return false;
+        }
+        return true;
+    }, [&enableDebugLogs, &printHelp](string flag){
+        if (flag == "verbose") {
+            enableDebugLogs = true;
+        } else if (flag == "help") {
+            printHelp = true;
+        } else {
+            cerr << "Invalid flag --" << flag << endl;
+            return false;
+        }
+        return true;
+    });
+    if (!parsingResult) {
+        return 1;
+    }
+
+    if (printHelp) {
+        cout << "usage: stream-h264 [-a opus_samples_folder] [-b h264_samples_folder] [-d ip_address] [-p port] [-v] [-h]" << endl
+        << "Arguments:" << endl
+        << "\t -a " << "Directory with opus samples (default: " << defaultOpusSamplesDirectory << ")." << endl
+        << "\t -b " << "Directory with H264 samples (default: " << defaultH264SamplesDirectory << ")." << endl
+        << "\t -d " << "Signaling server IP address (default: " << defaultIPAddress << ")." << endl
+        << "\t -p " << "Signaling server port (default: " << defaultPort << ")." << endl
+        << "\t -v " << "Enable debug logs." << endl
+        << "\t -h " << "Print this help and exit." << endl;
+        return 0;
+    }
+    if (enableDebugLogs) {
+        InitLogger(LogLevel::Debug);
+    }
+
+    Configuration config;
+    string stunServer = "stun:stun.l.google.com:19302";
+    cout << "Stun server is " << stunServer << endl;
+    config.iceServers.emplace_back(stunServer);
+
+
+    string localId = "server";
+    cout << "The local ID is: " << localId << endl;
+
+    auto ws = make_shared<WebSocket>();
+    ws->onOpen([]() { cout << "WebSocket connected, signaling ready" << endl; });
+
+    ws->onClosed([]() { cout << "WebSocket closed" << endl; });
+
+    ws->onError([](const string &error) { cout << "WebSocket failed: " << error << endl; });
+
+    ws->onMessage([&](variant<binary, string> data) {
+        if (!holds_alternative<string>(data))
+            return;
+
+        json message = json::parse(get<string>(data));
+        MainThread.dispatch([message, config, ws]() {
+            wsOnMessage(message, config, ws);
+        });
+    });
+
+    const string url = "ws://" + ip_address + ":" + to_string(port) + "/" + localId;
+    cout << "Url is " << url << endl;
+    ws->open(url);
+
+    cout << "Waiting for signaling to be connected..." << endl;
+    while (!ws->isOpen()) {
+        if (ws->isClosed())
+            return 1;
+        this_thread::sleep_for(100ms);
+    }
+
+    while (true) {
+        string id;
+        cout << "Enter to exit" << endl;
+        cin >> id;
+        cin.ignore();
+        cout << "exiting" << endl;
+        break;
+    }
+
+    cout << "Cleaning up..." << endl;
+    return 0;
+
+} catch (const std::exception &e) {
+    std::cout << "Error: " << e.what() << std::endl;
+    return -1;
+}
+
+shared_ptr<ClientTrackData> addVideo(const shared_ptr<PeerConnection> pc, const uint8_t payloadType, const uint32_t ssrc, const string cname, const string msid, const function<void (void)> onOpen) {
+    auto video = Description::Video(cname);
+    video.addH264Codec(payloadType);
+    video.addSSRC(ssrc, cname, msid);
+    auto track = pc->addTrack(video);
+    // create RTP configuration
+    auto rtpConfig = shared_ptr<RTPPacketizationConfig>(new RTPPacketizationConfig(ssrc, cname, payloadType, H264RTPPacketizer::defaultClockRate));
+    // create packetizer
+    auto packetizer = shared_ptr<H264RTPPacketizer>(new H264RTPPacketizer(rtpConfig));
+    // create H264 and RTCP SP handler
+    shared_ptr<H264PacketizationHandler> h264Handler(new H264PacketizationHandler(H264PacketizationHandler::Separator::Length, packetizer));
+    // set handler
+    track->setRtcpHandler(h264Handler);
+    track->onOpen(onOpen);
+    auto trackData = make_shared<ClientTrackData>(track, h264Handler);
+    return trackData;
+}
+
+shared_ptr<ClientTrackData> addAudio(const shared_ptr<PeerConnection> pc, const uint8_t payloadType, const uint32_t ssrc, const string cname, const string msid, const function<void (void)> onOpen) {
+    auto audio = Description::Audio(cname);
+    audio.addOpusCodec(payloadType);
+    audio.addSSRC(ssrc, cname, msid);
+    auto track = pc->addTrack(audio);
+    // create RTP configuration
+    auto rtpConfig = shared_ptr<RTPPacketizationConfig>(new RTPPacketizationConfig(ssrc, cname, payloadType, OpusRTPPacketizer::defaultClockRate));
+    // create packetizer
+    auto packetizer = make_shared<OpusRTPPacketizer>(rtpConfig);
+    // create opus and RTCP SP handler
+    auto opusHandler = make_shared<OpusPacketizationHandler>(packetizer);
+    // set handler
+    track->setRtcpHandler(opusHandler);
+    track->onOpen(onOpen);
+    auto trackData = make_shared<ClientTrackData>(track, opusHandler);
+    return trackData;
+}
+
+// Create and setup a PeerConnection
+shared_ptr<Client> createPeerConnection(const Configuration &config,
+                                                weak_ptr<WebSocket> wws,
+                                                string id) {
+
+    auto pc = make_shared<PeerConnection>(config);
+    shared_ptr<Client> client(new Client(pc));
+
+    pc->onStateChange([id](PeerConnection::State state) {
+        cout << "State: " << state << endl;
+        if (state == PeerConnection::State::Disconnected ||
+            state == PeerConnection::State::Failed ||
+            state == PeerConnection::State::Closed) {
+            // remove disconnected client
+            MainThread.dispatch([id]() {
+                clients.erase(id);
+            });
+        }
+    });
+
+    pc->onGatheringStateChange(
+        [wpc = make_weak_ptr(pc), id, wws](PeerConnection::GatheringState state) {
+        cout << "Gathering State: " << state << endl;
+        if (state == PeerConnection::GatheringState::Complete) {
+            if(auto pc = wpc.lock()) {
+                auto description = pc->localDescription();
+                json message = {
+                    {"id", id},
+                    {"type", description->typeString()},
+                    {"sdp", string(description.value())}
+                };
+                // Gathering complete, send answer
+                if (auto ws = wws.lock()) {
+                    ws->send(message.dump());
+                }
+            }
+        }
+    });
+
+    client->video = addVideo(pc, 102, 1, "video-stream", "stream1", [id, wc = make_weak_ptr(client)]() {
+        MainThread.dispatch([wc]() {
+            if (auto c = wc.lock()) {
+                addToStream(c, true);
+            }
+        });
+        cout << "Video from " << id << " opened" << endl;
+    });
+
+    client->audio = addAudio(pc, 111, 2, "audio-stream", "stream1", [id, wc = make_weak_ptr(client)]() {
+        MainThread.dispatch([wc]() {
+            if (auto c = wc.lock()) {
+                addToStream(c, false);
+            }
+        });
+        cout << "Audio from " << id << " opened" << endl;
+    });
+
+    auto dc = pc->addDataChannel("ping-pong");
+    dc->onOpen([id, wdc = make_weak_ptr(dc)]() {
+        if (auto dc = wdc.lock()) {
+            dc->send("Ping");
+        }
+    });
+
+    dc->onMessage(nullptr, [id, wdc = make_weak_ptr(dc)](string msg) {
+        cout << "Message from " << id << " received: " << msg << endl;
+        if (auto dc = wdc.lock()) {
+            dc->send("Ping");
+        }
+    });
+    client->dataChannel = dc;
+
+    pc->setLocalDescription();
+    return client;
+};
+
+/// Create stream
+shared_ptr<Stream> createStream(const string h264Samples, const unsigned fps, const string opusSamples) {
+    // video source
+    auto video = make_shared<H264FileParser>(h264Samples, fps, true);
+    // audio source
+    auto audio = make_shared<OPUSFileParser>(opusSamples, true);
+
+    auto stream = make_shared<Stream>(video, audio);
+    // set callback responsible for sample sending
+    stream->onSample([ws = make_weak_ptr(stream)](Stream::StreamSourceType type, uint64_t sampleTime, rtc::binary sample) {
+        vector<ClientTrack> tracks{};
+        string streamType = type == Stream::StreamSourceType::Video ? "video" : "audio";
+        // get track for given type
+        function<optional<shared_ptr<ClientTrackData>> (shared_ptr<Client>)> getTrackData = [type](shared_ptr<Client> client) {
+            return type == Stream::StreamSourceType::Video ? client->video : client->audio;
+        };
+        // get all clients with Ready state
+        for(auto id_client: clients) {
+            auto id = id_client.first;
+            auto client = id_client.second;
+            auto optTrackData = getTrackData(client);
+            if (client->getState() == Client::State::Ready && optTrackData.has_value()) {
+                auto trackData = optTrackData.value();
+                tracks.push_back(ClientTrack(id, trackData));
+            }
+        }
+        if (!tracks.empty()) {
+            auto message = make_message(move(sample));
+            for (auto clientTrack: tracks) {
+                auto client = clientTrack.id;
+                auto trackData = clientTrack.trackData;
+                // sample time is in us, we need to convert it to seconds
+                auto elapsedSeconds = double(sampleTime) / (1000 * 1000);
+                auto rtpConfig = trackData->sender->rtpConfig;
+                // get elapsed time in clock rate
+                uint32_t elapsedTimestamp = rtpConfig->secondsToTimestamp(elapsedSeconds);
+
+                // set new timestamp
+                rtpConfig->timestamp = rtpConfig->startTimestamp + elapsedTimestamp;
+
+                // get elapsed time in clock rate from last RTCP sender report
+                auto reportElapsedTimestamp = rtpConfig->timestamp - trackData->sender->previousReportedTimestamp;
+                // check if last report was at least 1 second ago
+                if (rtpConfig->timestampToSeconds(reportElapsedTimestamp) > 1) {
+                    trackData->sender->setNeedsToReport();
+                }
+                cout << "Sending " << streamType << " sample with size: " << to_string(message->size()) << " to " << client << endl;
+                bool send = false;
+                try {
+                    // send sample
+                    send = trackData->track->send(*message);
+                } catch (...) {
+                    send = false;
+                }
+                if (!send) {
+                    cerr << "Unable to send "<< streamType << " packet" << endl;
+                    break;
+                }
+            }
+        }
+        MainThread.dispatch([ws]() {
+            if (clients.empty()) {
+                // we have no clients, stop the stream
+                if (auto stream = ws.lock()) {
+                    stream->stop();
+                }
+            }
+        });
+    });
+    return stream;
+}
+
+/// Start stream
+void startStream() {
+    shared_ptr<Stream> stream;
+    if (avStream.has_value()) {
+        stream = avStream.value();
+        if (stream->isRunning) {
+            // stream is already running
+            return;
+        }
+    } else {
+        stream = createStream(h264SamplesDirectory, 30, opusSamplesDirectory);
+        avStream = stream;
+    }
+    stream->start();
+}
+
+/// Send previous key frame so browser can show something to user
+/// @param stream Stream
+/// @param video Video track data
+void sendInitialNalus(shared_ptr<Stream> stream, shared_ptr<ClientTrackData> video) {
+    auto h264 = dynamic_cast<H264FileParser *>(stream->video.get());
+    auto initialNalus = h264->initialNALUS();
+
+    // send previous NALU key frame so users don't have to wait to see stream works
+    if (!initialNalus.empty()) {
+        const double frameDuration_s = double(h264->sampleDuration_us) / (1000 * 1000);
+        const uint32_t frameTimestampDuration = video->sender->rtpConfig->secondsToTimestamp(frameDuration_s);
+        video->sender->rtpConfig->timestamp = video->sender->rtpConfig->startTimestamp - frameTimestampDuration * 2;
+        video->track->send(initialNalus);
+        video->sender->rtpConfig->timestamp += frameTimestampDuration;
+        // Send initial NAL units again to start stream in firefox browser
+        video->track->send(initialNalus);
+    }
+}
+
+/// Add client to stream
+/// @param client Client
+/// @param adding_video True if adding video
+void addToStream(shared_ptr<Client> client, bool isAddingVideo) {
+    if (client->getState() == Client::State::Waiting) {
+        client->setState(isAddingVideo ? Client::State::WaitingForAudio : Client::State::WaitingForVideo);
+    } else if ((client->getState() == Client::State::WaitingForAudio && !isAddingVideo)
+               || (client->getState() == Client::State::WaitingForVideo && isAddingVideo)) {
+
+        // Audio and video tracks are collected now
+        assert(client->video.has_value() && client->audio.has_value());
+
+        auto video = client->video.value();
+        auto audio = client->audio.value();
+
+        auto currentTime_us = double(currentTimeInMicroSeconds());
+        auto currentTime_s = currentTime_us / (1000 * 1000);
+
+        // set start time of stream
+        video->sender->rtpConfig->setStartTime(currentTime_s, RTPPacketizationConfig::EpochStart::T1970);
+        audio->sender->rtpConfig->setStartTime(currentTime_s, RTPPacketizationConfig::EpochStart::T1970);
+
+        // start stat recording of RTCP SR
+        video->sender->startRecording();
+        audio->sender->startRecording();
+
+        if (avStream.has_value()) {
+            sendInitialNalus(avStream.value(), video);
+        }
+
+        client->setState(Client::State::Ready);
+    }
+    if (client->getState() == Client::State::Ready) {
+        startStream();
+    }
+}

examples/streamer/opusfileparser.cpp

@@ -0,0 +1,23 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "opusfileparser.hpp"
+
+using namespace std;
+
+OPUSFileParser::OPUSFileParser(string directory, bool loop, uint32_t samplesPerSecond): FileParser(directory, ".opus", samplesPerSecond, loop) { }

examples/streamer/opusfileparser.hpp

@@ -0,0 +1,32 @@
+/*
+ * libdatachannel streamer example
+ * Copyright (c) 2020 Filip Klembara (in2core)
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef opusfileparser_hpp
+#define opusfileparser_hpp
+
+#include "fileparser.hpp"
+
+class OPUSFileParser: public FileParser {
+    static const uint32_t defaultSamplesPerSecond = 50;
+
+public:
+    OPUSFileParser(std::string directory, bool loop, uint32_t samplesPerSecond = OPUSFileParser::defaultSamplesPerSecond);
+};
+
+
+#endif /* opusfileparser_hpp */

examples/streamer/samples/generate_h264.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+
+import os
+import getopt
+import sys
+import glob
+from functools import reduce
+from typing import Optional, List
+
+
+class H264ByteStream:
+    @staticmethod
+    def nalu_type(nalu: bytes) -> int:
+        return nalu[0] & 0x1F
+
+    @staticmethod
+    def merge_sample(sample: List[bytes]) -> bytes:
+        result = bytes()
+        for nalu in sample:
+            result += len(nalu).to_bytes(4, byteorder='big') + nalu
+        return result
+
+    @staticmethod
+    def reduce_nalus_to_samples(samples: List[List[bytes]], current: bytes) -> List[List[bytes]]:
+        last_nalus = samples[-1]
+        samples[-1] = last_nalus + [current]
+        if H264ByteStream.nalu_type(current) in [1, 5]:
+            samples.append([])
+        return samples
+
+    def __init__(self, file_name: str):
+        with open(file_name, "rb") as file:
+            byte_stream = file.read()
+            long_split = byte_stream.split(b"\x00\x00\x00\x01")
+            splits = reduce(lambda acc, x: acc + x.split(b"\x00\x00\x01"), long_split, [])
+            nalus = filter(lambda x: len(x) > 0, splits)
+            self.samples = list(
+                filter(lambda x: len(x) > 0, reduce(H264ByteStream.reduce_nalus_to_samples, nalus, [[]])))
+
+
+def generate(input_file: str, output_dir: str, max_samples: Optional[int], fps: Optional[int]):
+    if output_dir[-1] != "/":
+        output_dir += "/"
+    if os.path.isdir(output_dir):
+        files_to_delete = glob.glob(output_dir + "*.h264")
+        if len(files_to_delete) > 0:
+            print("Remove following files?")
+            for file in files_to_delete:
+                print(file)
+            response = input("Remove files? [y/n] ").lower()
+            if response != "y" and response != "yes":
+                print("Cancelling...")
+                return
+            print("Removing files")
+            for file in files_to_delete:
+                os.remove(file)
+    else:
+        os.makedirs(output_dir, exist_ok=True)
+    video_stream_file = "_video_stream.h264"
+    if os.path.isfile(video_stream_file):
+        os.remove(video_stream_file)
+
+    fps_line = "" if fps is None else "-filter:v fps=fps={} ".format(fps)
+    command = 'ffmpeg -i {} -an -vcodec libx264 -preset slow -profile baseline {}{}'.format(input_file, fps_line,
+                                                                                            video_stream_file)
+    os.system(command)
+
+    data = H264ByteStream(video_stream_file)
+    index = 0
+    for sample in data.samples[:max_samples]:
+        name = "{}sample-{}.h264".format(output_dir, index)
+        index += 1
+        with open(name, 'wb') as file:
+            merged_sample = H264ByteStream.merge_sample(sample)
+            file.write(merged_sample)
+    os.remove(video_stream_file)
+
+
+def main(argv):
+    input_file = None
+    default_output_dir = "h264/"
+    output_dir = default_output_dir
+    max_samples = None
+    fps = None
+    try:
+        opts, args = getopt.getopt(argv, "hi:o:m:f:", ["help", "ifile=", "odir=", "max=", "fps"])
+    except getopt.GetoptError:
+        print('generate_h264.py -i <input_files> [-o <output_files>] [-m <max_samples>] [-f <fps>] [-h]')
+        sys.exit(2)
+    for opt, arg in opts:
+        if opt in ("-h", "--help"):
+            print("Usage: generate_h264.py -i <input_files> [-o <output_files>] [-m <max_samples>] [-f <fps>] [-h]")
+            print("Arguments:")
+            print("\t-i,--ifile: Input file")
+            print("\t-o,--odir: Output directory (default: " + default_output_dir + ")")
+            print("\t-m,--max: Maximum generated samples")
+            print("\t-f,--fps: Output fps")
+            print("\t-h,--help: Print this help and exit")
+            sys.exit()
+        elif opt in ("-i", "--ifile"):
+            input_file = arg
+        elif opt in ("-o", "--ofile"):
+            output_dir = arg
+        elif opt in ("-m", "--max"):
+            max_samples = int(arg)
+        elif opt in ("-f", "--fps"):
+            fps = int(arg)
+    if input_file is None:
+        print("Missing argument -i")
+        sys.exit(2)
+    generate(input_file, output_dir, max_samples, fps)
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])

examples/streamer/samples/generate_opus.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+
+from kaitaistruct import KaitaiStruct, ValidationNotEqualError
+import os
+import getopt
+import sys
+import glob
+from functools import reduce
+
+
+class Ogg(KaitaiStruct):
+    """Ogg is a popular media container format, which provides basic
+    streaming / buffering mechanisms and is content-agnostic. Most
+    popular codecs that are used within Ogg streams are Vorbis (thus
+    making Ogg/Vorbis streams) and Theora (Ogg/Theora).
+
+    Ogg stream is a sequence Ogg pages. They can be read sequentially,
+    or one can jump into arbitrary stream location and scan for "OggS"
+    sync code to find the beginning of a new Ogg page and continue
+    decoding the stream contents from that one.
+    """
+
+    def __init__(self, _io, _parent=None, _root=None):
+        KaitaiStruct.__init__(self, _io)
+        self._parent = _parent
+        self._root = _root if _root else self
+        self._read()
+
+    def _read(self):
+        self.pages = []
+        i = 0
+        while not self._io.is_eof():
+            self.pages.append(Ogg.Page(self._io, self, self._root))
+            i += 1
+
+    class Page(KaitaiStruct):
+        """Ogg page is a basic unit of data in an Ogg bitstream, usually
+        it's around 4-8 KB, with a maximum size of 65307 bytes.
+        """
+
+        def __init__(self, _io, _parent=None, _root=None):
+            KaitaiStruct.__init__(self, _io)
+            self._parent = _parent
+            self._root = _root if _root else self
+            self._read()
+
+        def _read(self):
+            self.sync_code = self._io.read_bytes(4)
+            if not self.sync_code == b"\x4F\x67\x67\x53":
+                raise ValidationNotEqualError(b"\x4F\x67\x67\x53", self.sync_code, self._io,
+                                              u"/types/page/seq/0")
+            self.version = self._io.read_bytes(1)
+            if not self.version == b"\x00":
+                raise ValidationNotEqualError(b"\x00", self.version, self._io, u"/types/page/seq/1")
+            self.reserved1 = self._io.read_bits_int_be(5)
+            self.is_end_of_stream = self._io.read_bits_int_be(1) != 0
+            self.is_beginning_of_stream = self._io.read_bits_int_be(1) != 0
+            self.is_continuation = self._io.read_bits_int_be(1) != 0
+            self._io.align_to_byte()
+            self.granule_pos = self._io.read_u8le()
+            self.bitstream_serial = self._io.read_u4le()
+            self.page_seq_num = self._io.read_u4le()
+            self.crc32 = self._io.read_u4le()
+            self.num_segments = self._io.read_u1()
+            self.len_segments = [None] * self.num_segments
+            for i in range(self.num_segments):
+                self.len_segments[i] = self._io.read_u1()
+
+            self.segments = [None] * self.num_segments
+            for i in range(self.num_segments):
+                self.segments[i] = self._io.read_bytes(self.len_segments[i])
+
+
+def generate(input_file: str, output_dir: str, max_samples: int):
+    if output_dir[-1] != "/":
+        output_dir += "/"
+    if os.path.isdir(output_dir):
+        files_to_delete = glob.glob(output_dir + "*.opus")
+        if len(files_to_delete) > 0:
+            print("Remove following files?")
+            for file in files_to_delete:
+                print(file)
+            response = input("Remove files? [y/n] ").lower()
+            if response != "y" and response != "yes":
+                print("Cancelling...")
+                return
+            print("Removing files")
+            for file in files_to_delete:
+                os.remove(file)
+    else:
+        os.makedirs(output_dir, exist_ok=True)
+    audio_stream_file = "_audio_stream.ogg"
+    if os.path.isfile(audio_stream_file):
+        os.remove(audio_stream_file)
+    os.system('ffmpeg -i {} -vn -ar 48000 -ac 2 -vbr off -acodec libopus -ab 64k {}'.format(input_file, audio_stream_file))
+
+    data = Ogg.from_file(audio_stream_file)
+    index = 0
+    valid_pages = data.pages[2:]
+    segments = list(reduce(lambda x, y: x + y.segments, valid_pages, []))[:max_samples]
+    for segment in segments:
+        name = "{}sample-{}.opus".format(output_dir, index)
+        index += 1
+        with open(name, 'wb') as file:
+            assert len(list(segment)) == 160
+            file.write(segment)
+    os.remove(audio_stream_file)
+
+
+def main(argv):
+    input_file = None
+    default_output_dir = "opus/"
+    output_dir = default_output_dir
+    max_samples = None
+    try:
+        opts, args = getopt.getopt(argv, "hi:o:m:", ["help", "ifile=", "odir=", "max="])
+    except getopt.GetoptError:
+        print('generate_opus.py -i <input_files> [-o <output_files>] [-m <max_samples>] [-h]')
+        sys.exit(2)
+    for opt, arg in opts:
+        if opt in ("-h", "--help"):
+            print("Usage: generate_opus.py -i <input_files> [-o <output_files>] [-m <max_samples>] [-h]")
+            print("Arguments:")
+            print("\t-i,--ifile: Input file")
+            print("\t-o,--odir: Output directory (default: " + default_output_dir + ")")
+            print("\t-m,--max: Maximum generated samples")
+            print("\t-h,--help: Print this help and exit")
+            sys.exit()
+        elif opt in ("-i", "--ifile"):
+            input_file = arg
+        elif opt in ("-o", "--ofile"):
+            output_dir = arg
+        elif opt in ("-m", "--max"):
+            max_samples = int(arg)
+    if input_file is None:
+        print("Missing argument -i")
+        sys.exit(2)
+    generate(input_file, output_dir, max_samples)
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])