yodau::backend::stream_manager Class Reference

Central coordinator for streams, geometry, frame processing and events. More...

#include <backend/include/stream_manager.hpp>

Public Types
using	local_stream_detector_fn = std::function<std::vector<stream>()>
	Custom detector for local streams.
using	manual_push_fn = std::function<void(const std::string& stream_name, frame&& f)>
	Hook for manual frame pushing.
using	daemon_start_fn
	Hook used to start a background daemon that produces frames.
using	frame_processor_fn = std::function<std::vector<event>(const stream& s, const frame& f)>
	Frame analysis function.
using	event_sink_fn = std::function<void(const event& e)>
	Sink for individual events.
using	event_batch_sink_fn = std::function<void(const std::vector<event>& events)>
	Sink for event batches.

Public Member Functions
	stream_manager ()
	Construct manager and attempt to discover local streams.
void	dump (std::ostream &out) const
	Dump all streams and lines to an output stream.
void	dump_lines (std::ostream &out) const
	Dump all registered lines.
void	dump_stream (std::ostream &out, bool connections=false) const
	Dump all registered streams.
void	set_local_stream_detector (local_stream_detector_fn detector)
	Set a custom local stream detector.
void	refresh_local_streams ()
	Refresh local streams.
stream &	add_stream (const std::string &path, const std::string &name={}, const std::string &type={}, bool loop=true)
	Add a new stream to the manager.
line_ptr	add_line (const std::string &points, bool closed=false, const std::string &name={})
	Add a new line (polyline/polygon) to the manager.
stream &	set_line (const std::string &stream_name, const std::string &line_name)
	Connect an existing line to an existing stream.
std::shared_ptr< const stream >	find_stream (const std::string &name) const
	Find a stream by name.
std::vector< std::string >	stream_names () const
	List names of all registered streams.
std::vector< std::string >	line_names () const
	List names of all registered lines.
std::vector< std::string >	stream_lines (const std::string &stream_name) const
	List names of lines connected to a given stream.
void	set_manual_push_hook (manual_push_fn hook)
	Set manual push hook.
void	set_daemon_start_hook (daemon_start_fn hook)
	Set daemon start hook.
void	push_frame (const std::string &stream_name, frame &&f)
	Push a frame into the manager for a specific stream.
void	start_daemon (const std::string &stream_name)
	Start a daemon for a stream.
void	set_frame_processor (frame_processor_fn fn)
	Set the frame processor.
std::vector< event >	process_frame (const std::string &stream_name, frame &&f)
	Analyze a frame and return generated events.
void	set_event_sink (event_sink_fn fn)
	Set per-event sink.
void	set_event_batch_sink (event_batch_sink_fn fn)
	Set batch event sink.
void	set_analysis_interval_ms (int ms)
	Set minimum analysis interval per stream, in milliseconds.
void	start_stream (const std::string &name)
	Start a stream daemon by name.
void	stop_stream (const std::string &name)
	Stop a running stream daemon by name.
bool	is_stream_running (const std::string &name) const
	Check whether a daemon for a stream is running.
void	enable_fake_events (int interval_ms=700)
	Enable periodic fake events generation.
void	disable_fake_events ()
	Disable fake events generation.
void	set_line_dir (const std::string &line_name, tripwire_dir dir)
	Change the direction constraint of a stored line.

Private Member Functions
std::vector< std::shared_ptr< stream > >	snapshot_streams () const
	Take a snapshot of current streams.
void	snapshot_hooks (frame_processor_fn &fp, event_sink_fn &es, event_batch_sink_fn &bes) const
	Snapshot currently installed hooks.
int	current_fake_interval_ms () const
	Get current fake-event interval.
void	run_fake_events (std::stop_token st)
	Background loop for fake-event generation.

Private Attributes
std::unordered_map< std::string, std::shared_ptr< stream > >	streams
	Registered streams keyed by name.
std::unordered_map< std::string, line_ptr >	lines
	Registered lines keyed by name.
size_t	stream_idx { 0 }
	Auto-name index for streams ("stream_N").
size_t	line_idx { 0 }
	Auto-name index for lines ("line_N").
local_stream_detector_fn	stream_detector {}
	Optional external local stream detector.
manual_push_fn	manual_push
	Optional manual frame push hook.
daemon_start_fn	daemon_start
	Optional daemon start hook.
frame_processor_fn	frame_processor
	Optional frame analysis hook.
event_sink_fn	event_sink
	Optional per-event sink.
event_batch_sink_fn	event_batch_sink
	Optional batch event sink.
int	analysis_interval_ms { 200 }
	Per-stream analysis throttle interval.
std::unordered_map< std::string, std::chrono::steady_clock::time_point >	last_analysis_ts
	Last analysis time per stream.
std::unordered_map< std::string, std::jthread >	daemons
	Running daemon threads keyed by stream name.
std::jthread	fake_thread
	Fake-event generator thread.
int	fake_interval_ms { 700 }
	Current fake-event interval.
bool	fake_enabled { false }
	Whether fake-event generation is enabled.
std::mutex	mtx
	Global mutex protecting manager state.

Detailed Description

Central coordinator for streams, geometry, frame processing and events.

The stream_manager owns:

• a registry of streams (stream) addressable by name,
• a registry of lines (line) addressable by name,
• hooks for stream discovery, frame ingestion, background capture daemons, frame analysis, and event delivery.

Typical responsibilities:

• Add and look up streams/lines.
• Connect lines to streams.
• Start/stop stream daemons (background frame producers).
• Accept manually pushed frames and throttle analysis per stream.
• Deliver produced events to configured sinks.

Thread-safety:

• All public methods lock mtx unless otherwise noted.
• Background threads (daemons and fake-event generator) also use mtx.

Note

All timestamps use std::chrono::steady_clock, i.e. monotonic time.

Definition at line 45 of file stream_manager.hpp.

Member Typedef Documentation

daemon_start_fn

using yodau::backend::stream_manager::daemon_start_fn

Initial value:

 std::function<void(
        const stream& s, std::function<void(frame&&)> on_frame,
        std::stop_token st
    )>

Hook used to start a background daemon that produces frames.

The manager provides an on-frame callback that the daemon should call for each produced frame. The daemon must also respect st and exit promptly when stop is requested.

Parameters

s	Stream to run.
on_frame	Callback to deliver produced frames to the manager.
st	Stop token to observe for cancellation.

Definition at line 78 of file stream_manager.hpp.

event_batch_sink_fn

using yodau::backend::stream_manager::event_batch_sink_fn = std::function<void(const std::vector<event>& events)>

Sink for event batches.

If set, push_frame delivers events to this sink as a batch.

Definition at line 107 of file stream_manager.hpp.

event_sink_fn

using yodau::backend::stream_manager::event_sink_fn = std::function<void(const event& e)>

Sink for individual events.

If batch sink is not set, events are delivered one-by-one to this sink.

Definition at line 100 of file stream_manager.hpp.

frame_processor_fn

using yodau::backend::stream_manager::frame_processor_fn = std::function<std::vector<event>(const stream& s, const frame& f)>

Frame analysis function.

Called by process_frame and (optionally) by the fake-event generator.

Parameters

s	Stream metadata/context.
f	Frame to analyze (const reference).

Returns

Vector of generated events (may be empty).

Definition at line 93 of file stream_manager.hpp.

local_stream_detector_fn

using yodau::backend::stream_manager::local_stream_detector_fn = std::function<std::vector<stream>()>

Custom detector for local streams.

The detector is expected to return a list of streams discovered at call time. Returned streams are moved into the manager if their names are not already present.

Definition at line 54 of file stream_manager.hpp.

manual_push_fn

using yodau::backend::stream_manager::manual_push_fn = std::function<void(const std::string& stream_name, frame&& f)>

Hook for manual frame pushing.

If set via set_manual_push_hook, push_frame will delegate to this hook instead of analyzing frames internally.

Parameters

stream_name	Name of the stream the frame belongs to.
f	Frame to process/consume (moved).

Definition at line 65 of file stream_manager.hpp.

Constructor & Destructor Documentation

stream_manager()

yodau::backend::stream_manager::stream_manager

(

)

Construct manager and attempt to discover local streams.

On Linux, the constructor probes /dev/video* devices and adds those that look like capture devices. After that, if a custom local detector is set, it may be used when refresh_local_streams is called.

Definition at line 44 of file stream_manager.cpp.

{ refresh_local_streams(); }

Member Function Documentation

add_line()

yodau::backend::line_ptr yodau::backend::stream_manager::add_line	(	const std::string &	points,
		bool	closed = false,
		const std::string &	name = {} )

Add a new line (polyline/polygon) to the manager.

If name is empty or already used, a unique name "line_N" is generated. The points string is parsed with parse_points.

Parameters

points	Textual representation of points.
closed	Whether the line is closed.
name	Optional explicit line name.

Returns

Shared pointer to the stored immutable line.

Exceptions

std::runtime_error

on invalid point string.

Definition at line 143 of file stream_manager.cpp.

  {
    std::scoped_lock lock(mtx);
    std::vector<point> parsed_points = parse_points(points);
    std::string line_name = name;
    while (line_name.empty() || lines.contains(line_name)) {
        line_name = "line_" + std::to_string(line_idx++);
    }
    auto new_line = make_line(std::move(parsed_points), line_name, closed);
    lines.emplace(line_name, new_line);
    return new_line;
}

add_stream()

yodau::backend::stream & yodau::backend::stream_manager::add_stream	(	const std::string &	path,
		const std::string &	name = {},
		const std::string &	type = {},
		bool	loop = true )

Add a new stream to the manager.

If name is empty or already used, a unique name "stream_N" is generated.

Parameters

path	Stream path/URL.
name	Optional explicit stream name.
type	Optional explicit type override passed to stream ctor.
loop	Whether file streams should loop on EOF.

Returns

Reference to the stored stream.

Definition at line 128 of file stream_manager.cpp.

  {
    std::scoped_lock lock(mtx);
    std::string stream_name = name;
    while (stream_name.empty() || streams.contains(stream_name)) {
        stream_name = "stream_" + std::to_string(stream_idx++);
    }
    auto new_stream = std::make_shared<stream>(path, stream_name, type, loop);
    auto& ref = *new_stream;
    streams.emplace(stream_name, std::move(new_stream));
    return ref;
}

current_fake_interval_ms()

int yodau::backend::stream_manager::current_fake_interval_ms ( ) const

private

Get current fake-event interval.

Returns

Interval in milliseconds.

Definition at line 491 of file stream_manager.cpp.

                                                               {
    std::scoped_lock lock(mtx);
    return fake_interval_ms;
}

disable_fake_events()

void yodau::backend::stream_manager::disable_fake_events

(

)

Disable fake events generation.

Requests stop on the fake-event thread (if running).

Definition at line 430 of file stream_manager.cpp.

                                                     {
    std::jthread th;
 
    {
        std::scoped_lock lock(mtx);
 
        if (!fake_enabled) {
            return;
        }
 
        fake_enabled = false;
        th = std::move(fake_thread);
        fake_thread = std::jthread();
    }
 
    if (th.joinable()) {
        th.request_stop();
    }
}

dump()

void yodau::backend::stream_manager::dump

(

std::ostream &

out

)

const

Dump all streams and lines to an output stream.

Equivalent to calling dump_stream and dump_lines.

Parameters

out	Output stream.

Definition at line 46 of file stream_manager.cpp.

                                                           {
    std::scoped_lock lock(mtx);
    dump_stream(out);
    out << "\n";
    dump_lines(out);
}

dump_lines()

void yodau::backend::stream_manager::dump_lines

(

std::ostream &

out

)

const

Dump all registered lines.

Each line is printed with line::dump.

Parameters

out	Output stream.

Definition at line 53 of file stream_manager.cpp.

                                                                 {
    std::scoped_lock lock(mtx);
    out << lines.size() << " lines:";
    for (const auto& line : lines | std::views::values) {
        out << "\n\t";
        line->dump(out);
    }
}

dump_stream()

void yodau::backend::stream_manager::dump_stream	(	std::ostream &	out,
		bool	connections = false ) const

Dump all registered streams.

If connections is true, prints connected line names per stream.

Parameters

out	Output stream.
connections	Whether to include stream-line connections.

Definition at line 62 of file stream_manager.cpp.

        {
    std::scoped_lock lock(mtx);
    out << streams.size() << " streams:";
    for (const auto& stream : streams | std::views::values) {
        out << "\n\t";
        stream->dump(out, connections);
    }
}

enable_fake_events()

void yodau::backend::stream_manager::enable_fake_events

(

int

interval_ms = 700

)

Enable periodic fake events generation.

Starts/keeps a background thread that calls the frame processor with a dummy frame on all streams every interval_ms milliseconds.

Parameters

interval_ms

Period in ms (ignored if <= 0).

Definition at line 403 of file stream_manager.cpp.

                                                                         {
    {
        std::scoped_lock lock(mtx);
 
        if (interval_ms > 0) {
            fake_interval_ms = interval_ms;
        }
 
        if (fake_enabled) {
            return;
        }
 
        fake_enabled = true;
    }
 
    std::jthread th([this](std::stop_token st) { run_fake_events(st); });
 
    {
        std::scoped_lock lock(mtx);
        if (!fake_enabled) {
            th.request_stop();
            return;
        }
        fake_thread = std::move(th);
    }
}

find_stream()

std::shared_ptr< const yodau::backend::stream > yodau::backend::stream_manager::find_stream

(

const std::string &

name

)

const

Find a stream by name.

Parameters

name	Stream name.

Returns

Shared pointer to const stream, or empty if not found.

Definition at line 174 of file stream_manager.cpp.

                                                                   {
    std::scoped_lock lock(mtx);
    const auto it = streams.find(name);
    if (it == streams.end()) {
        return {};
    }
    return it->second;
}

is_stream_running()

bool yodau::backend::stream_manager::is_stream_running

(

const std::string &

name

)

const

Check whether a daemon for a stream is running.

Parameters

name	Stream name.

Returns

true if running.

Definition at line 396 of file stream_manager.cpp.

        {
    std::scoped_lock lock(mtx);
    return daemons.contains(name);
}

line_names()

std::vector< std::string > yodau::backend::stream_manager::line_names

(

)

const

List names of all registered lines.

Definition at line 189 of file stream_manager.cpp.

                                                                  {
    std::scoped_lock lock(mtx);
    return lines | std::views::keys
        | std::ranges::to<std::vector<std::string>>();
}

process_frame()

std::vector< yodau::backend::event > yodau::backend::stream_manager::process_frame	(	const std::string &	stream_name,
		frame &&	f )

Analyze a frame and return generated events.

Analysis is throttled per stream by analysis_interval_ms. If the stream does not exist or processor is not set, returns empty list.

Parameters

stream_name	Stream name.
f	Frame to analyze (moved into function; read-only for processor).

Returns

Vector of produced events.

Definition at line 267 of file stream_manager.cpp.

  {
    std::shared_ptr<stream> sp;
    frame_processor_fn fp;
    const auto now = std::chrono::steady_clock::now();
    bool allow = false;
 
    {
        std::scoped_lock lock(mtx);
        auto it = streams.find(stream_name);
        if (it == streams.end() || !frame_processor) {
            return {};
        }
 
        fp = frame_processor;
        sp = it->second;
 
        const auto last_it = last_analysis_ts.find(stream_name);
        if (last_it == last_analysis_ts.end()) {
            last_analysis_ts[stream_name] = now;
            allow = true;
        } else {
            const auto dt
                = std::chrono::duration_cast<std::chrono::milliseconds>(
                      now - last_it->second
                )
                      .count();
            if (dt >= analysis_interval_ms) {
                last_analysis_ts[stream_name] = now;
                allow = true;
            }
        }
    }
 
    if (!allow || !sp || !fp) {
        return {};
    }
 
    return fp(*sp, f);
}

push_frame()

void yodau::backend::stream_manager::push_frame	(	const std::string &	stream_name,
		frame &&	f )

Push a frame into the manager for a specific stream.

Workflow:

• If manual push hook is set, delegate to it.
• Else analyze frame with process_frame (throttled).
• If batch sink is set, deliver whole batch.
• Else if single-event sink is set, deliver events one-by-one.

Parameters

stream_name	Stream name.
f	Frame to process (moved).

Definition at line 218 of file stream_manager.cpp.

  {
    manual_push_fn mp;
    event_sink_fn es;
    event_batch_sink_fn bes;
 
    {
        std::scoped_lock lock(mtx);
        mp = manual_push;
        es = event_sink;
        bes = event_batch_sink;
    }
 
    if (mp) {
        mp(stream_name, std::move(f));
        return;
    }
 
    auto events = process_frame(stream_name, std::move(f));
 
    if (bes) {
        bes(events);
        return;
    }
 
    if (!es) {
        return;
    }
 
    for (const auto& e : events) {
        es(e);
    }
}

refresh_local_streams()

void yodau::backend::stream_manager::refresh_local_streams

(

)

Refresh local streams.

Behavior:

• On Linux: scans /dev/video* devices, validates capture capability, and auto-adds any not yet registered.
• If a custom detector is set: calls it and adds returned streams that are not yet registered.

Definition at line 83 of file stream_manager.cpp.

                                                       {
#ifdef __linux__
    for (size_t idx = 0;; ++idx) {
        std::string path = "/dev/video" + std::to_string(idx);
        if (!std::filesystem::exists(path)) {
            break;
        }
        if (!is_capture_device(path)) {
            continue;
        }
 
        {
            std::scoped_lock lock(mtx);
            if (streams.contains("video" + std::to_string(idx))) {
                continue;
            }
        }
 
        add_stream(path, "video" + std::to_string(idx), "local");
    }
#endif
 
    local_stream_detector_fn det;
    {
        std::scoped_lock lock(mtx);
        det = stream_detector;
    }
 
    if (!det) {
        return;
    }
 
    auto detected_streams = det();
    for (auto& detected_stream : detected_streams) {
        const auto name = detected_stream.get_name();
 
        std::scoped_lock lock(mtx);
        if (!streams.contains(name)) { // todo: update existing streams?
            streams.emplace(
                name, std::make_shared<stream>(std::move(detected_stream))
            );
        }
    }
}

run_fake_events()

void yodau::backend::stream_manager::run_fake_events ( std::stop_token st )

private

Background loop for fake-event generation.

Periodically calls frame processor with a dummy frame on all streams and delivers events to configured sinks until st requests stop.

Parameters

st	Stop token.

Definition at line 496 of file stream_manager.cpp.

                                                                 {
    frame dummy;
 
    while (!st.stop_requested()) {
        auto snap = snapshot_streams();
 
        frame_processor_fn fp;
        event_sink_fn es;
        event_batch_sink_fn bes;
        snapshot_hooks(fp, es, bes);
 
        if (fp) {
            for (const auto& sp : snap) {
                auto evs = fp(*sp, dummy);
 
                if (bes) {
                    if (!evs.empty()) {
                        bes(evs);
                    }
                } else if (es) {
                    for (const auto& e : evs) {
                        es(e);
                    }
                }
            }
        }
 
        const int interval = current_fake_interval_ms();
        std::this_thread::sleep_for(std::chrono::milliseconds(interval));
    }
}

set_analysis_interval_ms()

void yodau::backend::stream_manager::set_analysis_interval_ms

(

int

ms

)

Set minimum analysis interval per stream, in milliseconds.

Values <= 0 are ignored.

Parameters

ms	Interval in ms.

Definition at line 321 of file stream_manager.cpp.

                                                                {
    if (ms <= 0) {
        return;
    }
    std::scoped_lock lock(mtx);
    analysis_interval_ms = ms;
}

set_daemon_start_hook()

void yodau::backend::stream_manager::set_daemon_start_hook

(

daemon_start_fn

hook

)

Set daemon start hook.

This hook is required for start_stream to do anything.

Parameters

hook	Hook functor (may be empty to unset).

Definition at line 211 of file stream_manager.cpp.

  {
    std::scoped_lock lock(mtx);
    daemon_start = std::move(hook);
}

set_event_batch_sink()

void yodau::backend::stream_manager::set_event_batch_sink

(

event_batch_sink_fn

fn

)

Set batch event sink.

When set, overrides per-event sink for delivery.

Parameters

fn	Sink functor (may be empty to unset).

Definition at line 314 of file stream_manager.cpp.

  {
    std::scoped_lock lock(mtx);
    event_batch_sink = std::move(fn);
}

set_event_sink()

void yodau::backend::stream_manager::set_event_sink

(

event_sink_fn

fn

)

Set per-event sink.

Ignored when batch sink is set.

Parameters

fn	Sink functor (may be empty to unset).

Definition at line 309 of file stream_manager.cpp.

                                                                {
    std::scoped_lock lock(mtx);
    event_sink = std::move(fn);
}

set_frame_processor()

void yodau::backend::stream_manager::set_frame_processor

(

frame_processor_fn

fn

)

Set the frame processor.

Parameters

fn	Analysis functor (may be empty to unset).

Definition at line 259 of file stream_manager.cpp.

  {
    std::scoped_lock lock(mtx);
    frame_processor = std::move(fn);
}

set_line()

yodau::backend::stream & yodau::backend::stream_manager::set_line	(	const std::string &	stream_name,
		const std::string &	line_name )

Connect an existing line to an existing stream.

Parameters

stream_name	Name of the target stream.
line_name	Name of the line to connect.

Returns

Reference to the stream.

Exceptions

std::runtime_error

if stream or line is not found.

Definition at line 157 of file stream_manager.cpp.

  {
    std::scoped_lock lock(mtx);
    const auto stream_it = streams.find(stream_name);
    if (stream_it == streams.end()) {
        throw std::runtime_error("stream not found: " + stream_name);
    }
    const auto line_it = lines.find(line_name);
    if (line_it == lines.end()) {
        throw std::runtime_error("line not found: " + line_name);
    }
    stream_it->second->connect_line(line_it->second);
    return *stream_it->second;
}

set_line_dir()

void yodau::backend::stream_manager::set_line_dir	(	const std::string &	line_name,
		tripwire_dir	dir )

Change the direction constraint of a stored line.

Since lines are stored as immutable shared pointers, this method clones the line, changes line::dir, and replaces the pointer in the registry.

Parameters

line_name	Name of the line to reconfigure.
dir	New direction constraint.

Exceptions

std::runtime_error

if the line does not exist.

Definition at line 450 of file stream_manager.cpp.

  {
    std::scoped_lock lock(mtx);
 
    const auto it = lines.find(line_name);
    if (it == lines.end() || !it->second) {
        throw std::runtime_error("line not found: " + line_name);
    }
 
    const auto old_ptr = it->second;
    auto new_ptr = std::make_shared<line>(*old_ptr);
    new_ptr->dir = dir;
 
    it->second = new_ptr;
}

set_local_stream_detector()

void yodau::backend::stream_manager::set_local_stream_detector

(

local_stream_detector_fn

detector

)

Set a custom local stream detector.

The detector is stored and refresh_local_streams is called immediately after setting.

Parameters

detector

Detector functor.

Definition at line 73 of file stream_manager.cpp.

  {
    {
        std::scoped_lock lock(mtx);
        stream_detector = std::move(detector);
    }
    refresh_local_streams();
}

set_manual_push_hook()

void yodau::backend::stream_manager::set_manual_push_hook

(

manual_push_fn

hook

)

Set manual push hook.

When set, push_frame will call this hook and return immediately, skipping internal analysis and sinks.

Parameters

hook	Hook functor (may be empty to unset).

Definition at line 206 of file stream_manager.cpp.

                                                                         {
    std::scoped_lock lock(mtx);
    manual_push = std::move(hook);
}

snapshot_hooks()

void yodau::backend::stream_manager::snapshot_hooks ( frame_processor_fn & fp, event_sink_fn & es, event_batch_sink_fn & bes ) const

private

Snapshot currently installed hooks.

Copies hooks under lock so that background threads can use stable callables without racing with setters.

Parameters

fp	Out: frame processor.
es	Out: per-event sink.
bes	Out: batch sink.

Definition at line 482 of file stream_manager.cpp.

        {
    std::scoped_lock lock(mtx);
    fp = frame_processor;
    es = event_sink;
    bes = event_batch_sink;
}

snapshot_streams()

std::vector< std::shared_ptr< yodau::backend::stream > > yodau::backend::stream_manager::snapshot_streams ( ) const

private

Take a snapshot of current streams.

Used to iterate without holding the manager lock for long periods.

Returns

Vector of shared pointers to streams.

Definition at line 468 of file stream_manager.cpp.

                                                   {
    std::vector<std::shared_ptr<stream>> snap;
 
    std::scoped_lock lock(mtx);
    snap.reserve(streams.size());
    for (auto& sp : streams | std::views::values) {
        if (sp) {
            snap.push_back(sp);
        }
    }
 
    return snap;
}

start_daemon()

void yodau::backend::stream_manager::start_daemon

(

const std::string &

stream_name

)

Start a daemon for a stream.

Alias for start_stream.

Parameters

stream_name

Stream name.

Definition at line 253 of file stream_manager.cpp.

  {
    start_stream(stream_name);
}

start_stream()

void yodau::backend::stream_manager::start_stream

(

const std::string &

name

)

Start a stream daemon by name.

Requirements:

• daemon_start hook must be set,
• stream must exist,
• daemon for this stream must not already be running.

On Linux, local capture devices may be validated again before starting.

Parameters

name	Stream name.

Definition at line 329 of file stream_manager.cpp.

                                                                   {
    std::shared_ptr<stream> sp;
    daemon_start_fn ds;
 
    {
        std::scoped_lock lock(mtx);
        if (!daemon_start || daemons.contains(name)) {
            return;
        }
 
        const auto it = streams.find(name);
        if (it == streams.end() || !it->second) {
            return;
        }
 
        sp = it->second;
        ds = daemon_start;
 
#ifdef __linux__
        if (!is_linux_capture_ok(*sp)) {
            return;
        }
#endif
 
        sp->activate(stream_pipeline::automatic);
    }
 
    std::jthread th([this, name, sp, ds](std::stop_token st) mutable {
        ds(
            *sp, [this, name](frame&& f) { push_frame(name, std::move(f)); }, st
        );
    });
 
    {
        std::scoped_lock lock(mtx);
        daemons.emplace(name, std::move(th));
    }
}

References yodau::backend::automatic.

stop_stream()

void yodau::backend::stream_manager::stop_stream

(

const std::string &

name

)

Stop a running stream daemon by name.

Requests stop on the associated std::jthread and deactivates the stream (sets pipeline to stream_pipeline::none).

Parameters

name	Stream name.

Definition at line 368 of file stream_manager.cpp.

                                                                  {
    std::jthread th;
    std::shared_ptr<stream> sp;
 
    {
        std::scoped_lock lock(mtx);
        const auto it = daemons.find(name);
        if (it == daemons.end()) {
            return;
        }
 
        th = std::move(it->second);
        daemons.erase(it);
 
        const auto sit = streams.find(name);
        if (sit != streams.end()) {
            sp = sit->second;
        }
    }
 
    th.request_stop();
 
    if (sp) {
        std::scoped_lock lock(mtx);
        sp->deactivate();
    }
}

stream_lines()

std::vector< std::string > yodau::backend::stream_manager::stream_lines

(

const std::string &

stream_name

)

const

List names of lines connected to a given stream.

Parameters

stream_name

Stream name.

Returns

Vector of connected line names, or empty if stream not found.

Definition at line 195 of file stream_manager.cpp.

        {
    std::scoped_lock lock(mtx);
    const auto stream_it = streams.find(stream_name);
    if (stream_it == streams.end()) {
        return {};
    }
    return stream_it->second->line_names();
}

stream_names()

std::vector< std::string > yodau::backend::stream_manager::stream_names

(

)

const

List names of all registered streams.

Definition at line 183 of file stream_manager.cpp.

                                                                    {
    std::scoped_lock lock(mtx);
    return streams | std::views::keys
        | std::ranges::to<std::vector<std::string>>();
}

Member Data Documentation

analysis_interval_ms

int yodau::backend::stream_manager::analysis_interval_ms { 200 }

private

Per-stream analysis throttle interval.

Definition at line 473 of file stream_manager.hpp.

daemon_start

daemon_start_fn yodau::backend::stream_manager::daemon_start

private

Optional daemon start hook.

Definition at line 461 of file stream_manager.hpp.

daemons

std::unordered_map<std::string, std::jthread> yodau::backend::stream_manager::daemons

private

Running daemon threads keyed by stream name.

Definition at line 484 of file stream_manager.hpp.

event_batch_sink

event_batch_sink_fn yodau::backend::stream_manager::event_batch_sink

private

Optional batch event sink.

Definition at line 470 of file stream_manager.hpp.

event_sink

event_sink_fn yodau::backend::stream_manager::event_sink

private

Optional per-event sink.

Definition at line 467 of file stream_manager.hpp.

fake_enabled

bool yodau::backend::stream_manager::fake_enabled { false }

private

Whether fake-event generation is enabled.

Definition at line 493 of file stream_manager.hpp.

{ 700 };

fake_interval_ms

int yodau::backend::stream_manager::fake_interval_ms { 700 }

private

Current fake-event interval.

Definition at line 490 of file stream_manager.hpp.

fake_thread

std::jthread yodau::backend::stream_manager::fake_thread

private

Fake-event generator thread.

Definition at line 487 of file stream_manager.hpp.

frame_processor

frame_processor_fn yodau::backend::stream_manager::frame_processor

private

Optional frame analysis hook.

Definition at line 464 of file stream_manager.hpp.

last_analysis_ts

std::unordered_map<std::string, std::chrono::steady_clock::time_point> yodau::backend::stream_manager::last_analysis_ts

private

Last analysis time per stream.

Used to enforce analysis_interval_ms.

Definition at line 481 of file stream_manager.hpp.

line_idx

size_t yodau::backend::stream_manager::line_idx { 0 }

private

Auto-name index for lines ("line_N").

Definition at line 452 of file stream_manager.hpp.

{ 0 };

lines

std::unordered_map<std::string, line_ptr> yodau::backend::stream_manager::lines

private

Registered lines keyed by name.

Definition at line 446 of file stream_manager.hpp.

manual_push

manual_push_fn yodau::backend::stream_manager::manual_push

private

Optional manual frame push hook.

Definition at line 458 of file stream_manager.hpp.

mtx

std::mutex yodau::backend::stream_manager::mtx

mutableprivate

Global mutex protecting manager state.

Definition at line 496 of file stream_manager.hpp.

stream_detector

local_stream_detector_fn yodau::backend::stream_manager::stream_detector {}

private

Optional external local stream detector.

Definition at line 455 of file stream_manager.hpp.

{ 0 };

stream_idx

size_t yodau::backend::stream_manager::stream_idx { 0 }

private

Auto-name index for streams ("stream_N").

Definition at line 449 of file stream_manager.hpp.

streams

std::unordered_map<std::string, std::shared_ptr<stream> > yodau::backend::stream_manager::streams

private

Registered streams keyed by name.

Definition at line 443 of file stream_manager.hpp.

---

The documentation for this class was generated from the following files:

• stream_manager.hpp
• stream_manager.cpp