diff --git a/CMakeLists.txt b/CMakeLists.txt index 1c7b244c..2b2d768d 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -18,6 +18,7 @@ include(warnings) option(LIVEKIT_BUILD_EXAMPLES "Build LiveKit examples" OFF) option(LIVEKIT_BUILD_TESTS "Build LiveKit tests" OFF) +option(LIVEKIT_ENABLE_CAPTURE "Build the Rust FFI with the capture feature (livekit-capture sources; links system GStreamer)" OFF) # vcpkg is only used on Windows; Linux/macOS use system package managers if(WIN32) @@ -85,6 +86,7 @@ set(FFI_PROTO_FILES ${FFI_PROTO_DIR}/data_track.proto ${FFI_PROTO_DIR}/rpc.proto ${FFI_PROTO_DIR}/track_publication.proto + ${FFI_PROTO_DIR}/capture.proto ) set(PROTO_BINARY_DIR ${LIVEKIT_BINARY_DIR}/generated) file(MAKE_DIRECTORY ${PROTO_BINARY_DIR}) @@ -166,6 +168,12 @@ set(GENERATED_BUILD_H "${LIVEKIT_ROOT_DIR}/include/livekit/build.h") set(GENERATED_BUILD_INFO_JSON "${LIVEKIT_BINARY_DIR}/build-info.json") set(RUST_ROOT ${LIVEKIT_ROOT_DIR}/client-sdk-rust) +# Cargo features for the Rust FFI build. +set(LIVEKIT_CARGO_FEATURES "") +if(LIVEKIT_ENABLE_CAPTURE) + set(LIVEKIT_CARGO_FEATURES "capture") +endif() + set(GIT_COMMIT "unknown") set(RUST_GIT_COMMIT "unknown") find_package(Git QUIET) @@ -261,6 +269,11 @@ if(NOT CFG STREQUAL \"Debug\") list(APPEND ARGS --release) endif() +if(DEFINED CARGO_FEATURES AND NOT CARGO_FEATURES STREQUAL \"\") + list(APPEND ARGS --features \"\${CARGO_FEATURES}\") + message(STATUS \"[run_cargo.cmake] Cargo features: \${CARGO_FEATURES}\") +endif() + if(DEFINED RUST_TARGET AND NOT RUST_TARGET STREQUAL \"\") list(APPEND ARGS --target \"\${RUST_TARGET}\") message(STATUS \"[run_cargo.cmake] Cross-compiling for target: \${RUST_TARGET}\") @@ -357,6 +370,7 @@ add_custom_command( -DPROTOC_PATH=${Protobuf_PROTOC_EXECUTABLE} -DRUST_TARGET=${RUST_TARGET_TRIPLE} -DGCC_LIB_DIR=${GCC_LIB_DIR} + -DCARGO_FEATURES=${LIVEKIT_CARGO_FEATURES} -P "${RUN_CARGO_SCRIPT}" WORKING_DIRECTORY "${RUST_ROOT}" DEPENDS ${RUST_SOURCES} @@ -373,6 +387,7 @@ add_custom_target(build_rust_ffi add_library(livekit SHARED src/audio_frame.cpp src/audio_processing_module.cpp + src/capture_source.cpp src/audio_source.cpp src/audio_stream.cpp src/data_track_frame.cpp diff --git a/client-sdk-rust b/client-sdk-rust index 06371a33..c6142dcc 160000 --- a/client-sdk-rust +++ b/client-sdk-rust @@ -1 +1 @@ -Subproject commit 06371a336d485cf6b51c75a3fee3d208f8f2eedb +Subproject commit c6142dcc872e6554c0c534c3bab9bc87f76a076e diff --git a/include/livekit/capture_source.h b/include/livekit/capture_source.h new file mode 100644 index 00000000..36aafcf1 --- /dev/null +++ b/include/livekit/capture_source.h @@ -0,0 +1,234 @@ +/* + * Copyright 2026 LiveKit + * + * 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 +#include +#include + +#include "livekit/ffi_handle.h" +#include "livekit/room_event_types.h" +#include "livekit/video_source.h" +#include "livekit/visibility.h" + +namespace livekit { + +namespace proto { +class NewCaptureSourceRequest; +class OwnedCaptureSource; +} // namespace proto + +/// Error raised when capture source creation or control fails. +class LIVEKIT_API CaptureSourceError : public std::runtime_error { +public: + /// Create a capture source error. + /// + /// @param message Human-readable error message. + explicit CaptureSourceError(const std::string& message) : std::runtime_error(message) {} +}; + +/// Kind of media a capture source produces. +enum class CaptureSourceKind { + /// Pixel frames, published through the WebRTC encoder. + Pixel = 0, + /// Pre-encoded access units, published as passthrough. + Encoded = 1, +}; + +/// Bitrate unit expected by a GStreamer encoder property. +enum class GstreamerBitrateUnit { + /// Bits per second. + Bps = 0, + /// Kilobits per second. + Kbps = 1, +}; + +/// Video resolution in pixels. +struct CaptureResolution { + int width = 0; + int height = 0; +}; + +/// Binding from WebRTC rate-control targets to a GStreamer encoder property. +struct GstreamerRateControl { + /// Name of the encoder element in the pipeline (e.g. `lk_encoder`). + std::string element; + + /// Bitrate property to set on the element (e.g. `bitrate` for x264enc, + /// `target-bitrate` for vp8enc/vp9enc). + std::string property; + + /// Unit the property expects. + GstreamerBitrateUnit unit = GstreamerBitrateUnit::Bps; +}; + +/// Configuration for encoded ingest from a GStreamer pipeline. +struct GstreamerVideoSourceConfig { + /// GStreamer launch description for the encoded producer pipeline. + /// + /// Must contain `appsink name=lk_appsink`, or leave exactly one encoded + /// video source pad unlinked for the source to attach one to. + std::string pipeline; + + /// Codec expected from the pipeline; inferred from pipeline caps when + /// omitted. + std::optional codec; + + /// Encoded frame resolution. When omitted, it is discovered from the + /// pipeline's negotiated caps; when set, the pipeline output is verified + /// against it. + std::optional resolution; + + /// Forwards WebRTC rate-control targets to an encoder element's bitrate + /// property. Without this, the pipeline encodes at a fixed bitrate. + std::optional rate_control; +}; + +/// Configuration for the built-in test source producing solid-color frames. +struct DemoVideoSourceConfig {}; + +/// Why a capture ended without error. +enum class CaptureExit { + /// Stopped via @ref CaptureSource::stop (or source destruction). + Stopped = 0, + /// The producer reached the end of its stream. + EndOfStream = 1, +}; + +/// Terminal result of a started capture. +struct CaptureResult { + /// Error message when the capture failed; empty on success. + std::optional error; + + /// Number of frames or access units captured. + std::uint64_t frames_captured = 0; + + /// Why the capture ended; meaningful only when @ref error is empty. + CaptureExit exit = CaptureExit::Stopped; +}; + +/// A server-side capture source (livekit-capture) that produces video +/// without per-frame FFI traffic. +/// +/// The source owns its producer (e.g. a GStreamer pipeline) and the pump +/// that feeds an RTC video source. Publish it like any other source: create +/// a track from @ref videoSource(), publish with application options merged +/// over @ref recommendedPublishOptions(), then call @ref start(). +/// +/// Requires the FFI library to be built with the `capture` feature +/// (`LIVEKIT_ENABLE_CAPTURE`); otherwise creation fails. +/// +/// @note Keep this object alive while the track is published. Destroying it +/// stops a running capture. +class LIVEKIT_API CaptureSource { +public: + /// Terminal notification for a started capture, invoked exactly once on + /// the FFI event thread (like room delegate callbacks). + using FinishedCallback = std::function; + + /// Create a capture source from a GStreamer pipeline configuration. + /// + /// Completes asynchronously: construction starts the pipeline and may wait + /// for its first output to discover stream settings. Errors (invalid + /// pipeline, missing capture feature, discovery timeout) are thrown from + /// the future as @ref CaptureSourceError. + static std::future> + create(GstreamerVideoSourceConfig config); + + /// Create the built-in demo capture source (solid cycling colors). + static std::future> + create(DemoVideoSourceConfig config); + + ~CaptureSource(); + + CaptureSource(const CaptureSource &) = delete; + CaptureSource &operator=(const CaptureSource &) = delete; + CaptureSource(CaptureSource &&) = delete; + CaptureSource &operator=(CaptureSource &&) = delete; + + /// Kind of media this source produces. + CaptureSourceKind kind() const noexcept { return kind_; } + + /// Declared or discovered stream resolution. + int width() const noexcept { return width_; } + int height() const noexcept { return height_; } + + /// Codec produced by the source; encoded sources only. + std::optional codec() const noexcept { return codec_; } + + /// RTC video source fed by this capture source; use it with + /// LocalVideoTrack::createLocalVideoTrack(). + std::shared_ptr videoSource() const noexcept { + return video_source_; + } + + /// Returns publish options for this track, applying application options. + /// + /// Fields the source dictates (e.g. codec, encoder backend, and simulcast + /// for encoded sources) are required for correct publication and override + /// the application values; all other fields are taken from @p options + /// unchanged. + TrackPublishOptions publishOptions(TrackPublishOptions options = {}) const; + + /// Set the terminal notification. Set this before @ref start(). + void setOnFinishedCallback(FinishedCallback callback); + + /// Start pumping frames into the RTC video source. + /// + /// @throws CaptureSourceError if the capture was already started or the + /// FFI call fails. + void start(); + + /// Signal a running capture to stop after the frame in flight. The + /// finished callback fires shortly after. Stopping an already-finished + /// capture is a no-op. + /// + /// @throws CaptureSourceError if the FFI call fails. + void stop(); + +private: + CaptureSource() = default; + + /// Shared creation path: sends the request and maps the callback payload + /// onto a wrapper instance. + static std::future> + createFromRequest(proto::NewCaptureSourceRequest request); + + /// Builds the wrapper from the callback payload, adopting its handles. + static std::shared_ptr + fromOwned(const proto::OwnedCaptureSource &owned); + + FfiHandle handle_; + CaptureSourceKind kind_ = CaptureSourceKind::Pixel; + int width_ = 0; + int height_ = 0; + std::optional codec_; + std::shared_ptr video_source_; + /// Fields set here are dictated by the source and win in publishOptions(). + TrackPublishOptions source_publish_options_; + + std::mutex callback_mutex_; + FinishedCallback on_finished_; + int listener_id_ = 0; +}; + +} // namespace livekit diff --git a/include/livekit/livekit.h b/include/livekit/livekit.h index d0aaee89..8859aef1 100644 --- a/include/livekit/livekit.h +++ b/include/livekit/livekit.h @@ -21,6 +21,7 @@ #include "livekit/audio_source.h" #include "livekit/audio_stream.h" #include "livekit/build.h" +#include "livekit/capture_source.h" #include "livekit/e2ee.h" #include "livekit/local_audio_track.h" #include "livekit/local_participant.h" diff --git a/include/livekit/room_event_types.h b/include/livekit/room_event_types.h index 86589681..edc1851d 100644 --- a/include/livekit/room_event_types.h +++ b/include/livekit/room_event_types.h @@ -34,7 +34,37 @@ class LocalTrackPublication; class RemoteTrackPublication; class TrackPublication; -enum class VideoCodec; +/// Video codec used when publishing a track. +/// +/// Values mirror proto_room.VideoCodec. +enum class VideoCodec { + VP8 = 0, + H264 = 1, + AV1 = 2, + VP9 = 3, + H265 = 4, +}; + +/// Preferred encoder backend when publishing a video track. +/// +/// Values mirror proto_room.VideoEncoderBackend. +enum class VideoEncoderBackend { + /// Pick the best available backend. + Auto = 0, + /// Software encoder. + Software = 1, + /// Any hardware encoder. + Hardware = 2, + /// NVIDIA NVENC hardware encoder. + Nvenc = 3, + /// VA-API hardware encoder. + Vaapi = 4, + /// Apple VideoToolbox hardware encoder. + VideoToolbox = 5, + /// Pre-encoded passthrough: the application supplies encoded frames. + PreEncoded = 6, +}; + enum class TrackSource; /// Overall quality of a participant's connection. @@ -359,6 +389,9 @@ struct TrackPublishOptions { /// Controls how the encoder trades off between resolution and framerate /// when bandwidth is constrained. If not set, the server defaults apply. std::optional degradation_preference; + + /// Preferred encoder backend for published video. + std::optional video_encoder; }; // --------------------------------------------------------- diff --git a/include/livekit/video_source.h b/include/livekit/video_source.h index 44e69d8f..402b71f4 100644 --- a/include/livekit/video_source.h +++ b/include/livekit/video_source.h @@ -95,6 +95,12 @@ class LIVEKIT_API VideoSource { VideoRotation rotation = VideoRotation::VIDEO_ROTATION_0); private: + friend class CaptureSource; + + /// Adopt an existing FFI video source handle (used by CaptureSource). + VideoSource(FfiHandle&& handle, int width, int height) noexcept + : handle_(std::move(handle)), width_(width), height_(height) {} + FfiHandle handle_; // owned FFI handle int width_{0}; int height_{0}; diff --git a/src/capture_source.cpp b/src/capture_source.cpp new file mode 100644 index 00000000..7444c5fc --- /dev/null +++ b/src/capture_source.cpp @@ -0,0 +1,210 @@ +/* + * Copyright 2026 LiveKit + * + * 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 "livekit/capture_source.h" + +#include + +#include "capture.pb.h" +#include "ffi.pb.h" +#include "ffi_client.h" +#include "room_proto_converter.h" + +namespace livekit { + +namespace { + +proto::GstreamerBitrateUnit toProto(GstreamerBitrateUnit unit) { + switch (unit) { + case GstreamerBitrateUnit::Bps: + return proto::GstreamerBitrateUnit::GSTREAMER_BITRATE_UNIT_BPS; + case GstreamerBitrateUnit::Kbps: + return proto::GstreamerBitrateUnit::GSTREAMER_BITRATE_UNIT_KBPS; + } + return proto::GstreamerBitrateUnit::GSTREAMER_BITRATE_UNIT_BPS; +} + +CaptureResult resultFromEvent(const proto::CaptureSourceEvent &event) { + CaptureResult result; + if (event.has_error()) { + result.error = event.error().error(); + return result; + } + if (event.has_finished()) { + result.frames_captured = event.finished().frames_captured(); + result.exit = event.finished().exit() == + proto::CaptureExit::CAPTURE_EXIT_END_OF_STREAM + ? CaptureExit::EndOfStream + : CaptureExit::Stopped; + } + return result; +} + +} // namespace + +std::future> +CaptureSource::create(GstreamerVideoSourceConfig config) { + proto::NewCaptureSourceRequest request; + auto *gstreamer = request.mutable_gstreamer(); + gstreamer->set_pipeline(config.pipeline); + if (config.codec) { + gstreamer->set_codec(static_cast(*config.codec)); + } + if (config.resolution) { + gstreamer->mutable_resolution()->set_width(config.resolution->width); + gstreamer->mutable_resolution()->set_height(config.resolution->height); + } + if (config.rate_control) { + auto *rate_control = gstreamer->mutable_rate_control(); + rate_control->set_element(config.rate_control->element); + rate_control->set_property(config.rate_control->property); + rate_control->set_unit(toProto(config.rate_control->unit)); + } + return createFromRequest(std::move(request)); +} + +std::future> +CaptureSource::create(DemoVideoSourceConfig /*config*/) { + proto::NewCaptureSourceRequest request; + request.mutable_demo(); + return createFromRequest(std::move(request)); +} + +std::future> +CaptureSource::createFromRequest(proto::NewCaptureSourceRequest request) { + auto owned = FfiClient::instance().newCaptureSourceAsync(std::move(request)); + // Map the FFI payload onto a wrapper once the callback resolves. A helper + // thread keeps the returned future's wait semantics standard. + return std::async(std::launch::async, [owned = std::move(owned)]() mutable { + try { + return fromOwned(owned.get()); + } catch (const CaptureSourceError&) { + throw; + } catch (const std::exception& e) { + throw CaptureSourceError(e.what()); + } + }); +} + +std::shared_ptr +CaptureSource::fromOwned(const proto::OwnedCaptureSource &owned) { + const proto::CaptureSourceInfo &info = owned.info(); + + std::shared_ptr source(new CaptureSource()); + source->handle_ = FfiHandle(static_cast(owned.handle().id())); + source->kind_ = + info.kind() == proto::CaptureSourceKind::CAPTURE_SOURCE_ENCODED + ? CaptureSourceKind::Encoded + : CaptureSourceKind::Pixel; + source->width_ = static_cast(info.resolution().width()); + source->height_ = static_cast(info.resolution().height()); + if (info.has_codec()) { + source->codec_ = static_cast(info.codec()); + } + source->source_publish_options_ = fromProto(info.recommended_publish_options()); + source->video_source_ = std::shared_ptr(new VideoSource( + FfiHandle(static_cast(info.video_source().handle().id())), + source->width_, source->height_)); + + // The terminal CaptureSourceEvent is unsolicited (not async-id + // correlated); observe it with a listener filtered by our handle. The + // destructor removes the listener before releasing the handle, so no + // callback can outlive the wrapper. + CaptureSource *raw = source.get(); + const std::uint64_t capture_handle = source->handle_.get(); + source->listener_id_ = FfiClient::instance().addListener( + [raw, capture_handle](const proto::FfiEvent &event) { + if (!event.has_capture_source_event() || + event.capture_source_event().capture_handle() != capture_handle) { + return; + } + const CaptureResult result = + resultFromEvent(event.capture_source_event()); + FinishedCallback callback; + { + const std::scoped_lock lock(raw->callback_mutex_); + callback = raw->on_finished_; + } + if (callback) { + callback(result); + } + }); + + return source; +} + +TrackPublishOptions CaptureSource::publishOptions(TrackPublishOptions options) const { + const TrackPublishOptions &dictated = source_publish_options_; + const auto overlay = [](auto &target, const auto &source) { + if (source.has_value()) { + target = source; + } + }; + overlay(options.video_encoding, dictated.video_encoding); + overlay(options.audio_encoding, dictated.audio_encoding); + overlay(options.video_codec, dictated.video_codec); + overlay(options.dtx, dictated.dtx); + overlay(options.red, dictated.red); + overlay(options.simulcast, dictated.simulcast); + overlay(options.source, dictated.source); + overlay(options.stream, dictated.stream); + overlay(options.preconnect_buffer, dictated.preconnect_buffer); + overlay(options.frame_metadata_features, dictated.frame_metadata_features); + overlay(options.degradation_preference, dictated.degradation_preference); + overlay(options.video_encoder, dictated.video_encoder); + return options; +} + +CaptureSource::~CaptureSource() { + if (listener_id_ != 0) { + FfiClient::instance().removeListener(listener_id_); + listener_id_ = 0; + } + // Dropping the handle stops a running capture. +} + +void CaptureSource::setOnFinishedCallback(FinishedCallback callback) { + const std::scoped_lock lock(callback_mutex_); + on_finished_ = std::move(callback); +} + +void CaptureSource::start() { + proto::FfiRequest req; + req.mutable_start_capture()->set_capture_handle(handle_.get()); + + const proto::FfiResponse resp = FfiClient::instance().sendRequest(req); + if (!resp.has_start_capture()) { + throw CaptureSourceError("FfiResponse missing start_capture"); + } + if (resp.start_capture().has_error()) { + throw CaptureSourceError(resp.start_capture().error()); + } +} + +void CaptureSource::stop() { + proto::FfiRequest req; + req.mutable_stop_capture()->set_capture_handle(handle_.get()); + + const proto::FfiResponse resp = FfiClient::instance().sendRequest(req); + if (!resp.has_stop_capture()) { + throw CaptureSourceError("FfiResponse missing stop_capture"); + } + if (resp.stop_capture().has_error()) { + throw CaptureSourceError(resp.stop_capture().error()); + } +} + +} // namespace livekit diff --git a/src/ffi_client.cpp b/src/ffi_client.cpp index 8c0516b0..6bb6f5d6 100644 --- a/src/ffi_client.cpp +++ b/src/ffi_client.cpp @@ -107,6 +107,8 @@ std::optional ExtractAsyncId(const proto::FfiEvent& event) { return event.chat_message().async_id(); case E::kPerformRpc: return event.perform_rpc().async_id(); + case E::kNewCaptureSource: + return event.new_capture_source().async_id(); // low-level data stream callbacks case E::kSendStreamHeader: @@ -708,6 +710,49 @@ std::future FfiClient::publishTrackAsync(std::uint return fut; } +std::future FfiClient::newCaptureSourceAsync(proto::NewCaptureSourceRequest request) { + // Generate client-side async_id first + const AsyncId async_id = generateAsyncId(); + + // Register the async handler BEFORE sending the request + auto fut = registerAsync( + async_id, + [async_id](const proto::FfiEvent& event) { + return event.has_new_capture_source() && event.new_capture_source().async_id() == async_id; + }, + [](const proto::FfiEvent& event, std::promise& pr) { + const auto& cb = event.new_capture_source(); + + // Oneof message { string error = 2; OwnedCaptureSource source = 3; } + if (cb.has_error() && !cb.error().empty()) { + pr.set_exception(std::make_exception_ptr(std::runtime_error(cb.error()))); + return; + } + if (!cb.has_source()) { + pr.set_exception(std::make_exception_ptr(std::runtime_error("NewCaptureSourceCallback missing source"))); + return; + } + + pr.set_value(cb.source()); + }); + + request.set_request_async_id(async_id); + proto::FfiRequest req; + req.mutable_new_capture_source()->CopyFrom(request); + + try { + const proto::FfiResponse resp = sendRequest(req); + if (!resp.has_new_capture_source()) { + logAndThrow("FfiResponse missing new_capture_source"); + } + } catch (...) { + cancelPendingByAsyncId(async_id); + throw; + } + + return fut; +} + std::future FfiClient::unpublishTrackAsync(std::uint64_t local_participant_handle, const std::string& track_sid, bool stop_on_unpublish) { // Generate client-side async_id first diff --git a/src/ffi_client.h b/src/ffi_client.h index c0835027..8ea097bc 100644 --- a/src/ffi_client.h +++ b/src/ffi_client.h @@ -29,6 +29,7 @@ #include #include +#include "capture.pb.h" #include "data_track.pb.h" #include "livekit/data_track_error.h" #include "livekit/data_track_options.h" @@ -118,6 +119,9 @@ class LIVEKIT_INTERNAL_API FfiClient { std::future publishSipDtmfAsync(std::uint64_t local_participant_handle, std::uint32_t code, const std::string& digit, const std::vector& destination_identities); + + // Capture APIs + std::future newCaptureSourceAsync(proto::NewCaptureSourceRequest request); std::future setLocalMetadataAsync(std::uint64_t local_participant_handle, const std::string& metadata); std::future captureAudioFrameAsync(std::uint64_t source_handle, const proto::AudioFrameBufferInfo& buffer); std::future performRpcAsync(std::uint64_t local_participant_handle, diff --git a/src/room_proto_converter.cpp b/src/room_proto_converter.cpp index a1265aa3..2a1d682e 100644 --- a/src/room_proto_converter.cpp +++ b/src/room_proto_converter.cpp @@ -533,6 +533,9 @@ proto::TrackPublishOptions toProto(const TrackPublishOptions& in) { if (in.degradation_preference) { msg.set_degradation_preference(static_cast(*in.degradation_preference)); } + if (in.video_encoder) { + msg.set_video_encoder(static_cast(*in.video_encoder)); + } return msg; } @@ -573,6 +576,9 @@ TrackPublishOptions fromProto(const proto::TrackPublishOptions& in) { if (in.has_degradation_preference()) { out.degradation_preference = static_cast(in.degradation_preference()); } + if (in.has_video_encoder()) { + out.video_encoder = static_cast(in.video_encoder()); + } return out; } diff --git a/src/tests/integration/test_capture_source.cpp b/src/tests/integration/test_capture_source.cpp new file mode 100644 index 00000000..e961e221 --- /dev/null +++ b/src/tests/integration/test_capture_source.cpp @@ -0,0 +1,161 @@ +/* + * Copyright 2026 LiveKit + * + * 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. + */ + +/// End-to-end test for capture sources (livekit-capture over FFI). +/// +/// Publishes the built-in demo source — the capture pump pushes frames +/// server-side, so the test drives no frames itself — and verifies that a +/// second participant receives real video through the SFU. +/// +/// Requires the Rust FFI built with the `capture` feature +/// (-DLIVEKIT_ENABLE_CAPTURE=ON); otherwise the test skips. + +#include +#include +#include +#include +#include + +#include "../common/test_common.h" +#include "livekit/capture_source.h" + +namespace livekit::test { + +using namespace std::chrono_literals; + +namespace { + +constexpr int kMinFramesReceived = 5; + +/// Creates the capture source, or skips the test when the FFI library was +/// built without the capture feature. +std::shared_ptr createDemoCaptureOrSkip() { + try { + return CaptureSource::create(DemoVideoSourceConfig{}).get(); + } catch (const std::exception &e) { + if (std::string(e.what()).find("without the 'capture' feature") != + std::string::npos) { + return nullptr; + } + throw; + } +} + +} // namespace + +class CaptureSourceServerTest : public LiveKitTestBase {}; + +TEST_F(CaptureSourceServerTest, DemoCaptureSourcePublishesFramesEndToEnd) { + failIfNotConfigured(); + + auto capture = createDemoCaptureOrSkip(); + if (capture == nullptr) { + GTEST_SKIP() << "livekit-ffi built without the 'capture' feature; " + "configure with -DLIVEKIT_ENABLE_CAPTURE=ON"; + } + + // The demo source produces 1280x720 pixel frames. + ASSERT_EQ(capture->kind(), CaptureSourceKind::Pixel); + ASSERT_EQ(capture->width(), 1280); + ASSERT_EQ(capture->height(), 720); + ASSERT_FALSE(capture->codec().has_value()); + ASSERT_NE(capture->videoSource(), nullptr); + + Room sender_room; + Room receiver_room; + RoomOptions options; + + ASSERT_TRUE(receiver_room.connect(config_.url, config_.token_b, options)); + ASSERT_TRUE(sender_room.connect(config_.url, config_.token_a, options)); + ASSERT_FALSE(sender_room.localParticipant().expired()); + ASSERT_FALSE(receiver_room.localParticipant().expired()); + + const std::string sender_identity = + lockLocalParticipant(sender_room)->identity(); + ASSERT_FALSE(sender_identity.empty()); + ASSERT_TRUE(waitForParticipant(&receiver_room, sender_identity, 10s)); + + // Count frames arriving at the receiver: proof of media flowing through + // the SFU without the test pushing a single frame. + std::mutex mutex; + std::condition_variable cv; + int frames_received = 0; + + const std::string track_name = "demo-capture-track"; + receiver_room.setOnVideoFrameEventCallback( + sender_identity, track_name, + [&mutex, &cv, &frames_received](const VideoFrameEvent & /*event*/) { + std::lock_guard lock(mutex); + if (++frames_received >= kMinFramesReceived) { + cv.notify_all(); + } + }); + + // Publish a track backed by the capture source's RTC video source, + // merging application options over the source-derived ones. + auto track = LocalVideoTrack::createLocalVideoTrack(track_name, + capture->videoSource()); + + // Application options are merged in; source-dictated fields win. + TrackPublishOptions app_options; + app_options.source = TrackSource::SOURCE_CAMERA; + ASSERT_NO_THROW(lockLocalParticipant(sender_room) + ->publishTrack(track, capture->publishOptions(app_options))); + + // Observe the capture's terminal event. + std::optional finished; + capture->setOnFinishedCallback([&mutex, &cv, &finished](const CaptureResult &result) { + std::lock_guard lock(mutex); + finished = result; + cv.notify_all(); + }); + + ASSERT_NO_THROW(capture->start()); + EXPECT_THROW(capture->start(), CaptureSourceError) + << "double start must be rejected"; + + { + std::unique_lock lock(mutex); + const bool got_frames = cv.wait_for(lock, 30s, [&frames_received] { + return frames_received >= kMinFramesReceived; + }); + ASSERT_TRUE(got_frames) + << "Timed out waiting for demo capture frames; received " + << frames_received; + } + + // Stop is a signal; the terminal callback delivers the stats. + ASSERT_NO_THROW(capture->stop()); + { + std::unique_lock lock(mutex); + const bool got_finished = + cv.wait_for(lock, 10s, [&finished] { return finished.has_value(); }); + ASSERT_TRUE(got_finished) + << "Timed out waiting for the capture finished callback"; + } + + ASSERT_FALSE(finished->error.has_value()) << *finished->error; + EXPECT_EQ(finished->exit, CaptureExit::Stopped); + EXPECT_GT(finished->frames_captured, 0u); + + receiver_room.clearOnVideoFrameCallback(sender_identity, track_name); + if (track->publication()) { + lockLocalParticipant(sender_room) + ->unpublishTrack(track->publication()->sid()); + } +} + +} // namespace livekit::test