Merge feature/plan-4-gateways: RTMP-RoQR gateways, media relay, ffmpeg e2e, and CI

Author: mondain

.github/workflows/ci.yml

@@ -0,0 +1,54 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  core:
+    name: core-only (no picoquic)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install build tools
+        run: sudo apt-get update && sudo apt-get install -y cmake g++ ninja-build
+      - name: Configure (core only)
+        run: cmake -S . -B build -DCMAKE_BUILD_TYPE=Debug
+             -DROQR_BUILD_QUIC=OFF -DROQR_BUILD_RTMP=ON
+             -DROQR_BUILD_EXAMPLES=OFF -DROQR_BUILD_TESTS=ON
+      - name: Build
+        run: cmake --build build --parallel
+      - name: Test
+        run: ctest --test-dir build --output-on-failure
+
+  full:
+    name: full (${{ matrix.cc }})
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - { cc: gcc, cxx: g++ }
+          - { cc: clang, cxx: clang++ }
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install dependencies
+        run: sudo apt-get update && sudo apt-get install -y
+             cmake g++ clang ninja-build libssl-dev ffmpeg
+      - name: Build picoquic deps
+        run: eval "$(scripts/setup_picoquic_deps.sh)"; echo "$ROQR_PICOQUIC_SOURCE_DIR"
+      - name: Configure
+        env:
+          CC: ${{ matrix.cc }}
+          CXX: ${{ matrix.cxx }}
+        run: |
+          eval "$(scripts/setup_picoquic_deps.sh)"
+          cmake -S . -B build -DCMAKE_BUILD_TYPE=Debug \
+            -DROQR_BUILD_QUIC=ON -DROQR_BUILD_RTMP=ON \
+            -DROQR_BUILD_TOOLS=ON -DROQR_BUILD_EXAMPLES=ON \
+            -DROQR_BUILD_TESTS=ON
+      - name: Build
+        run: cmake --build build --parallel
+      - name: Test
+        run: ctest --test-dir build --output-on-failure

CMakeLists.txt

@@ -9,28 +9,50 @@ set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_EXTENSIONS OFF)
 
+# Collocate all built executables (roqr-relayd, roqr-ingest, roqr-egress,
+# roqr-duplex, test binaries) in a single directory so tooling (e.g. the
+# ffmpeg e2e test) can find the gateway trio via one bin dir.
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/bin")
+
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
 
 option(ROQR_BUILD_TESTS "Build unit tests" ON)
 option(ROQR_BUILD_QUIC "Build the picoquic transport (needs picoquic)" OFF)
 option(ROQR_BUILD_TOOLS "Build tools (roqr-relayd; needs ROQR_BUILD_QUIC)" OFF)
 option(ROQR_BUILD_RTMP "Build the RTMP/AMF gateway module" ON)
+option(ROQR_BUILD_EXAMPLES "Build gateway library and example apps" ON)
 
 add_subdirectory(core)
 
 if(ROQR_BUILD_RTMP)
   add_subdirectory(rtmp)
 endif()
 
+# The gateway library is needed by both the example binaries and the
+# relay's Media mode, so build it whenever its dependencies are present.
+if(ROQR_BUILD_QUIC AND ROQR_BUILD_RTMP)
+  add_subdirectory(gateway)
+endif()
+
+if(ROQR_BUILD_EXAMPLES)
+  if(NOT ROQR_BUILD_QUIC OR NOT ROQR_BUILD_RTMP)
+    message(FATAL_ERROR "ROQR_BUILD_EXAMPLES requires ROQR_BUILD_QUIC and ROQR_BUILD_RTMP")
+  endif()
+  if(ROQR_BUILD_TESTS AND NOT ROQR_BUILD_TOOLS)
+    message(FATAL_ERROR "ROQR_BUILD_EXAMPLES with ROQR_BUILD_TESTS requires ROQR_BUILD_TOOLS")
+  endif()
+  add_subdirectory(examples)
+endif()
+
 if(ROQR_BUILD_QUIC)
   find_package(Picoquic REQUIRED)
   find_package(OpenSSL REQUIRED)
   add_subdirectory(quic)
 endif()
 
 if(ROQR_BUILD_TOOLS)
-  if(NOT ROQR_BUILD_QUIC)
-    message(FATAL_ERROR "ROQR_BUILD_TOOLS requires ROQR_BUILD_QUIC=ON")
+  if(NOT ROQR_BUILD_QUIC OR NOT ROQR_BUILD_RTMP)
+    message(FATAL_ERROR "ROQR_BUILD_TOOLS requires ROQR_BUILD_QUIC and ROQR_BUILD_RTMP")
   endif()
   add_subdirectory(tools/relayd)
 endif()

README.md

@@ -0,0 +1,37 @@
+# libroqr
+
+A C++20 implementation of RoQR (RTMP over QUIC,
+draft-gregoire-rtmp-over-quic) with a sans-I/O protocol core, a picoquic
+transport, an RTMP/AMF gateway module, a test relay, and example gateways.
+
+## Reference media path
+
+```
+ffmpeg (RTMP publish) -> roqr-ingest -> roqr-relayd -> roqr-egress -> ffmpeg (RTMP play)
+```
+
+RoQR carries RTMP message metadata and payloads over QUIC streams and
+DATAGRAM frames.
+
+## Build
+
+```
+eval "$(scripts/setup_picoquic_deps.sh)"   # clone + build pinned picoquic/picotls
+cmake --preset dev
+cmake --build --preset dev
+ctest --preset dev
+```
+
+The core protocol library builds without picoquic:
+`cmake -S . -B build -DROQR_BUILD_QUIC=OFF -DROQR_BUILD_EXAMPLES=OFF && cmake --build build`.
+
+## Layout
+
+- `core/` sans-I/O RoQR frame codec, flow table (no dependencies)
+- `quic/` picoquic client transport
+- `rtmp/` RTMP handshake, chunking, AMF0, E-RTMP media classifier
+- `gateway/` RTMP<->RoQR bridge, ingest/egress gateways
+- `tools/relayd/` the RoQR test relay
+- `examples/` roqr-ingest, roqr-egress, roqr-duplex
+
+License: Apache-2.0.

examples/CMakeLists.txt

@@ -0,0 +1,13 @@
+add_executable(roqr-ingest ingest_main.cpp)
+target_link_libraries(roqr-ingest PRIVATE roqr-gateway)
+
+add_executable(roqr-egress egress_main.cpp)
+target_link_libraries(roqr-egress PRIVATE roqr-gateway)
+
+add_executable(roqr-duplex duplex_main.cpp)
+target_link_libraries(roqr-duplex PRIVATE roqr-gateway)
+
+foreach(t roqr-ingest roqr-egress roqr-duplex)
+  target_compile_options(${t} PRIVATE
+    $<$<CXX_COMPILER_ID:GNU,Clang>:-Wall -Wextra>)
+endforeach()

examples/duplex_main.cpp

@@ -0,0 +1,52 @@
+#include <csignal>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+
+#include "roqr/gateway/egress.hpp"
+#include "roqr/gateway/ingest.hpp"
+
+namespace {
+volatile std::sig_atomic_t g_stop = 0;
+void on_signal(int) { g_stop = 1; }
+}  // namespace
+
+int main(int argc, char** argv) {
+    roqr::gateway::IngestOptions in;
+    roqr::gateway::EgressOptions eg;
+    in.rtmp_port = 1935;
+    eg.rtmp_port = 1936;
+    for (int i = 1; i < argc; ++i) {
+        if (!std::strcmp(argv[i], "--roqr-host") && i + 1 < argc) {
+            in.roqr_host = eg.roqr_host = argv[++i];
+        } else if (!std::strcmp(argv[i], "--roqr-port") && i + 1 < argc) {
+            in.roqr_port = eg.roqr_port =
+                static_cast<uint16_t>(std::atoi(argv[++i]));
+        } else if (!std::strcmp(argv[i], "--stream") && i + 1 < argc) {
+            eg.stream_name = argv[++i];
+        } else {
+            std::fprintf(stderr,
+                         "usage: roqr-duplex [--roqr-host H] "
+                         "[--roqr-port P] [--stream NAME]\n");
+            return 2;
+        }
+    }
+    roqr::gateway::IngestGateway ingest;
+    roqr::gateway::EgressGateway egress;
+    if (!ingest.start(in) || !egress.start(eg)) {
+        std::fprintf(stderr, "duplex: failed to start\n");
+        return 1;
+    }
+    std::printf("roqr-duplex: ingest RTMP :%u, egress RTMP :%u, RoQR %s:%u\n",
+                in.rtmp_port, eg.rtmp_port, in.roqr_host.c_str(),
+                in.roqr_port);
+    std::signal(SIGINT, on_signal);
+    std::signal(SIGTERM, on_signal);
+    while (!g_stop) {
+        struct timespec ts {0, 200'000'000};
+        nanosleep(&ts, nullptr);
+    }
+    ingest.stop();
+    egress.stop();
+    return 0;
+}

examples/egress_main.cpp

@@ -0,0 +1,46 @@
+#include <csignal>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+
+#include "roqr/gateway/egress.hpp"
+
+namespace {
+volatile std::sig_atomic_t g_stop = 0;
+void on_signal(int) { g_stop = 1; }
+}  // namespace
+
+int main(int argc, char** argv) {
+    roqr::gateway::EgressOptions o;
+    for (int i = 1; i < argc; ++i) {
+        if (!std::strcmp(argv[i], "--rtmp-port") && i + 1 < argc) {
+            o.rtmp_port = static_cast<uint16_t>(std::atoi(argv[++i]));
+        } else if (!std::strcmp(argv[i], "--roqr-host") && i + 1 < argc) {
+            o.roqr_host = argv[++i];
+        } else if (!std::strcmp(argv[i], "--roqr-port") && i + 1 < argc) {
+            o.roqr_port = static_cast<uint16_t>(std::atoi(argv[++i]));
+        } else if (!std::strcmp(argv[i], "--stream") && i + 1 < argc) {
+            o.stream_name = argv[++i];
+        } else {
+            std::fprintf(stderr,
+                         "usage: roqr-egress [--rtmp-port P] "
+                         "[--roqr-host H] [--roqr-port P] [--stream NAME]\n");
+            return 2;
+        }
+    }
+    roqr::gateway::EgressGateway g;
+    if (!g.start(o)) {
+        std::fprintf(stderr, "egress: failed to start\n");
+        return 1;
+    }
+    std::printf("roqr-egress: RTMP :%u <- RoQR %s:%u (stream: %s)\n", o.rtmp_port,
+                o.roqr_host.c_str(), o.roqr_port, o.stream_name.c_str());
+    std::signal(SIGINT, on_signal);
+    std::signal(SIGTERM, on_signal);
+    while (!g_stop) {
+        struct timespec ts {0, 200'000'000};
+        nanosleep(&ts, nullptr);
+    }
+    g.stop();
+    return 0;
+}

examples/ingest_main.cpp

@@ -0,0 +1,44 @@
+#include <csignal>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+
+#include "roqr/gateway/ingest.hpp"
+
+namespace {
+volatile std::sig_atomic_t g_stop = 0;
+void on_signal(int) { g_stop = 1; }
+}  // namespace
+
+int main(int argc, char** argv) {
+    roqr::gateway::IngestOptions o;
+    for (int i = 1; i < argc; ++i) {
+        if (!std::strcmp(argv[i], "--rtmp-port") && i + 1 < argc) {
+            o.rtmp_port = static_cast<uint16_t>(std::atoi(argv[++i]));
+        } else if (!std::strcmp(argv[i], "--roqr-host") && i + 1 < argc) {
+            o.roqr_host = argv[++i];
+        } else if (!std::strcmp(argv[i], "--roqr-port") && i + 1 < argc) {
+            o.roqr_port = static_cast<uint16_t>(std::atoi(argv[++i]));
+        } else {
+            std::fprintf(stderr,
+                         "usage: roqr-ingest [--rtmp-port P] "
+                         "[--roqr-host H] [--roqr-port P]\n");
+            return 2;
+        }
+    }
+    roqr::gateway::IngestGateway g;
+    if (!g.start(o)) {
+        std::fprintf(stderr, "ingest: failed to start\n");
+        return 1;
+    }
+    std::printf("roqr-ingest: RTMP :%u -> RoQR %s:%u\n", o.rtmp_port,
+                o.roqr_host.c_str(), o.roqr_port);
+    std::signal(SIGINT, on_signal);
+    std::signal(SIGTERM, on_signal);
+    while (!g_stop) {
+        struct timespec ts {0, 200'000'000};
+        nanosleep(&ts, nullptr);
+    }
+    g.stop();
+    return 0;
+}

gateway/CMakeLists.txt

@@ -0,0 +1,16 @@
+add_library(roqr-gateway STATIC
+  src/bridge.cpp
+  src/rtmp_commands.cpp
+  src/ingest.cpp
+  src/egress.cpp
+)
+
+target_include_directories(roqr-gateway PUBLIC
+  $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
+)
+
+target_link_libraries(roqr-gateway PUBLIC roqr-core roqr-quic roqr-rtmp)
+
+target_compile_features(roqr-gateway PUBLIC cxx_std_20)
+target_compile_options(roqr-gateway PRIVATE
+  $<$<CXX_COMPILER_ID:GNU,Clang>:-Wall -Wextra>)

gateway/include/roqr/gateway/bridge.hpp

@@ -0,0 +1,22 @@
+#pragma once
+
+#include <cstdint>
+
+#include "roqr/frame.hpp"
+#include "roqr/rtmp/message.hpp"
+
+namespace roqr::gateway {
+
+// Widen an RTMP message into a RoQR frame on the given flow. All RTMP
+// metadata fits in the wider RoQR fields, so this never fails. The RoQR
+// timestamp carries the fully resolved RTMP message timestamp (draft
+// s7.3); the caller must not send a frame with an empty payload (RoQR
+// requires Payload Length > 0).
+roqr::Frame to_frame(const roqr::rtmp::RtmpMessage& msg, uint64_t flow_id);
+
+// Narrow a RoQR frame back into an RTMP message. Returns false and leaves
+// out untouched if timestamp, message_stream_id, or chunk_stream_id exceeds
+// 0xFFFFFFFF (the Width bridge rule).
+bool to_rtmp(const roqr::Frame& frame, roqr::rtmp::RtmpMessage& out);
+
+}  // namespace roqr::gateway

gateway/include/roqr/gateway/egress.hpp

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <string>
+
+namespace roqr::gateway {
+
+struct EgressOptions {
+    uint16_t rtmp_port = 1936;
+    std::string roqr_host = "127.0.0.1";
+    uint16_t roqr_port = 4443;
+    std::string stream_name = "cam";
+    bool insecure_skip_verify = true;
+};
+
+// Accepts one RTMP player (ffplay) on rtmp_port, connects to the RoQR
+// server, plays stream_name, and serves received media to the player.
+// Applies draft s8 gap recovery: after a suspected datagram gap, drops
+// non-keyframe video until the next keyframe/sequence header.
+class EgressGateway {
+public:
+    EgressGateway();
+    ~EgressGateway();
+    bool start(const EgressOptions& options);
+    void stop();
+    bool wait_playing(std::chrono::milliseconds timeout);
+
+private:
+    struct Impl;
+    std::unique_ptr<Impl> impl_;
+};
+
+}  // namespace roqr::gateway