From 2632f4a4a8f67f77655db61c9b6acc2de6d821a9 Mon Sep 17 00:00:00 2001 From: Haffi Mazhar Date: Sat, 7 Feb 2026 17:50:44 +0000 Subject: [PATCH 1/4] Add raw NV12 TCP video source example - RawNv12TcpSource reads fixed-size NV12 frames from TCP - CLI: --raw-nv12, --raw-width, --raw-height, --raw-fps - Publish via VideoSource::captureFrame (no decode) --- examples/CMakeLists.txt | 17 +- examples/publish_nv12_tcp_source/README.md | 13 + examples/publish_nv12_tcp_source/main.cpp | 235 ++++++++++++++++++ .../raw_nv12_tcp_source.cpp | 158 ++++++++++++ .../raw_nv12_tcp_source.h | 67 +++++ 5 files changed, 488 insertions(+), 2 deletions(-) create mode 100644 examples/publish_nv12_tcp_source/README.md create mode 100644 examples/publish_nv12_tcp_source/main.cpp create mode 100644 examples/publish_nv12_tcp_source/raw_nv12_tcp_source.cpp create mode 100644 examples/publish_nv12_tcp_source/raw_nv12_tcp_source.h diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 3d36664..26bd3d6 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -116,6 +116,19 @@ target_link_libraries(SimpleDataStream livekit ) +add_executable(PublishNv12TcpSource + publish_nv12_tcp_source/main.cpp + publish_nv12_tcp_source/raw_nv12_tcp_source.cpp + publish_nv12_tcp_source/raw_nv12_tcp_source.h +) + +target_include_directories(PublishNv12TcpSource PRIVATE ${EXAMPLES_PRIVATE_INCLUDE_DIRS}) + +target_link_libraries(PublishNv12TcpSource + PRIVATE + livekit +) + add_custom_command( TARGET SimpleDataStream POST_BUILD @@ -135,7 +148,7 @@ if(WIN32) ) # Copy DLLs to each example's output directory - foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream) + foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream PublishNv12TcpSource) foreach(DLL ${REQUIRED_DLLS}) add_custom_command(TARGET ${EXAMPLE} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different @@ -159,7 +172,7 @@ if(UNIX) endif() # Copy shared library to each example's output directory - foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream) + foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream PublishNv12TcpSource) add_custom_command(TARGET ${EXAMPLE} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${LIVEKIT_LIB_DIR}/${FFI_SHARED_LIB}" diff --git a/examples/publish_nv12_tcp_source/README.md b/examples/publish_nv12_tcp_source/README.md new file mode 100644 index 0000000..c0e1dcd --- /dev/null +++ b/examples/publish_nv12_tcp_source/README.md @@ -0,0 +1,13 @@ +### Generate yuv nv12 stream with gstreamer + +```bash +gst-launch-1.0 avfvideosrc device-index=0 ! videoconvert ! videorate ! videoscale ! video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! queue ! tcpserversink host=0.0.0.0 port=5004 sync=false +``` + + + +### Publish stream to LiveKit + +```bash +./build/examples/PublishNv12TcpSource --url wss://... --token --raw-nv12 0.0.0.0:5004 --raw-width 1280 --raw-height 720 --raw-fps 30 +``` \ No newline at end of file diff --git a/examples/publish_nv12_tcp_source/main.cpp b/examples/publish_nv12_tcp_source/main.cpp new file mode 100644 index 0000000..265474e --- /dev/null +++ b/examples/publish_nv12_tcp_source/main.cpp @@ -0,0 +1,235 @@ +/* + * Copyright 2025 LiveKit, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "livekit/livekit.h" +#include "raw_nv12_tcp_source.h" + +using namespace livekit; + +namespace { + +std::atomic g_running{true}; + +void printUsage(const char *prog) { + std::cerr + << "Usage: " << prog + << " --url --token --raw-nv12 [options]\n\n" + << " --url LiveKit WebSocket URL\n" + << " --token JWT token\n" + << " --enable_e2ee Enable E2EE\n" + << " --e2ee_key E2EE shared key\n\n" + << " --raw-nv12 TCP server for raw NV12 (default 127.0.0.1:5004)\n" + << " --raw-width Frame width (default: 1280)\n" + << " --raw-height Frame height (default: 720)\n" + << " --raw-fps Frame rate (default: 30)\n\n" + << "Env: LIVEKIT_URL, LIVEKIT_TOKEN, LIVEKIT_E2EE_KEY\n"; +} + +void handleSignal(int) { g_running.store(false); } + +struct RawNv12Args { + std::string host = "127.0.0.1"; + std::uint16_t port = 5004; + int width = 1280; + int height = 720; + int fps = 30; +}; + +bool parseArgs(int argc, char *argv[], std::string &url, std::string &token, + bool &enable_e2ee, std::string &e2ee_key, RawNv12Args &raw_nv12) { + enable_e2ee = false; + raw_nv12 = RawNv12Args{}; + auto get_flag_value = [&](const std::string &name, int &i) -> std::string { + std::string arg = argv[i]; + const std::string eq = name + "="; + if (arg.rfind(name, 0) == 0) { + if (arg.size() > name.size() && arg[name.size()] == '=') + return arg.substr(eq.size()); + if (i + 1 < argc) + return std::string(argv[++i]); + } + return {}; + }; + + for (int i = 1; i < argc; ++i) { + std::string a = argv[i]; + if (a == "-h" || a == "--help") return false; + if (a == "--enable_e2ee") { enable_e2ee = true; continue; } + if (a.rfind("--raw-nv12", 0) == 0) { + std::string v = get_flag_value("--raw-nv12", i); + if (v.empty()) v = "127.0.0.1:5004"; + size_t colon = v.find(':'); + if (colon != std::string::npos) { + raw_nv12.host = v.substr(0, colon); + try { + raw_nv12.port = static_cast(std::stoul(v.substr(colon + 1))); + } catch (...) { raw_nv12.port = 5004; } + } else { + raw_nv12.host = v; + } + continue; + } + if (a.rfind("--raw-width", 0) == 0) { + std::string v = get_flag_value("--raw-width", i); + if (!v.empty()) try { raw_nv12.width = std::stoi(v); } catch (...) {} + continue; + } + if (a.rfind("--raw-height", 0) == 0) { + std::string v = get_flag_value("--raw-height", i); + if (!v.empty()) try { raw_nv12.height = std::stoi(v); } catch (...) {} + continue; + } + if (a.rfind("--raw-fps", 0) == 0) { + std::string v = get_flag_value("--raw-fps", i); + if (!v.empty()) try { raw_nv12.fps = std::stoi(v); } catch (...) {} + continue; + } + if (a.rfind("--url", 0) == 0) { + std::string v = get_flag_value("--url", i); + if (!v.empty()) url = v; + continue; + } + if (a.rfind("--token", 0) == 0) { + std::string v = get_flag_value("--token", i); + if (!v.empty()) token = v; + continue; + } + if (a.rfind("--e2ee_key", 0) == 0) { + std::string v = get_flag_value("--e2ee_key", i); + if (!v.empty()) e2ee_key = v; + } + } + + if (url.empty()) { const char *e = std::getenv("LIVEKIT_URL"); if (e) url = e; } + if (token.empty()) { const char *e = std::getenv("LIVEKIT_TOKEN"); if (e) token = e; } + if (e2ee_key.empty()) { const char *e = std::getenv("LIVEKIT_E2EE_KEY"); if (e) e2ee_key = e; } + return !(url.empty() || token.empty()); +} + +class LoggingDelegate : public livekit::RoomDelegate { +public: + void onParticipantConnected(livekit::Room &, const livekit::ParticipantConnectedEvent &ev) override { + std::cout << "[Room] participant connected: " << ev.participant->identity() << "\n"; + } + void onTrackSubscribed(livekit::Room &, const livekit::TrackSubscribedEvent &ev) override { + std::cout << "[Room] track subscribed: " << (ev.publication ? ev.publication->name() : "?") << "\n"; + } +}; + +static std::vector toBytes(const std::string &s) { + return std::vector(s.begin(), s.end()); +} + +} // namespace + +int main(int argc, char *argv[]) { + std::string url, token, e2ee_key; + bool enable_e2ee = false; + RawNv12Args raw_nv12; + if (!parseArgs(argc, argv, url, token, enable_e2ee, e2ee_key, raw_nv12)) { + printUsage(argv[0]); + return 1; + } + if (url.empty() || token.empty()) { + std::cerr << "LIVEKIT_URL and LIVEKIT_TOKEN (or --url/--token) are required\n"; + return 1; + } + + std::signal(SIGINT, handleSignal); + livekit::initialize(livekit::LogSink::kConsole); + + auto room = std::make_unique(); + LoggingDelegate delegate; + room->setDelegate(&delegate); + + RoomOptions options; + options.auto_subscribe = true; + options.dynacast = false; + if (enable_e2ee) { + livekit::E2EEOptions enc; + enc.encryption_type = livekit::EncryptionType::GCM; + if (!e2ee_key.empty()) enc.key_provider_options.shared_key = toBytes(e2ee_key); + options.encryption = enc; + } + + if (!room->Connect(url, token, options)) { + std::cerr << "Failed to connect\n"; + livekit::shutdown(); + return 1; + } + std::cout << "Connected to room: " << room->room_info().name << "\n"; + + const int width = raw_nv12.width; + const int height = raw_nv12.height; + const std::size_t expected_size = static_cast(width) * static_cast(height) * 3 / 2; + + auto videoSource = std::make_shared(width, height); + auto videoTrack = LocalVideoTrack::createLocalVideoTrack("nv12_tcp", videoSource); + TrackPublishOptions videoOpts; + videoOpts.source = TrackSource::SOURCE_CAMERA; + videoOpts.dtx = false; + videoOpts.simulcast = true; + + std::shared_ptr videoPub; + try { + videoPub = room->localParticipant()->publishTrack(videoTrack, videoOpts); + std::cout << "Published video track: SID=" << videoPub->sid() << " name=" << videoPub->name() << "\n"; + } catch (const std::exception &e) { + std::cerr << "Failed to publish track: " << e.what() << "\n"; + livekit::shutdown(); + return 1; + } + + auto rawSource = std::make_unique( + raw_nv12.host, raw_nv12.port, width, height, raw_nv12.fps, + [videoSource, expected_size, width, height](publish_nv12::RawNv12Frame frame) { + if (frame.data.size() != expected_size) { + std::cerr << "Raw NV12 frame size mismatch\n"; + return; + } + try { + VideoFrame vf(width, height, VideoBufferType::NV12, std::move(frame.data)); + videoSource->captureFrame(vf, frame.timestamp_us, VideoRotation::VIDEO_ROTATION_0); + } catch (const std::exception &e) { + std::cerr << "captureFrame: " << e.what() << "\n"; + } + }); + rawSource->start(); + + while (g_running.load()) + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + + rawSource->stop(); + room->setDelegate(nullptr); + if (videoPub) + room->localParticipant()->unpublishTrack(videoPub->sid()); + room.reset(); + livekit::shutdown(); + std::cout << "Exiting.\n"; + return 0; +} diff --git a/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.cpp b/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.cpp new file mode 100644 index 0000000..761e0b8 --- /dev/null +++ b/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.cpp @@ -0,0 +1,158 @@ +/* + * Copyright 2025 LiveKit, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "raw_nv12_tcp_source.h" + +#include +#include + +#ifdef _WIN32 +#include +#include +#pragma comment(lib, "ws2_32.lib") +typedef SOCKET socket_t; +#define INVALID_SOCKET_VALUE INVALID_SOCKET +#define close_socket closesocket +#else +#include +#include +#include +#include +typedef int socket_t; +#define INVALID_SOCKET_VALUE (-1) +#define close_socket close +#endif + +namespace publish_nv12 { + +namespace { + +socket_t connectTcp(const std::string &host, std::uint16_t port) { +#ifdef _WIN32 + WSADATA wsa; + if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) return INVALID_SOCKET_VALUE; +#endif + struct addrinfo hints = {}, *res = nullptr; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + std::string portStr = std::to_string(port); + if (getaddrinfo(host.c_str(), portStr.c_str(), &hints, &res) != 0) { + std::cerr << "RawNv12TcpSource: getaddrinfo failed for " << host << ":" << port + << "\n"; + return INVALID_SOCKET_VALUE; + } + socket_t fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol); + if (fd == INVALID_SOCKET_VALUE) { + freeaddrinfo(res); + return INVALID_SOCKET_VALUE; + } + if (connect(fd, res->ai_addr, static_cast(res->ai_addrlen)) != 0) { + close_socket(fd); + freeaddrinfo(res); + return INVALID_SOCKET_VALUE; + } + freeaddrinfo(res); + return fd; +} + +} // namespace + +RawNv12TcpSource::RawNv12TcpSource(const std::string &host, + std::uint16_t port, + int width, + int height, + int fps, + RawNv12FrameCallback callback) + : host_(host), + port_(port), + width_(width), + height_(height), + fps_(fps), + callback_(std::move(callback)) { + if (width_ > 0 && height_ > 0) { + frame_size_ = static_cast(width_) * + static_cast(height_) * 3 / 2; + } +} + +RawNv12TcpSource::~RawNv12TcpSource() { stop(); } + +void RawNv12TcpSource::start() { + if (running_.exchange(true)) return; + thread_ = std::thread(&RawNv12TcpSource::loop, this); +} + +void RawNv12TcpSource::stop() { + running_.store(false); + if (thread_.joinable()) thread_.join(); +} + +void RawNv12TcpSource::loop() { + if (frame_size_ == 0) { + std::cerr << "RawNv12TcpSource: invalid frame size\n"; + running_.store(false); + return; + } + + socket_t fd = connectTcp(host_, port_); + if (fd == INVALID_SOCKET_VALUE) { + std::cerr << "RawNv12TcpSource: failed to connect to " << host_ << ":" << port_ + << "\n"; + running_.store(false); + return; + } + + std::cout << "RawNv12TcpSource: connected to " << host_ << ":" << port_ + << " (" << width_ << "x" << height_ << "@" << fps_ << "fps, frame=" + << frame_size_ << " bytes)\n"; + + auto t0 = std::chrono::steady_clock::now(); + + while (running_.load()) { + std::vector frame(frame_size_); + std::size_t filled = 0; + while (filled < frame_size_ && running_.load()) { +#ifdef _WIN32 + int n = recv(fd, reinterpret_cast(frame.data() + filled), + static_cast(frame_size_ - filled), 0); +#else + ssize_t n = recv(fd, frame.data() + filled, frame_size_ - filled, 0); +#endif + if (n <= 0) { + running_.store(false); + break; + } + filled += static_cast(n); + } + if (!running_.load() || filled < frame_size_) break; + + std::int64_t ts_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0) + .count(); + if (callback_) { + RawNv12Frame out; + out.data = std::move(frame); + out.timestamp_us = ts_us; + callback_(std::move(out)); + } + } + + close_socket(fd); + running_.store(false); +} + +} // namespace publish_nv12 diff --git a/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.h b/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.h new file mode 100644 index 0000000..8468d7f --- /dev/null +++ b/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.h @@ -0,0 +1,67 @@ +/* + * Copyright 2025 LiveKit, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace publish_nv12 { + +struct RawNv12Frame { + std::vector data; + std::int64_t timestamp_us{0}; +}; + +using RawNv12FrameCallback = std::function; + +/** + * Reads raw NV12 frames from a TCP server (fixed-size frames). + * Runs a background thread; call stop() to disconnect. + */ +class RawNv12TcpSource { +public: + RawNv12TcpSource(const std::string &host, + std::uint16_t port, + int width, + int height, + int fps, + RawNv12FrameCallback callback); + ~RawNv12TcpSource(); + + void start(); + void stop(); + bool running() const { return running_.load(); } + +private: + void loop(); + + std::string host_; + std::uint16_t port_; + int width_; + int height_; + int fps_; + std::size_t frame_size_{0}; + RawNv12FrameCallback callback_; + std::atomic running_{false}; + std::thread thread_; +}; + +} // namespace publish_nv12 From 01162f07b3abe2d0c36f0033443ee9780e009bd1 Mon Sep 17 00:00:00 2001 From: Haffi Mazhar Date: Sun, 15 Feb 2026 11:46:42 +0000 Subject: [PATCH 2/4] rename to yuv_source --- examples/CMakeLists.txt | 16 +-- .../README.md | 6 +- .../main.cpp | 113 +++++++++++++----- .../yuv_source.cpp} | 48 ++++---- .../yuv_source.h} | 26 ++-- 5 files changed, 127 insertions(+), 82 deletions(-) rename examples/{publish_nv12_tcp_source => publish_yuv_source}/README.md (67%) rename examples/{publish_nv12_tcp_source => publish_yuv_source}/main.cpp (69%) rename examples/{publish_nv12_tcp_source/raw_nv12_tcp_source.cpp => publish_yuv_source/yuv_source.cpp} (73%) rename examples/{publish_nv12_tcp_source/raw_nv12_tcp_source.h => publish_yuv_source/yuv_source.h} (73%) diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 26bd3d6..d79acb5 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -116,15 +116,15 @@ target_link_libraries(SimpleDataStream livekit ) -add_executable(PublishNv12TcpSource - publish_nv12_tcp_source/main.cpp - publish_nv12_tcp_source/raw_nv12_tcp_source.cpp - publish_nv12_tcp_source/raw_nv12_tcp_source.h +add_executable(PublishYuvSource + publish_yuv_source/main.cpp + publish_yuv_source/yuv_source.cpp + publish_yuv_source/yuv_source.h ) -target_include_directories(PublishNv12TcpSource PRIVATE ${EXAMPLES_PRIVATE_INCLUDE_DIRS}) +target_include_directories(PublishYuvSource PRIVATE ${EXAMPLES_PRIVATE_INCLUDE_DIRS}) -target_link_libraries(PublishNv12TcpSource +target_link_libraries(PublishYuvSource PRIVATE livekit ) @@ -148,7 +148,7 @@ if(WIN32) ) # Copy DLLs to each example's output directory - foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream PublishNv12TcpSource) + foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream PublishYuvSource) foreach(DLL ${REQUIRED_DLLS}) add_custom_command(TARGET ${EXAMPLE} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different @@ -172,7 +172,7 @@ if(UNIX) endif() # Copy shared library to each example's output directory - foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream PublishNv12TcpSource) + foreach(EXAMPLE SimpleRoom SimpleRpc SimpleDataStream PublishYuvSource) add_custom_command(TARGET ${EXAMPLE} POST_BUILD COMMAND ${CMAKE_COMMAND} -E copy_if_different "${LIVEKIT_LIB_DIR}/${FFI_SHARED_LIB}" diff --git a/examples/publish_nv12_tcp_source/README.md b/examples/publish_yuv_source/README.md similarity index 67% rename from examples/publish_nv12_tcp_source/README.md rename to examples/publish_yuv_source/README.md index c0e1dcd..d1d914e 100644 --- a/examples/publish_nv12_tcp_source/README.md +++ b/examples/publish_yuv_source/README.md @@ -4,10 +4,8 @@ gst-launch-1.0 avfvideosrc device-index=0 ! videoconvert ! videorate ! videoscale ! video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! queue ! tcpserversink host=0.0.0.0 port=5004 sync=false ``` - - ### Publish stream to LiveKit ```bash -./build/examples/PublishNv12TcpSource --url wss://... --token --raw-nv12 0.0.0.0:5004 --raw-width 1280 --raw-height 720 --raw-fps 30 -``` \ No newline at end of file +./build-release/bin/PublishYuvSource --url wss://... --token --raw-nv12 0.0.0.0:5004 --raw-width 1280 --raw-height 720 --raw-fps 30 +``` diff --git a/examples/publish_nv12_tcp_source/main.cpp b/examples/publish_yuv_source/main.cpp similarity index 69% rename from examples/publish_nv12_tcp_source/main.cpp rename to examples/publish_yuv_source/main.cpp index 265474e..83a17d9 100644 --- a/examples/publish_nv12_tcp_source/main.cpp +++ b/examples/publish_yuv_source/main.cpp @@ -27,7 +27,7 @@ #include #include "livekit/livekit.h" -#include "raw_nv12_tcp_source.h" +#include "yuv_source.h" using namespace livekit; @@ -43,7 +43,8 @@ void printUsage(const char *prog) { << " --token JWT token\n" << " --enable_e2ee Enable E2EE\n" << " --e2ee_key E2EE shared key\n\n" - << " --raw-nv12 TCP server for raw NV12 (default 127.0.0.1:5004)\n" + << " --raw-nv12 TCP server for raw NV12 (default " + "127.0.0.1:5004)\n" << " --raw-width Frame width (default: 1280)\n" << " --raw-height Frame height (default: 720)\n" << " --raw-fps Frame rate (default: 30)\n\n" @@ -61,7 +62,8 @@ struct RawNv12Args { }; bool parseArgs(int argc, char *argv[], std::string &url, std::string &token, - bool &enable_e2ee, std::string &e2ee_key, RawNv12Args &raw_nv12) { + bool &enable_e2ee, std::string &e2ee_key, + RawNv12Args &raw_nv12) { enable_e2ee = false; raw_nv12 = RawNv12Args{}; auto get_flag_value = [&](const std::string &name, int &i) -> std::string { @@ -78,17 +80,25 @@ bool parseArgs(int argc, char *argv[], std::string &url, std::string &token, for (int i = 1; i < argc; ++i) { std::string a = argv[i]; - if (a == "-h" || a == "--help") return false; - if (a == "--enable_e2ee") { enable_e2ee = true; continue; } + if (a == "-h" || a == "--help") + return false; + if (a == "--enable_e2ee") { + enable_e2ee = true; + continue; + } if (a.rfind("--raw-nv12", 0) == 0) { std::string v = get_flag_value("--raw-nv12", i); - if (v.empty()) v = "127.0.0.1:5004"; + if (v.empty()) + v = "127.0.0.1:5004"; size_t colon = v.find(':'); if (colon != std::string::npos) { raw_nv12.host = v.substr(0, colon); try { - raw_nv12.port = static_cast(std::stoul(v.substr(colon + 1))); - } catch (...) { raw_nv12.port = 5004; } + raw_nv12.port = + static_cast(std::stoul(v.substr(colon + 1))); + } catch (...) { + raw_nv12.port = 5004; + } } else { raw_nv12.host = v; } @@ -96,48 +106,79 @@ bool parseArgs(int argc, char *argv[], std::string &url, std::string &token, } if (a.rfind("--raw-width", 0) == 0) { std::string v = get_flag_value("--raw-width", i); - if (!v.empty()) try { raw_nv12.width = std::stoi(v); } catch (...) {} + if (!v.empty()) + try { + raw_nv12.width = std::stoi(v); + } catch (...) { + } continue; } if (a.rfind("--raw-height", 0) == 0) { std::string v = get_flag_value("--raw-height", i); - if (!v.empty()) try { raw_nv12.height = std::stoi(v); } catch (...) {} + if (!v.empty()) + try { + raw_nv12.height = std::stoi(v); + } catch (...) { + } continue; } if (a.rfind("--raw-fps", 0) == 0) { std::string v = get_flag_value("--raw-fps", i); - if (!v.empty()) try { raw_nv12.fps = std::stoi(v); } catch (...) {} + if (!v.empty()) + try { + raw_nv12.fps = std::stoi(v); + } catch (...) { + } continue; } if (a.rfind("--url", 0) == 0) { std::string v = get_flag_value("--url", i); - if (!v.empty()) url = v; + if (!v.empty()) + url = v; continue; } if (a.rfind("--token", 0) == 0) { std::string v = get_flag_value("--token", i); - if (!v.empty()) token = v; + if (!v.empty()) + token = v; continue; } if (a.rfind("--e2ee_key", 0) == 0) { std::string v = get_flag_value("--e2ee_key", i); - if (!v.empty()) e2ee_key = v; + if (!v.empty()) + e2ee_key = v; } } - if (url.empty()) { const char *e = std::getenv("LIVEKIT_URL"); if (e) url = e; } - if (token.empty()) { const char *e = std::getenv("LIVEKIT_TOKEN"); if (e) token = e; } - if (e2ee_key.empty()) { const char *e = std::getenv("LIVEKIT_E2EE_KEY"); if (e) e2ee_key = e; } + if (url.empty()) { + const char *e = std::getenv("LIVEKIT_URL"); + if (e) + url = e; + } + if (token.empty()) { + const char *e = std::getenv("LIVEKIT_TOKEN"); + if (e) + token = e; + } + if (e2ee_key.empty()) { + const char *e = std::getenv("LIVEKIT_E2EE_KEY"); + if (e) + e2ee_key = e; + } return !(url.empty() || token.empty()); } class LoggingDelegate : public livekit::RoomDelegate { public: - void onParticipantConnected(livekit::Room &, const livekit::ParticipantConnectedEvent &ev) override { - std::cout << "[Room] participant connected: " << ev.participant->identity() << "\n"; + void onParticipantConnected( + livekit::Room &, const livekit::ParticipantConnectedEvent &ev) override { + std::cout << "[Room] participant connected: " << ev.participant->identity() + << "\n"; } - void onTrackSubscribed(livekit::Room &, const livekit::TrackSubscribedEvent &ev) override { - std::cout << "[Room] track subscribed: " << (ev.publication ? ev.publication->name() : "?") << "\n"; + void onTrackSubscribed(livekit::Room &, + const livekit::TrackSubscribedEvent &ev) override { + std::cout << "[Room] track subscribed: " + << (ev.publication ? ev.publication->name() : "?") << "\n"; } }; @@ -156,7 +197,8 @@ int main(int argc, char *argv[]) { return 1; } if (url.empty() || token.empty()) { - std::cerr << "LIVEKIT_URL and LIVEKIT_TOKEN (or --url/--token) are required\n"; + std::cerr + << "LIVEKIT_URL and LIVEKIT_TOKEN (or --url/--token) are required\n"; return 1; } @@ -173,7 +215,8 @@ int main(int argc, char *argv[]) { if (enable_e2ee) { livekit::E2EEOptions enc; enc.encryption_type = livekit::EncryptionType::GCM; - if (!e2ee_key.empty()) enc.key_provider_options.shared_key = toBytes(e2ee_key); + if (!e2ee_key.empty()) + enc.key_provider_options.shared_key = toBytes(e2ee_key); options.encryption = enc; } @@ -186,45 +229,51 @@ int main(int argc, char *argv[]) { const int width = raw_nv12.width; const int height = raw_nv12.height; - const std::size_t expected_size = static_cast(width) * static_cast(height) * 3 / 2; + const std::size_t expected_size = static_cast(width) * + static_cast(height) * 3 / 2; auto videoSource = std::make_shared(width, height); - auto videoTrack = LocalVideoTrack::createLocalVideoTrack("nv12_tcp", videoSource); + auto videoTrack = + LocalVideoTrack::createLocalVideoTrack("yuv_source", videoSource); TrackPublishOptions videoOpts; videoOpts.source = TrackSource::SOURCE_CAMERA; videoOpts.dtx = false; videoOpts.simulcast = true; + videoOpts.video_codec = static_cast(1); // H264 std::shared_ptr videoPub; try { videoPub = room->localParticipant()->publishTrack(videoTrack, videoOpts); - std::cout << "Published video track: SID=" << videoPub->sid() << " name=" << videoPub->name() << "\n"; + std::cout << "Published video track: SID=" << videoPub->sid() + << " name=" << videoPub->name() << "\n"; } catch (const std::exception &e) { std::cerr << "Failed to publish track: " << e.what() << "\n"; livekit::shutdown(); return 1; } - auto rawSource = std::make_unique( + auto yuvSource = std::make_unique( raw_nv12.host, raw_nv12.port, width, height, raw_nv12.fps, - [videoSource, expected_size, width, height](publish_nv12::RawNv12Frame frame) { + [videoSource, expected_size, width, height](publish_yuv::YuvFrame frame) { if (frame.data.size() != expected_size) { std::cerr << "Raw NV12 frame size mismatch\n"; return; } try { - VideoFrame vf(width, height, VideoBufferType::NV12, std::move(frame.data)); - videoSource->captureFrame(vf, frame.timestamp_us, VideoRotation::VIDEO_ROTATION_0); + VideoFrame vf(width, height, VideoBufferType::NV12, + std::move(frame.data)); + videoSource->captureFrame(vf, frame.timestamp_us, + VideoRotation::VIDEO_ROTATION_0); } catch (const std::exception &e) { std::cerr << "captureFrame: " << e.what() << "\n"; } }); - rawSource->start(); + yuvSource->start(); while (g_running.load()) std::this_thread::sleep_for(std::chrono::milliseconds(10)); - rawSource->stop(); + yuvSource->stop(); room->setDelegate(nullptr); if (videoPub) room->localParticipant()->unpublishTrack(videoPub->sid()); diff --git a/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.cpp b/examples/publish_yuv_source/yuv_source.cpp similarity index 73% rename from examples/publish_nv12_tcp_source/raw_nv12_tcp_source.cpp rename to examples/publish_yuv_source/yuv_source.cpp index 761e0b8..f719f17 100644 --- a/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.cpp +++ b/examples/publish_yuv_source/yuv_source.cpp @@ -14,7 +14,7 @@ * limitations under the License. */ -#include "raw_nv12_tcp_source.h" +#include "yuv_source.h" #include #include @@ -36,7 +36,7 @@ typedef int socket_t; #define close_socket close #endif -namespace publish_nv12 { +namespace publish_yuv { namespace { @@ -50,8 +50,7 @@ socket_t connectTcp(const std::string &host, std::uint16_t port) { hints.ai_socktype = SOCK_STREAM; std::string portStr = std::to_string(port); if (getaddrinfo(host.c_str(), portStr.c_str(), &hints, &res) != 0) { - std::cerr << "RawNv12TcpSource: getaddrinfo failed for " << host << ":" << port - << "\n"; + std::cerr << "YuvSource: getaddrinfo failed for " << host << ":" << port << "\n"; return INVALID_SOCKET_VALUE; } socket_t fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol); @@ -70,12 +69,12 @@ socket_t connectTcp(const std::string &host, std::uint16_t port) { } // namespace -RawNv12TcpSource::RawNv12TcpSource(const std::string &host, - std::uint16_t port, - int width, - int height, - int fps, - RawNv12FrameCallback callback) +YuvSource::YuvSource(const std::string &host, + std::uint16_t port, + int width, + int height, + int fps, + YuvFrameCallback callback) : host_(host), port_(port), width_(width), @@ -83,41 +82,40 @@ RawNv12TcpSource::RawNv12TcpSource(const std::string &host, fps_(fps), callback_(std::move(callback)) { if (width_ > 0 && height_ > 0) { - frame_size_ = static_cast(width_) * - static_cast(height_) * 3 / 2; + frame_size_ = + static_cast(width_) * static_cast(height_) * 3 / 2; } } -RawNv12TcpSource::~RawNv12TcpSource() { stop(); } +YuvSource::~YuvSource() { stop(); } -void RawNv12TcpSource::start() { +void YuvSource::start() { if (running_.exchange(true)) return; - thread_ = std::thread(&RawNv12TcpSource::loop, this); + thread_ = std::thread(&YuvSource::loop, this); } -void RawNv12TcpSource::stop() { +void YuvSource::stop() { running_.store(false); if (thread_.joinable()) thread_.join(); } -void RawNv12TcpSource::loop() { +void YuvSource::loop() { if (frame_size_ == 0) { - std::cerr << "RawNv12TcpSource: invalid frame size\n"; + std::cerr << "YuvSource: invalid frame size\n"; running_.store(false); return; } socket_t fd = connectTcp(host_, port_); if (fd == INVALID_SOCKET_VALUE) { - std::cerr << "RawNv12TcpSource: failed to connect to " << host_ << ":" << port_ - << "\n"; + std::cerr << "YuvSource: failed to connect to " << host_ << ":" << port_ << "\n"; running_.store(false); return; } - std::cout << "RawNv12TcpSource: connected to " << host_ << ":" << port_ - << " (" << width_ << "x" << height_ << "@" << fps_ << "fps, frame=" - << frame_size_ << " bytes)\n"; + std::cout << "YuvSource: connected to " << host_ << ":" << port_ << " (" << width_ + << "x" << height_ << "@" << fps_ << "fps, frame=" << frame_size_ + << " bytes)\n"; auto t0 = std::chrono::steady_clock::now(); @@ -144,7 +142,7 @@ void RawNv12TcpSource::loop() { std::chrono::steady_clock::now() - t0) .count(); if (callback_) { - RawNv12Frame out; + YuvFrame out; out.data = std::move(frame); out.timestamp_us = ts_us; callback_(std::move(out)); @@ -155,4 +153,4 @@ void RawNv12TcpSource::loop() { running_.store(false); } -} // namespace publish_nv12 +} // namespace publish_yuv diff --git a/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.h b/examples/publish_yuv_source/yuv_source.h similarity index 73% rename from examples/publish_nv12_tcp_source/raw_nv12_tcp_source.h rename to examples/publish_yuv_source/yuv_source.h index 8468d7f..4196dd5 100644 --- a/examples/publish_nv12_tcp_source/raw_nv12_tcp_source.h +++ b/examples/publish_yuv_source/yuv_source.h @@ -23,28 +23,28 @@ #include #include -namespace publish_nv12 { +namespace publish_yuv { -struct RawNv12Frame { +struct YuvFrame { std::vector data; std::int64_t timestamp_us{0}; }; -using RawNv12FrameCallback = std::function; +using YuvFrameCallback = std::function; /** * Reads raw NV12 frames from a TCP server (fixed-size frames). * Runs a background thread; call stop() to disconnect. */ -class RawNv12TcpSource { +class YuvSource { public: - RawNv12TcpSource(const std::string &host, - std::uint16_t port, - int width, - int height, - int fps, - RawNv12FrameCallback callback); - ~RawNv12TcpSource(); + YuvSource(const std::string &host, + std::uint16_t port, + int width, + int height, + int fps, + YuvFrameCallback callback); + ~YuvSource(); void start(); void stop(); @@ -59,9 +59,9 @@ class RawNv12TcpSource { int height_; int fps_; std::size_t frame_size_{0}; - RawNv12FrameCallback callback_; + YuvFrameCallback callback_; std::atomic running_{false}; std::thread thread_; }; -} // namespace publish_nv12 +} // namespace publish_yuv From e0ea498ba2428ad750672d3374853133659d7c7a Mon Sep 17 00:00:00 2001 From: Haffi Mazhar Date: Sun, 15 Feb 2026 11:48:22 +0000 Subject: [PATCH 3/4] clang format --- examples/publish_yuv_source/yuv_source.cpp | 49 ++++++++++------------ examples/publish_yuv_source/yuv_source.h | 8 +--- 2 files changed, 25 insertions(+), 32 deletions(-) diff --git a/examples/publish_yuv_source/yuv_source.cpp b/examples/publish_yuv_source/yuv_source.cpp index f719f17..763c872 100644 --- a/examples/publish_yuv_source/yuv_source.cpp +++ b/examples/publish_yuv_source/yuv_source.cpp @@ -43,14 +43,16 @@ namespace { socket_t connectTcp(const std::string &host, std::uint16_t port) { #ifdef _WIN32 WSADATA wsa; - if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) return INVALID_SOCKET_VALUE; + if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) + return INVALID_SOCKET_VALUE; #endif struct addrinfo hints = {}, *res = nullptr; hints.ai_family = AF_UNSPEC; hints.ai_socktype = SOCK_STREAM; std::string portStr = std::to_string(port); if (getaddrinfo(host.c_str(), portStr.c_str(), &hints, &res) != 0) { - std::cerr << "YuvSource: getaddrinfo failed for " << host << ":" << port << "\n"; + std::cerr << "YuvSource: getaddrinfo failed for " << host << ":" << port + << "\n"; return INVALID_SOCKET_VALUE; } socket_t fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol); @@ -69,34 +71,28 @@ socket_t connectTcp(const std::string &host, std::uint16_t port) { } // namespace -YuvSource::YuvSource(const std::string &host, - std::uint16_t port, - int width, - int height, - int fps, - YuvFrameCallback callback) - : host_(host), - port_(port), - width_(width), - height_(height), - fps_(fps), +YuvSource::YuvSource(const std::string &host, std::uint16_t port, int width, + int height, int fps, YuvFrameCallback callback) + : host_(host), port_(port), width_(width), height_(height), fps_(fps), callback_(std::move(callback)) { if (width_ > 0 && height_ > 0) { - frame_size_ = - static_cast(width_) * static_cast(height_) * 3 / 2; + frame_size_ = static_cast(width_) * + static_cast(height_) * 3 / 2; } } YuvSource::~YuvSource() { stop(); } void YuvSource::start() { - if (running_.exchange(true)) return; + if (running_.exchange(true)) + return; thread_ = std::thread(&YuvSource::loop, this); } void YuvSource::stop() { running_.store(false); - if (thread_.joinable()) thread_.join(); + if (thread_.joinable()) + thread_.join(); } void YuvSource::loop() { @@ -108,14 +104,15 @@ void YuvSource::loop() { socket_t fd = connectTcp(host_, port_); if (fd == INVALID_SOCKET_VALUE) { - std::cerr << "YuvSource: failed to connect to " << host_ << ":" << port_ << "\n"; + std::cerr << "YuvSource: failed to connect to " << host_ << ":" << port_ + << "\n"; running_.store(false); return; } - std::cout << "YuvSource: connected to " << host_ << ":" << port_ << " (" << width_ - << "x" << height_ << "@" << fps_ << "fps, frame=" << frame_size_ - << " bytes)\n"; + std::cout << "YuvSource: connected to " << host_ << ":" << port_ << " (" + << width_ << "x" << height_ << "@" << fps_ + << "fps, frame=" << frame_size_ << " bytes)\n"; auto t0 = std::chrono::steady_clock::now(); @@ -135,12 +132,12 @@ void YuvSource::loop() { } filled += static_cast(n); } - if (!running_.load() || filled < frame_size_) break; + if (!running_.load() || filled < frame_size_) + break; - std::int64_t ts_us = - std::chrono::duration_cast( - std::chrono::steady_clock::now() - t0) - .count(); + std::int64_t ts_us = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0) + .count(); if (callback_) { YuvFrame out; out.data = std::move(frame); diff --git a/examples/publish_yuv_source/yuv_source.h b/examples/publish_yuv_source/yuv_source.h index 4196dd5..62adc09 100644 --- a/examples/publish_yuv_source/yuv_source.h +++ b/examples/publish_yuv_source/yuv_source.h @@ -38,12 +38,8 @@ using YuvFrameCallback = std::function; */ class YuvSource { public: - YuvSource(const std::string &host, - std::uint16_t port, - int width, - int height, - int fps, - YuvFrameCallback callback); + YuvSource(const std::string &host, std::uint16_t port, int width, int height, + int fps, YuvFrameCallback callback); ~YuvSource(); void start(); From 770f4a82add666f9f8c615c8fbd9f5f3f0901b67 Mon Sep 17 00:00:00 2001 From: Haffi Mazhar Date: Sun, 15 Feb 2026 11:53:55 +0000 Subject: [PATCH 4/4] rename cli arg --- examples/publish_yuv_source/README.md | 2 +- examples/publish_yuv_source/main.cpp | 29 +++++++++++++-------------- 2 files changed, 15 insertions(+), 16 deletions(-) diff --git a/examples/publish_yuv_source/README.md b/examples/publish_yuv_source/README.md index d1d914e..43d90f6 100644 --- a/examples/publish_yuv_source/README.md +++ b/examples/publish_yuv_source/README.md @@ -7,5 +7,5 @@ gst-launch-1.0 avfvideosrc device-index=0 ! videoconvert ! videorate ! videoscal ### Publish stream to LiveKit ```bash -./build-release/bin/PublishYuvSource --url wss://... --token --raw-nv12 0.0.0.0:5004 --raw-width 1280 --raw-height 720 --raw-fps 30 +./build-release/bin/PublishYuvSource --url wss://... --token --tcp 0.0.0.0:5004 --raw-width 1280 --raw-height 720 --raw-fps 30 ``` diff --git a/examples/publish_yuv_source/main.cpp b/examples/publish_yuv_source/main.cpp index 83a17d9..e36d446 100644 --- a/examples/publish_yuv_source/main.cpp +++ b/examples/publish_yuv_source/main.cpp @@ -36,19 +36,18 @@ namespace { std::atomic g_running{true}; void printUsage(const char *prog) { - std::cerr - << "Usage: " << prog - << " --url --token --raw-nv12 [options]\n\n" - << " --url LiveKit WebSocket URL\n" - << " --token JWT token\n" - << " --enable_e2ee Enable E2EE\n" - << " --e2ee_key E2EE shared key\n\n" - << " --raw-nv12 TCP server for raw NV12 (default " - "127.0.0.1:5004)\n" - << " --raw-width Frame width (default: 1280)\n" - << " --raw-height Frame height (default: 720)\n" - << " --raw-fps Frame rate (default: 30)\n\n" - << "Env: LIVEKIT_URL, LIVEKIT_TOKEN, LIVEKIT_E2EE_KEY\n"; + std::cerr << "Usage: " << prog + << " --url --token --tcp [options]\n\n" + << " --url LiveKit WebSocket URL\n" + << " --token JWT token\n" + << " --enable_e2ee Enable E2EE\n" + << " --e2ee_key E2EE shared key\n\n" + << " --tcp TCP server for raw NV12 (default " + "127.0.0.1:5004)\n" + << " --raw-width Frame width (default: 1280)\n" + << " --raw-height Frame height (default: 720)\n" + << " --raw-fps Frame rate (default: 30)\n\n" + << "Env: LIVEKIT_URL, LIVEKIT_TOKEN, LIVEKIT_E2EE_KEY\n"; } void handleSignal(int) { g_running.store(false); } @@ -86,8 +85,8 @@ bool parseArgs(int argc, char *argv[], std::string &url, std::string &token, enable_e2ee = true; continue; } - if (a.rfind("--raw-nv12", 0) == 0) { - std::string v = get_flag_value("--raw-nv12", i); + if (a.rfind("--tcp", 0) == 0) { + std::string v = get_flag_value("--tcp", i); if (v.empty()) v = "127.0.0.1:5004"; size_t colon = v.find(':');