/* streamer-tools OBS Camera Plugin - LiveKit session wrapper Copyright (C) 2026 CyberCoveLLC 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 */ #include "stplugin/session.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include namespace stplugin { namespace { // --- LiveKit <-> plugin type conversion ------------------------------------ MediaKind toMediaKind(livekit::TrackKind kind) { switch (kind) { case livekit::TrackKind::KIND_AUDIO: return MediaKind::Audio; case livekit::TrackKind::KIND_VIDEO: return MediaKind::Video; case livekit::TrackKind::KIND_UNKNOWN: break; } return MediaKind::Unknown; } MediaSource toMediaSource(livekit::TrackSource source) { switch (source) { case livekit::TrackSource::SOURCE_CAMERA: return MediaSource::Camera; case livekit::TrackSource::SOURCE_MICROPHONE: return MediaSource::Microphone; case livekit::TrackSource::SOURCE_SCREENSHARE: return MediaSource::Screenshare; case livekit::TrackSource::SOURCE_SCREENSHARE_AUDIO: return MediaSource::ScreenshareAudio; case livekit::TrackSource::SOURCE_UNKNOWN: break; } return MediaSource::Unknown; } livekit::VideoBufferType toLiveKitBufferType(PixelFormat format) { switch (format) { case PixelFormat::I420: return livekit::VideoBufferType::I420; case PixelFormat::NV12: return livekit::VideoBufferType::NV12; case PixelFormat::BGRA: return livekit::VideoBufferType::BGRA; } return livekit::VideoBufferType::I420; } /// Returns false when the SDK handed us a format the OBS adapter cannot /// consume, in which case the caller converts. bool fromLiveKitBufferType(livekit::VideoBufferType type, PixelFormat &out) { switch (type) { case livekit::VideoBufferType::I420: out = PixelFormat::I420; return true; case livekit::VideoBufferType::NV12: out = PixelFormat::NV12; return true; case livekit::VideoBufferType::BGRA: out = PixelFormat::BGRA; return true; default: return false; } } /// Which disconnect reasons are worth telling the operator "this will not fix /// itself" about. Everything else is reported as an ordinary disconnect, /// because the SDK's own reconnect logic covers it. bool isFatalDisconnect(livekit::DisconnectReason reason) { switch (reason) { case livekit::DisconnectReason::DuplicateIdentity: case livekit::DisconnectReason::ParticipantRemoved: case livekit::DisconnectReason::RoomDeleted: case livekit::DisconnectReason::JoinFailure: case livekit::DisconnectReason::UserRejected: return true; default: return false; } } const char *describeDisconnectReason(livekit::DisconnectReason reason) { switch (reason) { case livekit::DisconnectReason::Unknown: return "connection lost"; case livekit::DisconnectReason::ClientInitiated: return "disconnected"; case livekit::DisconnectReason::DuplicateIdentity: return "another client joined with the same identity"; case livekit::DisconnectReason::ServerShutdown: return "the LiveKit server is shutting down"; case livekit::DisconnectReason::ParticipantRemoved: return "removed from the room"; case livekit::DisconnectReason::RoomDeleted: return "the room was deleted"; case livekit::DisconnectReason::StateMismatch: return "session could not be resumed"; case livekit::DisconnectReason::JoinFailure: return "could not join the room (token rejected or expired?)"; case livekit::DisconnectReason::Migration: return "migrating to another server"; case livekit::DisconnectReason::SignalClose: return "the signalling connection closed"; case livekit::DisconnectReason::RoomClosed: return "the room closed"; case livekit::DisconnectReason::UserUnavailable: return "user unavailable"; case livekit::DisconnectReason::UserRejected: return "connection rejected"; case livekit::DisconnectReason::SipTrunkFailure: return "SIP trunk failure"; case livekit::DisconnectReason::ConnectionTimeout: return "connection timed out"; case livekit::DisconnectReason::MediaFailure: return "media connection failed"; case livekit::DisconnectReason::AgentError: return "agent error"; } return "disconnected"; } // --- Process-wide SDK lifetime --------------------------------------------- std::mutex &globalMutex() { static std::mutex m; return m; } int &globalRefCount() { static int n = 0; return n; } } // namespace // --------------------------------------------------------------------------- // Impl // --------------------------------------------------------------------------- struct LiveKitSession::Impl : public livekit::RoomDelegate { enum class CommandType { AttachVideo, DetachVideo, AttachAudio, DetachAudio, Stop }; struct Command { CommandType type; std::shared_ptr track; }; livekit::Room room; SessionConfig config; mutable std::mutex state_mutex; SessionStateMachine machine; VideoFrameHandler on_video; AudioFrameHandler on_audio; SessionStateHandler on_state; std::atomic video_frames{0}; std::atomic audio_frames{0}; std::atomic dropped_frames{0}; // Command queue. Every interaction with livekit::VideoStream / // livekit::AudioStream happens on `worker`, never on a room event thread: // the SDK's room callbacks run on its own event thread and blocking or // re-entering there stalls every other event (and Room::disconnect() from // inside one is documented to deadlock outright). std::mutex queue_mutex; std::condition_variable queue_cv; std::deque queue; std::thread worker; bool worker_running = false; // Owned exclusively by the worker thread. std::shared_ptr video_stream; std::thread video_thread; std::shared_ptr audio_stream; std::thread audio_thread; bool connected = false; ~Impl() override = default; // --- state helpers ----------------------------------------------------- template void mutateState(Fn &&fn) { SessionState state; std::string detail; SessionStateHandler handler; { std::lock_guard guard(state_mutex); fn(machine); state = machine.state(); detail = machine.detail(); handler = on_state; } // Notified outside the lock: the handler is OBS adapter code and must // never be able to deadlock against a concurrent state query. if (handler) handler(state, detail); } void post(CommandType type, std::shared_ptr track = nullptr) { { std::lock_guard guard(queue_mutex); if (!worker_running) return; queue.push_back(Command{type, std::move(track)}); } queue_cv.notify_one(); } // Handles the wanted video track once matched, shared by onTrackSubscribed // (a fresh subscription) and attachExistingTracks (one already up when // this session started watching). Two responsibilities that only make // sense together, both keyed off the SAME publication: // // - subscribe_video: an audio-only source (the soundboard) never wants // this video at all. Rather than attach it and let the OBS adapter // discard every decoded frame, disable the publication itself // (RemoteTrackPublication::setEnabled(false)) so the SFU stops // sending it -- real bandwidth saved, not just wasted decode. // - Fixed video quality: LiveKit's default subscriber behaviour lets // the SFU switch simulcast layers per its own adaptive/bandwidth // logic, which for a source with no rendered-size hint (this is a // native C++ subscriber, not a sized