amends to prev commit (oops)

the test source thread will always exist for now
the debug view will open a window for each connection

src/CMakeLists.txt

@@ -102,6 +102,21 @@ target_sources(tomato PUBLIC
 
 	./chat_gui4.hpp
 	./chat_gui4.cpp
+
+	./frame_streams/frame_stream2.hpp
+	./frame_streams/audio_stream2.hpp
+	./frame_streams/stream_manager.hpp
+	./frame_streams/stream_manager.cpp
+
+	./frame_streams/sdl/sdl_audio2_frame_stream2.hpp
+	./frame_streams/sdl/sdl_audio2_frame_stream2.cpp
+	./frame_streams/sdl/video.hpp
+
+	./stream_manager_ui.hpp
+	./stream_manager_ui.cpp
+
+	./debug_video_tap.hpp
+	./debug_video_tap.cpp
 )
 
 if (TOMATO_TOX_AV)

src/chat_gui/file_selector.cpp

@@ -111,6 +111,7 @@ void FileSelector::render(void) {
 				}
 			}
 
+			try {
 				// do sorting here
 				// TODO: cache the result (lol)
 				if (ImGuiTableSortSpecs* sorts_specs = ImGui::TableGetSortSpecs(); sorts_specs != nullptr && sorts_specs->SpecsCount >= 1) {
@@ -162,6 +163,9 @@ void FileSelector::render(void) {
 						break; default: ;
 					}
 				}
+			} catch (...) {
+				// we likely saw a file disapear
+			}
 
 			for (auto const& dir_entry : dirs) {
 				if (ImGui::TableNextColumn()) {

src/debug_video_tap.cpp

@@ -0,0 +1,296 @@
+#include "./debug_video_tap.hpp"
+
+#include <solanaceae/object_store/object_store.hpp>
+
+#include <entt/entity/entity.hpp>
+
+#include <SDL3/SDL.h>
+
+#include <imgui/imgui.h>
+
+#include "./frame_streams/sdl/video.hpp"
+#include "./frame_streams/frame_stream2.hpp"
+
+#include <string>
+#include <memory>
+#include <mutex>
+#include <deque>
+
+#include <thread>
+#include <chrono>
+#include <atomic>
+
+#include <iostream>
+
+// fwd
+namespace Message {
+uint64_t getTimeMS(void);
+}
+
+// threadsafe queue frame stream
+// protected by a simple mutex lock
+template<typename FrameType>
+struct LockedFrameStream2 : public FrameStream2I<FrameType> {
+	std::mutex _lock;
+
+	std::deque<FrameType> _frames;
+
+	~LockedFrameStream2(void) {}
+
+	int32_t size(void) { return -1; }
+
+	std::optional<FrameType> pop(void) {
+		std::lock_guard lg{_lock};
+
+		if (_frames.empty()) {
+			return std::nullopt;
+		}
+
+		FrameType new_frame = std::move(_frames.front());
+		_frames.pop_front();
+
+		return std::move(new_frame);
+	}
+
+	bool push(const FrameType& value) {
+		std::lock_guard lg{_lock};
+
+		_frames.push_back(value);
+
+		return true;
+	}
+};
+
+struct DebugVideoTapSink : public FrameStream2SinkI<SDLVideoFrame> {
+	TextureUploaderI& _tu;
+
+	uint32_t _id_counter {0};
+
+	struct Writer {
+		struct View {
+			uint32_t _id {0}; // for stable imgui ids
+
+			uint64_t _tex {0};
+			uint32_t _tex_w {0};
+			uint32_t _tex_h {0};
+
+			bool _mirror {false}; // flip horizontally
+
+			uint64_t _v_last_ts {0}; // us
+			float _v_interval_avg {0.f}; // s
+		} view;
+
+		std::shared_ptr<LockedFrameStream2<SDLVideoFrame>> stream;
+	};
+	std::vector<Writer> _writers;
+
+	DebugVideoTapSink(TextureUploaderI& tu) : _tu(tu) {}
+	~DebugVideoTapSink(void) {}
+
+	// sink
+	std::shared_ptr<FrameStream2I<SDLVideoFrame>> subscribe(void) override {
+		_writers.emplace_back(Writer{
+			Writer::View{_id_counter++},
+			std::make_shared<LockedFrameStream2<SDLVideoFrame>>()
+		});
+
+		return _writers.back().stream;
+	}
+
+	bool unsubscribe(const std::shared_ptr<FrameStream2I<SDLVideoFrame>>& sub) override {
+		if (!sub || _writers.empty()) {
+			// nah
+			return false;
+		}
+
+		for (auto it = _writers.cbegin(); it != _writers.cend(); it++) {
+			if (it->stream == sub) {
+				_tu.destroy(it->view._tex);
+				_writers.erase(it);
+				return true;
+			}
+		}
+
+		// what
+		return false;
+	}
+};
+
+struct DebugVideoTestSource : public FrameStream2SourceI<SDLVideoFrame> {
+	std::vector<std::shared_ptr<LockedFrameStream2<SDLVideoFrame>>> _readers;
+
+	std::atomic_bool _stop {false};
+	std::thread _thread;
+
+	DebugVideoTestSource(void) {
+		std::cout << "DVTS: starting new test video source\n";
+		_thread = std::thread([this](void) {
+			while (!_stop) {
+				if (!_readers.empty()) {
+					auto* surf = SDL_CreateSurface(960, 720, SDL_PIXELFORMAT_ARGB32);
+
+					// color
+					static auto start_time = Message::getTimeMS();
+					const float time = (Message::getTimeMS() - start_time)/1000.f;
+					SDL_ClearSurface(surf, std::sin(time), std::cos(time), 0.5f, 1.f);
+
+					SDLVideoFrame frame{ // non-owning
+						Message::getTimeMS()*1000,
+						surf,
+					};
+
+					for (auto& stream : _readers) {
+						stream->push(frame); // copy
+					}
+
+					SDL_DestroySurface(surf);
+				}
+
+				std::this_thread::sleep_for(std::chrono::milliseconds(50));
+			}
+		});
+	}
+	~DebugVideoTestSource(void) {
+		_stop = true;
+		_thread.join();
+	}
+
+	std::shared_ptr<FrameStream2I<SDLVideoFrame>> subscribe(void) override {
+		return _readers.emplace_back(std::make_shared<LockedFrameStream2<SDLVideoFrame>>());
+	}
+
+	bool unsubscribe(const std::shared_ptr<FrameStream2I<SDLVideoFrame>>& sub) override {
+		for (auto it = _readers.cbegin(); it != _readers.cend(); it++) {
+			if (it->get() == sub.get()) {
+				_readers.erase(it);
+				return true;
+			}
+		}
+
+		return false;
+	}
+};
+
+DebugVideoTap::DebugVideoTap(ObjectStore2& os, StreamManager& sm, TextureUploaderI& tu) : _os(os), _sm(sm), _tu(tu) {
+	// post self as video sink
+	_tap = {_os.registry(), _os.registry().create()};
+	try {
+		auto dvts = std::make_unique<DebugVideoTapSink>(_tu);
+		_tap.emplace<DebugVideoTapSink*>(dvts.get()); // to get our data back
+		_tap.emplace<Components::FrameStream2Sink<SDLVideoFrame>>(
+			std::move(dvts)
+		);
+
+		_tap.emplace<Components::StreamSink>(Components::StreamSink::create<SDLVideoFrame>("DebugVideoTap"));
+		_tap.emplace<Components::TagDefaultTarget>();
+
+		_os.throwEventConstruct(_tap);
+	} catch (...) {
+		_os.registry().destroy(_tap);
+	}
+
+	_src = {_os.registry(), _os.registry().create()};
+	try {
+		auto dvts = std::make_unique<DebugVideoTestSource>();
+		_src.emplace<DebugVideoTestSource*>(dvts.get());
+		_src.emplace<Components::FrameStream2Source<SDLVideoFrame>>(
+			std::move(dvts)
+		);
+
+		_src.emplace<Components::StreamSource>(Components::StreamSource::create<SDLVideoFrame>("DebugVideoTest"));
+
+		_os.throwEventConstruct(_src);
+	} catch (...) {
+		_os.registry().destroy(_src);
+	}
+}
+
+DebugVideoTap::~DebugVideoTap(void) {
+	if (static_cast<bool>(_tap)) {
+		_os.registry().destroy(_tap);
+	}
+	if (static_cast<bool>(_src)) {
+		_os.registry().destroy(_src);
+	}
+}
+
+float DebugVideoTap::render(void) {
+	float min_interval {2.f};
+	auto& dvtsw = _tap.get<DebugVideoTapSink*>()->_writers;
+	for (auto& [view, stream] : dvtsw) {
+		std::string window_title {"DebugVideoTap #"};
+		window_title += std::to_string(view._id);
+		ImGui::SetNextWindowSize({250, 250}, ImGuiCond_Appearing);
+		if (ImGui::Begin(window_title.c_str())) {
+			while (auto new_frame_opt = stream->pop()) {
+				// timing
+				if (view._v_last_ts == 0) {
+					view._v_last_ts = new_frame_opt.value().timestampUS;
+				} else {
+					auto delta = int64_t(new_frame_opt.value().timestampUS) - int64_t(view._v_last_ts);
+					view._v_last_ts = new_frame_opt.value().timestampUS;
+
+					if (view._v_interval_avg == 0) {
+						view._v_interval_avg = delta/1'000'000.f;
+					} else {
+						const float r = 0.2f;
+						view._v_interval_avg = view._v_interval_avg * (1.f-r) + (delta/1'000'000.f) * r;
+					}
+				}
+
+				SDL_Surface* new_frame_surf = new_frame_opt.value().surface.get();
+
+				SDL_Surface* converted_surf = new_frame_surf;
+				if (new_frame_surf->format != SDL_PIXELFORMAT_RGBA32) {
+					// we need to convert
+					//std::cerr << "DVT: need to convert\n";
+					converted_surf = SDL_ConvertSurfaceAndColorspace(new_frame_surf, SDL_PIXELFORMAT_RGBA32, nullptr, SDL_COLORSPACE_RGB_DEFAULT, 0);
+					assert(converted_surf->format == SDL_PIXELFORMAT_RGBA32);
+				}
+
+				SDL_LockSurface(converted_surf);
+				if (view._tex == 0 || (int)view._tex_w != converted_surf->w || (int)view._tex_h != converted_surf->h) {
+					_tu.destroy(view._tex);
+					view._tex = _tu.uploadRGBA(
+						static_cast<const uint8_t*>(converted_surf->pixels),
+						converted_surf->w,
+						converted_surf->h,
+						TextureUploaderI::LINEAR,
+						TextureUploaderI::STREAMING
+					);
+
+					view._tex_w = converted_surf->w;
+					view._tex_h = converted_surf->h;
+				} else {
+					_tu.updateRGBA(view._tex, static_cast<const uint8_t*>(converted_surf->pixels), converted_surf->w * converted_surf->h * 4);
+				}
+				SDL_UnlockSurface(converted_surf);
+
+				if (new_frame_surf != converted_surf) {
+					// clean up temp
+					SDL_DestroySurface(converted_surf);
+				}
+			}
+
+			ImGui::Checkbox("mirror", &view._mirror);
+
+			// img here
+			if (view._tex != 0) {
+				ImGui::SameLine();
+				ImGui::Text("moving avg interval: %f", view._v_interval_avg);
+				const float img_w = ImGui::GetContentRegionAvail().x;
+				ImGui::Image(
+					reinterpret_cast<ImTextureID>(view._tex),
+					ImVec2{img_w, img_w * float(view._tex_h)/view._tex_w},
+					ImVec2{view._mirror?1.f:0.f, 0},
+					ImVec2{view._mirror?0.f:1.f, 1}
+				);
+			}
+
+		}
+		ImGui::End();
+	}
+
+	return min_interval;
+}
+

src/debug_video_tap.hpp

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <solanaceae/object_store/fwd.hpp>
+#include "./frame_streams/stream_manager.hpp"
+#include "./texture_uploader.hpp"
+
+// provides a sink and a small window displaying a SDLVideoFrame
+// HACK: provides a test video source
+class DebugVideoTap {
+	ObjectStore2& _os;
+	StreamManager& _sm;
+	TextureUploaderI& _tu;
+
+	ObjectHandle _tap;
+	ObjectHandle _src;
+
+	public:
+		DebugVideoTap(ObjectStore2& os, StreamManager& sm, TextureUploaderI& tu);
+		~DebugVideoTap(void);
+
+		float render(void);
+};
+

src/frame_streams/audio_stream2.hpp

@@ -0,0 +1,39 @@
+#pragma once
+
+#include "./frame_stream2.hpp"
+
+#include <solanaceae/util/span.hpp>
+
+#include <cstdint>
+#include <variant>
+#include <vector>
+
+// raw audio
+// channels make samples interleaved,
+// planar channels are not supported
+// s16 only stopgap audio frame (simplified)
+struct AudioFrame2 {
+	// samples per second
+	uint32_t sample_rate {48'000};
+
+	// only >0 is valid
+	size_t channels {0};
+
+	std::variant<
+		std::vector<int16_t>, // S16, platform endianess
+		Span<int16_t> // non owning variant, for direct consumption
+	> buffer;
+
+	// helpers
+	Span<int16_t> getSpan(void) const {
+		if (std::holds_alternative<std::vector<int16_t>>(buffer)) {
+			return Span<int16_t>{std::get<std::vector<int16_t>>(buffer)};
+		} else {
+			return std::get<Span<int16_t>>(buffer);
+		}
+		return {};
+	}
+};
+
+using AudioFrame2Stream2I = FrameStream2I<AudioFrame2>;
+

src/frame_streams/frame_stream2.hpp

@@ -0,0 +1,47 @@
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <vector>
+
+// Frames often consist of:
+// - seq id // incremental sequential id, gaps in ids can be used to detect loss
+// - or timestamp
+// - data // the frame data
+// eg:
+//struct ExampleFrame {
+	//int64_t seq_id {0};
+	//std::vector<uint8_t> data;
+//};
+
+template<typename FrameType>
+struct FrameStream2I {
+	virtual ~FrameStream2I(void) {}
+
+	// get number of available frames
+	// returns -1 if unknown
+	[[nodiscard]] virtual int32_t size(void) = 0;
+
+	// get next frame
+	// data sharing? -> no, data is copied for each fsr, if concurency supported
+	[[nodiscard]] virtual std::optional<FrameType> pop(void) = 0;
+
+	// returns true if there are readers (or we dont know)
+	virtual bool push(const FrameType& value) = 0;
+};
+
+template<typename FrameType>
+struct FrameStream2SourceI {
+	virtual ~FrameStream2SourceI(void) {}
+	[[nodiscard]] virtual std::shared_ptr<FrameStream2I<FrameType>> subscribe(void) = 0;
+	virtual bool unsubscribe(const std::shared_ptr<FrameStream2I<FrameType>>& sub) = 0;
+};
+
+template<typename FrameType>
+struct FrameStream2SinkI {
+	virtual ~FrameStream2SinkI(void) {}
+	[[nodiscard]] virtual std::shared_ptr<FrameStream2I<FrameType>> subscribe(void) = 0;
+	virtual bool unsubscribe(const std::shared_ptr<FrameStream2I<FrameType>>& sub) = 0;
+};
+

src/frame_streams/sdl/sdl_audio2_frame_stream2.cpp

@@ -0,0 +1,271 @@
+#include "./sdl_audio2_frame_stream2.hpp"
+
+#include <cassert>
+#include <iostream>
+#include <optional>
+
+// "thin" wrapper around sdl audio streams
+// we dont needs to get fance, as they already provide everything we need
+struct SDLAudio2StreamReader : public AudioFrame2Stream2I {
+	std::unique_ptr<SDL_AudioStream, decltype(&SDL_DestroyAudioStream)> _stream;
+
+	uint32_t _sample_rate {48'000};
+	size_t _channels {0};
+
+	// buffer gets reused!
+	std::vector<int16_t> _buffer;
+
+	SDLAudio2StreamReader(void) : _stream(nullptr, nullptr) {}
+	SDLAudio2StreamReader(SDLAudio2StreamReader&& other) :
+		_stream(std::move(other._stream)),
+		_sample_rate(other._sample_rate),
+		_channels(other._channels)
+	{
+		const size_t buffer_size {960*_channels};
+		_buffer.resize(buffer_size);
+	}
+
+	~SDLAudio2StreamReader(void) {
+		if (_stream) {
+			SDL_UnbindAudioStream(_stream.get());
+		}
+	}
+
+	int32_t size(void) override {
+		//assert(_stream);
+		// returns bytes
+		//SDL_GetAudioStreamAvailable(_stream.get());
+		return -1;
+	}
+
+	std::optional<AudioFrame2> pop(void) override {
+		assert(_stream);
+		if (!_stream) {
+			return std::nullopt;
+		}
+
+		const size_t buffer_size {960*_channels};
+		_buffer.resize(buffer_size); // noop?
+
+		const auto read_bytes = SDL_GetAudioStreamData(
+			_stream.get(),
+			_buffer.data(),
+			_buffer.size()*sizeof(int16_t)
+		);
+
+		// no new frame yet, or error
+		if (read_bytes <= 0) {
+			return std::nullopt;
+		}
+
+		return AudioFrame2 {
+			_sample_rate, _channels,
+			Span<int16_t>(_buffer.data(), read_bytes/sizeof(int16_t)),
+		};
+	}
+
+	bool push(const AudioFrame2&) override {
+		// TODO: make universal sdl stream wrapper (combine with SDLAudioOutputDeviceDefaultInstance)
+		assert(false && "read only");
+		return false;
+	}
+};
+
+SDLAudio2InputDevice::SDLAudio2InputDevice(void) : SDLAudio2InputDevice(SDL_AUDIO_DEVICE_DEFAULT_RECORDING) {
+}
+
+SDLAudio2InputDevice::SDLAudio2InputDevice(SDL_AudioDeviceID conf_device_id) : _configured_device_id(conf_device_id) {
+	if (_configured_device_id == 0) {
+		// TODO: proper error handling
+		throw int(1);
+	}
+}
+
+SDLAudio2InputDevice::~SDLAudio2InputDevice(void) {
+	_streams.clear();
+
+	if (_virtual_device_id != 0) {
+		SDL_CloseAudioDevice(_virtual_device_id);
+		_virtual_device_id = 0;
+	}
+}
+
+std::shared_ptr<FrameStream2I<AudioFrame2>> SDLAudio2InputDevice::subscribe(void) {
+	if (_virtual_device_id == 0) {
+		// first stream, open device
+		// this spec is more like a hint to the hardware
+		SDL_AudioSpec spec {
+			SDL_AUDIO_S16,
+			1, // TODO: conf
+			48'000,
+		};
+		_virtual_device_id = SDL_OpenAudioDevice(_configured_device_id, &spec);
+	}
+
+	if (_virtual_device_id == 0) {
+		std::cerr << "SDLAID error: failed opening device " << _configured_device_id << "\n";
+		return nullptr;
+	}
+
+	SDL_AudioSpec spec {
+		SDL_AUDIO_S16, // required, as AudioFrame2 only supports s16
+		1, // TODO: conf
+		48'000,
+	};
+
+	SDL_AudioSpec device_spec {
+		SDL_AUDIO_S16,
+		1, // TODO: conf
+		48'000,
+	};
+	// TODO: error check
+	SDL_GetAudioDeviceFormat(_virtual_device_id, &device_spec, nullptr);
+
+	// error check
+	auto* sdl_stream = SDL_CreateAudioStream(&device_spec, &spec);
+
+	// error check
+	SDL_BindAudioStream(_virtual_device_id, sdl_stream);
+
+	auto new_stream = std::make_shared<SDLAudio2StreamReader>();
+	// TODO: move to ctr
+	new_stream->_stream = {sdl_stream, &SDL_DestroyAudioStream};
+	new_stream->_sample_rate = spec.freq;
+	new_stream->_channels = spec.channels;
+
+	_streams.emplace_back(new_stream);
+
+	return new_stream;
+}
+
+bool SDLAudio2InputDevice::unsubscribe(const std::shared_ptr<FrameStream2I<AudioFrame2>>& sub) {
+	for (auto it = _streams.cbegin(); it != _streams.cend(); it++) {
+		if (*it == sub) {
+			_streams.erase(it);
+			if (_streams.empty()) {
+				// last stream, close
+				// TODO: make sure no shared ptr still exists???
+				SDL_CloseAudioDevice(_virtual_device_id);
+				std::cout << "SDLAID: closing device " << _virtual_device_id << " (" << _configured_device_id << ")\n";
+				_virtual_device_id = 0;
+			}
+			return true;
+		}
+	}
+
+	return false;
+}
+
+// does not need to be visible in the header
+struct SDLAudio2OutputDeviceDefaultInstance : public AudioFrame2Stream2I {
+	std::unique_ptr<SDL_AudioStream, decltype(&SDL_DestroyAudioStream)> _stream;
+
+	uint32_t _last_sample_rate {48'000};
+	size_t _last_channels {0};
+
+	// TODO: audio device
+	SDLAudio2OutputDeviceDefaultInstance(void);
+	SDLAudio2OutputDeviceDefaultInstance(SDLAudio2OutputDeviceDefaultInstance&& other);
+
+	~SDLAudio2OutputDeviceDefaultInstance(void);
+
+	int32_t size(void) override;
+	std::optional<AudioFrame2> pop(void) override;
+	bool push(const AudioFrame2& value) override;
+};
+
+SDLAudio2OutputDeviceDefaultInstance::SDLAudio2OutputDeviceDefaultInstance(void) : _stream(nullptr, nullptr) {
+}
+
+SDLAudio2OutputDeviceDefaultInstance::SDLAudio2OutputDeviceDefaultInstance(SDLAudio2OutputDeviceDefaultInstance&& other) : _stream(std::move(other._stream)) {
+}
+
+SDLAudio2OutputDeviceDefaultInstance::~SDLAudio2OutputDeviceDefaultInstance(void) {
+}
+int32_t SDLAudio2OutputDeviceDefaultInstance::size(void) {
+	return -1;
+}
+
+std::optional<AudioFrame2> SDLAudio2OutputDeviceDefaultInstance::pop(void) {
+	assert(false);
+	// this is an output device, there is no data to pop
+	return std::nullopt;
+}
+
+bool SDLAudio2OutputDeviceDefaultInstance::push(const AudioFrame2& value) {
+	if (!static_cast<bool>(_stream)) {
+		return false;
+	}
+
+	// verify here the fame has the same channel count and sample freq
+	// if something changed, we need to either use a temporary stream, just for conversion, or reopen _stream with the new params
+	// because of data temporality, the second options looks like a better candidate
+	if (
+		value.sample_rate != _last_sample_rate ||
+		value.channels != _last_channels
+	) {
+		const SDL_AudioSpec spec = {
+			static_cast<SDL_AudioFormat>(SDL_AUDIO_S16),
+			static_cast<int>(value.channels),
+			static_cast<int>(value.sample_rate)
+		};
+
+		SDL_SetAudioStreamFormat(_stream.get(), &spec, nullptr);
+
+		std::cerr << "SDLAOD: audio format changed\n";
+	}
+
+	auto data = value.getSpan();
+
+	if (data.size == 0) {
+		std::cerr << "empty audio frame??\n";
+	}
+
+	if (!SDL_PutAudioStreamData(_stream.get(), data.ptr, data.size * sizeof(int16_t))) {
+		std::cerr << "put data error\n";
+		return false; // return true?
+	}
+
+	_last_sample_rate = value.sample_rate;
+	_last_channels = value.channels;
+
+	return true;
+}
+
+
+SDLAudio2OutputDeviceDefaultSink::~SDLAudio2OutputDeviceDefaultSink(void) {
+	// TODO: pause and close device?
+}
+
+std::shared_ptr<FrameStream2I<AudioFrame2>> SDLAudio2OutputDeviceDefaultSink::subscribe(void) {
+	auto new_instance = std::make_shared<SDLAudio2OutputDeviceDefaultInstance>();
+
+	constexpr SDL_AudioSpec spec = { SDL_AUDIO_S16, 2, 48000 };
+
+	new_instance->_stream = {
+		SDL_OpenAudioDeviceStream(SDL_AUDIO_DEVICE_DEFAULT_PLAYBACK, &spec, nullptr, nullptr),
+		&SDL_DestroyAudioStream
+	};
+	new_instance->_last_sample_rate = spec.freq;
+	new_instance->_last_channels = spec.channels;
+
+	if (!static_cast<bool>(new_instance->_stream)) {
+		std::cerr << "SDL open audio device failed!\n";
+		return nullptr;
+	}
+
+	const auto audio_device_id = SDL_GetAudioStreamDevice(new_instance->_stream.get());
+	SDL_ResumeAudioDevice(audio_device_id);
+
+	return new_instance;
+}
+
+bool SDLAudio2OutputDeviceDefaultSink::unsubscribe(const std::shared_ptr<FrameStream2I<AudioFrame2>>& sub) {
+	// TODO: i think we should keep track of them
+	if (!sub) {
+		return false;
+	}
+
+	return true;
+}
+

src/frame_streams/sdl/sdl_audio2_frame_stream2.hpp

@@ -0,0 +1,43 @@
+#pragma once
+
+#include "../frame_stream2.hpp"
+#include "../audio_stream2.hpp"
+
+#include <SDL3/SDL.h>
+
+#include <cstdint>
+#include <vector>
+
+// we dont have to multicast ourself, because sdl streams and virtual devices already do this
+
+// source
+// opens device
+// creates a sdl audio stream for each subscribed reader stream
+struct SDLAudio2InputDevice : public FrameStream2SourceI<AudioFrame2> {
+	// held by instances
+	using sdl_stream_type = std::unique_ptr<SDL_AudioStream, decltype(&SDL_DestroyAudioStream)>;
+
+	SDL_AudioDeviceID _configured_device_id {0};
+	SDL_AudioDeviceID _virtual_device_id {0};
+
+	std::vector<std::shared_ptr<FrameStream2I<AudioFrame2>>> _streams;
+
+	SDLAudio2InputDevice(void);
+	SDLAudio2InputDevice(SDL_AudioDeviceID conf_device_id);
+	~SDLAudio2InputDevice(void);
+
+	std::shared_ptr<FrameStream2I<AudioFrame2>> subscribe(void) override;
+	bool unsubscribe(const std::shared_ptr<FrameStream2I<AudioFrame2>>& sub) override;
+};
+
+// sink
+// constructs entirely new streams, since sdl handles sync and mixing for us (or should)
+struct SDLAudio2OutputDeviceDefaultSink : public FrameStream2SinkI<AudioFrame2> {
+	// TODO: pause device?
+
+	~SDLAudio2OutputDeviceDefaultSink(void);
+
+	std::shared_ptr<FrameStream2I<AudioFrame2>> subscribe(void) override;
+	bool unsubscribe(const std::shared_ptr<FrameStream2I<AudioFrame2>>& sub) override;
+};
+

src/frame_streams/sdl/video.hpp

@@ -0,0 +1,41 @@
+#pragma once
+
+#include <SDL3/SDL.h>
+
+#include <cstdint>
+#include <memory>
+
+// https://youtu.be/71Iw4Q74OaE
+
+inline void nopSurfaceDestructor(SDL_Surface*) {}
+
+// this is very sdl specific
+// but allows us to autoconvert between formats (to a degree)
+struct SDLVideoFrame {
+	// micro seconds (nano is way too much)
+	uint64_t timestampUS {0};
+
+	std::unique_ptr<SDL_Surface, decltype(&SDL_DestroySurface)> surface {nullptr, &SDL_DestroySurface};
+
+	// special non-owning constructor
+	SDLVideoFrame(
+		uint64_t ts,
+		SDL_Surface* surf
+	) {
+		timestampUS = ts;
+		surface = {surf, &nopSurfaceDestructor};
+	}
+	SDLVideoFrame(SDLVideoFrame&& other) = default;
+	// copy
+	SDLVideoFrame(const SDLVideoFrame& other) {
+		timestampUS = other.timestampUS;
+		if (static_cast<bool>(other.surface)) {
+			surface = {
+				SDL_DuplicateSurface(other.surface.get()),
+				&SDL_DestroySurface
+			};
+		}
+	}
+	SDLVideoFrame& operator=(const SDLVideoFrame& other) = delete;
+};
+

src/frame_streams/stream_manager.cpp

@@ -0,0 +1,206 @@
+#include "./stream_manager.hpp"
+
+StreamManager::Connection::Connection(
+	ObjectHandle src_,
+	ObjectHandle sink_,
+	std::unique_ptr<Data>&& data_,
+	std::function<void(Connection&)>&& pump_fn_,
+	std::function<void(Connection&)>&& unsubscribe_fn_,
+	bool on_main_thread_
+) :
+	src(src_),
+	sink(sink_),
+	data(std::move(data_)),
+	pump_fn(std::move(pump_fn_)),
+	unsubscribe_fn(std::move(unsubscribe_fn_)),
+	on_main_thread(on_main_thread_)
+{
+	if (!on_main_thread) {
+		// start thread
+		pump_thread = std::thread([this](void) {
+			while (!stop) {
+				pump_fn(*this);
+				std::this_thread::sleep_for(std::chrono::milliseconds(5));
+			}
+			finished = true;
+		});
+	}
+}
+
+StreamManager::StreamManager(ObjectStore2& os) : _os(os) {
+	_os.subscribe(this, ObjectStore_Event::object_construct);
+	//_os.subscribe(this, ObjectStore_Event::object_update);
+	_os.subscribe(this, ObjectStore_Event::object_destroy);
+}
+
+StreamManager::~StreamManager(void) {
+	// stop all connetions
+	for (const auto& con : _connections) {
+		con->stop = true;
+		if (!con->on_main_thread) {
+			con->pump_thread.join(); // we skip the finished check and wait
+		}
+		con->unsubscribe_fn(*con);
+	}
+}
+
+bool StreamManager::connect(Object src, Object sink, bool threaded) {
+	auto h_src = _os.objectHandle(src);
+	auto h_sink = _os.objectHandle(sink);
+	if (!static_cast<bool>(h_src) || !static_cast<bool>(h_sink)) {
+		// an object does not exist
+		return false;
+	}
+
+	// get src and sink comps
+	if (!h_src.all_of<Components::StreamSource>()) {
+		// src not stream source
+		return false;
+	}
+
+	if (!h_sink.all_of<Components::StreamSink>()) {
+		// sink not stream sink
+		return false;
+	}
+
+	const auto& ssrc = h_src.get<Components::StreamSource>();
+	const auto& ssink = h_sink.get<Components::StreamSink>();
+
+	// compare type
+	if (ssrc.frame_type_name != ssink.frame_type_name) {
+		return false;
+	}
+
+	// always fail in debug mode
+	assert(static_cast<bool>(ssrc.connect_fn));
+	if (!static_cast<bool>(ssrc.connect_fn)) {
+		return false;
+	}
+
+	// use connect fn from src
+	return ssrc.connect_fn(*this, src, sink, threaded);
+}
+
+bool StreamManager::disconnect(Object src, Object sink) {
+	auto res = std::find_if(
+		_connections.cbegin(), _connections.cend(),
+		[&](const auto& a) { return a->src == src && a->sink == sink; }
+	);
+	if (res == _connections.cend()) {
+		// not found
+		return false;
+	}
+
+	// do disconnect
+	(*res)->stop = true;
+
+	return true;
+}
+
+bool StreamManager::disconnectAll(Object o) {
+	bool succ {false};
+	for (const auto& con : _connections) {
+		if (con->src == o || con->sink == o) {
+			con->stop = true;
+			succ = true;
+		}
+	}
+
+	return succ;
+}
+
+// do we need the time delta?
+float StreamManager::tick(float) {
+	// pump all mainthread connections
+	for (auto it = _connections.begin(); it != _connections.end();) {
+		auto& con = **it;
+
+		if (!static_cast<bool>(con.src) || !static_cast<bool>(con.sink)) {
+			// either side disappeard without disconnectAll
+			// TODO: warn/error log
+			con.stop = true;
+		}
+
+		if (con.on_main_thread) {
+			con.pump_fn(con);
+		}
+
+		if (con.stop && (con.finished || con.on_main_thread)) {
+			if (!con.on_main_thread) {
+				assert(con.pump_thread.joinable());
+				con.pump_thread.join();
+			}
+			con.unsubscribe_fn(con);
+			it = _connections.erase(it);
+		} else {
+			it++;
+		}
+	}
+
+	// return min over intervals instead
+	return 2.f; // TODO: 2sec makes mainthread connections unusable
+}
+
+bool StreamManager::onEvent(const ObjectStore::Events::ObjectConstruct& e) {
+	if (!e.e.any_of<Components::StreamSink, Components::StreamSource>()) {
+		return false;
+	}
+
+	// update default targets
+	if (e.e.all_of<Components::TagDefaultTarget>()) {
+		if (e.e.all_of<Components::StreamSource>()) {
+			_default_sources[e.e.get<Components::StreamSource>().frame_type_name] = e.e;
+		} else { // sink
+			_default_sinks[e.e.get<Components::StreamSink>().frame_type_name] = e.e;
+		}
+	}
+
+	// connect to default
+	// only ever do this on new objects
+	if (e.e.all_of<Components::TagConnectToDefault>()) {
+		if (e.e.all_of<Components::StreamSource>()) {
+			auto it_d_sink = _default_sinks.find(e.e.get<Components::StreamSource>().frame_type_name);
+			if (it_d_sink != _default_sinks.cend()) {
+				// TODO: threaded
+				connect(e.e, it_d_sink->second);
+			}
+		} else { // sink
+			auto it_d_src = _default_sources.find(e.e.get<Components::StreamSink>().frame_type_name);
+			if (it_d_src != _default_sources.cend()) {
+				// TODO: threaded
+				connect(it_d_src->second, e.e);
+			}
+		}
+	}
+
+	return false;
+}
+
+bool StreamManager::onEvent(const ObjectStore::Events::ObjectUpdate&) {
+	// what do we do here?
+	return false;
+}
+
+bool StreamManager::onEvent(const ObjectStore::Events::ObjectDestory& e) {
+	// typeless
+	for (auto it = _default_sources.cbegin(); it != _default_sources.cend();) {
+		if (it->second == e.e) {
+			it = _default_sources.erase(it);
+		} else {
+			it++;
+		}
+	}
+	for (auto it = _default_sinks.cbegin(); it != _default_sinks.cend();) {
+		if (it->second == e.e) {
+			it = _default_sinks.erase(it);
+		} else {
+			it++;
+		}
+	}
+
+	// TODO: destroy connections
+	// TODO: auto reconnect default following devices if another default exists
+
+	return false;
+}
+

src/frame_streams/stream_manager.hpp

@@ -0,0 +1,222 @@
+#pragma once
+
+#include <solanaceae/object_store/fwd.hpp>
+#include <solanaceae/object_store/object_store.hpp>
+
+#include <entt/core/type_info.hpp>
+
+#include "./frame_stream2.hpp"
+
+#include <unordered_map>
+#include <vector>
+#include <memory>
+#include <algorithm>
+#include <thread>
+#include <chrono>
+#include <atomic>
+
+// fwd
+class StreamManager;
+
+namespace Components {
+
+	// mark a source or sink as the(a) default
+	struct TagDefaultTarget {};
+
+	// mark a source/sink as to be connected to a default sink/source
+	// of the same type
+	struct TagConnectToDefault {};
+
+	struct StreamSource {
+		std::string name;
+		std::string frame_type_name;
+
+		std::function<bool(StreamManager&, Object, Object, bool)> connect_fn;
+
+		template<typename FrameType>
+		static StreamSource create(const std::string& name);
+	};
+
+	struct StreamSink {
+		std::string name;
+		std::string frame_type_name;
+
+		template<typename FrameType>
+		static StreamSink create(const std::string& name);
+	};
+
+	template<typename FrameType>
+	using FrameStream2Source = std::unique_ptr<FrameStream2SourceI<FrameType>>;
+
+	template<typename FrameType>
+	using FrameStream2Sink = std::unique_ptr<FrameStream2SinkI<FrameType>>;
+
+} // Components
+
+
+class StreamManager : protected ObjectStoreEventI {
+	friend class StreamManagerUI; // TODO: make this go away
+	ObjectStore2& _os;
+
+	struct Connection {
+		ObjectHandle src;
+		ObjectHandle sink;
+
+		struct Data {
+			virtual ~Data(void) {}
+		};
+		std::unique_ptr<Data> data; // stores reader writer type erased
+		std::function<void(Connection&)> pump_fn; // TODO: make it return next interval?
+		std::function<void(Connection&)> unsubscribe_fn;
+
+		bool on_main_thread {true};
+		std::atomic_bool stop {false}; // disconnect
+		std::atomic_bool finished {false}; // disconnect
+
+		// pump thread
+		std::thread pump_thread;
+
+		// frame interval counters and estimates
+		// TODO
+
+		Connection(void) = default;
+		Connection(
+			ObjectHandle src_,
+			ObjectHandle sink_,
+			std::unique_ptr<Data>&& data_,
+			std::function<void(Connection&)>&& pump_fn_,
+			std::function<void(Connection&)>&& unsubscribe_fn_,
+			bool on_main_thread_ = true
+		);
+	};
+	std::vector<std::unique_ptr<Connection>> _connections;
+
+	std::unordered_map<std::string, Object> _default_sources;
+	std::unordered_map<std::string, Object> _default_sinks;
+
+	public:
+		StreamManager(ObjectStore2& os);
+		virtual ~StreamManager(void);
+
+		template<typename FrameType>
+		bool connect(Object src, Object sink, bool threaded = true);
+
+		bool connect(Object src, Object sink, bool threaded = true);
+		bool disconnect(Object src, Object sink);
+		bool disconnectAll(Object o);
+
+		// do we need the time delta?
+		float tick(float);
+
+	protected:
+		bool onEvent(const ObjectStore::Events::ObjectConstruct&) override;
+		bool onEvent(const ObjectStore::Events::ObjectUpdate&) override;
+		bool onEvent(const ObjectStore::Events::ObjectDestory&) override;
+};
+
+// template impls
+
+namespace Components {
+
+	// we require the complete sm type here
+	template<typename FrameType>
+	StreamSource StreamSource::create(const std::string& name) {
+		return StreamSource{
+			name,
+			std::string{entt::type_name<FrameType>::value()},
+			+[](StreamManager& sm, Object src, Object sink, bool threaded) {
+				return sm.connect<FrameType>(src, sink, threaded);
+			},
+		};
+	}
+
+	template<typename FrameType>
+	StreamSink StreamSink::create(const std::string& name) {
+		return StreamSink{
+			name,
+			std::string{entt::type_name<FrameType>::value()},
+		};
+	}
+
+} // Components
+
+template<typename FrameType>
+bool StreamManager::connect(Object src, Object sink, bool threaded) {
+	auto res = std::find_if(
+		_connections.cbegin(), _connections.cend(),
+		[&](const auto& a) { return a->src == src && a->sink == sink; }
+	);
+	if (res != _connections.cend()) {
+		// already exists
+		return false;
+	}
+
+	auto h_src = _os.objectHandle(src);
+	auto h_sink = _os.objectHandle(sink);
+	if (!static_cast<bool>(h_src) || !static_cast<bool>(h_sink)) {
+		// an object does not exist
+		return false;
+	}
+
+	if (!h_src.all_of<Components::FrameStream2Source<FrameType>>()) {
+		// src not stream source
+		return false;
+	}
+
+	if (!h_sink.all_of<Components::FrameStream2Sink<FrameType>>()) {
+		// sink not stream sink
+		return false;
+	}
+
+	auto& src_stream = h_src.get<Components::FrameStream2Source<FrameType>>();
+	auto& sink_stream = h_sink.get<Components::FrameStream2Sink<FrameType>>();
+
+	struct inlineData : public Connection::Data {
+		virtual ~inlineData(void) {}
+		std::shared_ptr<FrameStream2I<FrameType>> reader;
+		std::shared_ptr<FrameStream2I<FrameType>> writer;
+	};
+
+	auto our_data = std::make_unique<inlineData>();
+
+	our_data->reader = src_stream->subscribe();
+	if (!our_data->reader) {
+		return false;
+	}
+	our_data->writer = sink_stream->subscribe();
+	if (!our_data->writer) {
+		return false;
+	}
+
+	_connections.push_back(std::make_unique<Connection>(
+		h_src,
+		h_sink,
+		std::move(our_data),
+		[](Connection& con) -> void {
+			// there might be more stored
+			for (size_t i = 0; i < 10; i++) {
+				auto new_frame_opt = static_cast<inlineData*>(con.data.get())->reader->pop();
+				// TODO: frame interval estimates
+				if (new_frame_opt.has_value()) {
+					static_cast<inlineData*>(con.data.get())->writer->push(new_frame_opt.value());
+				} else {
+					break;
+				}
+			}
+		},
+		[](Connection& con) -> void {
+			auto* src_stream_ptr = con.src.try_get<Components::FrameStream2Source<FrameType>>();
+			if (src_stream_ptr != nullptr) {
+				(*src_stream_ptr)->unsubscribe(static_cast<inlineData*>(con.data.get())->reader);
+			}
+			auto* sink_stream_ptr = con.sink.try_get<Components::FrameStream2Sink<FrameType>>();
+			if (sink_stream_ptr != nullptr) {
+				(*sink_stream_ptr)->unsubscribe(static_cast<inlineData*>(con.data.get())->writer);
+			}
+		},
+		!threaded
+	));
+
+	return true;
+}
+

src/main.cpp

@@ -28,6 +28,8 @@ int main(int argc, char** argv) {
 
 	runSysCheck();
 
+	SDL_SetAppMetadata("tomato", "0.0.0-wip", nullptr);
+
 #ifdef __ANDROID__
 	// change current working dir to internal storage
 	std::filesystem::current_path(SDL_GetAndroidInternalStoragePath());
@@ -35,7 +37,7 @@ int main(int argc, char** argv) {
 
 	// setup hints
 #ifndef __ANDROID__
-	if (SDL_SetHint(SDL_HINT_VIDEO_ALLOW_SCREENSAVER, "1") != SDL_TRUE) {
+	if (!SDL_SetHint(SDL_HINT_VIDEO_ALLOW_SCREENSAVER, "1")) {
 		std::cerr << "Failed to set '" << SDL_HINT_VIDEO_ALLOW_SCREENSAVER << "' to 1\n";
 	}
 #endif

src/main_screen.cpp

@@ -5,6 +5,8 @@
 
 #include <solanaceae/contact/components.hpp>
 
+#include "./frame_streams/sdl/sdl_audio2_frame_stream2.hpp"
+
 #include <imgui/imgui.h>
 
 #include <SDL3/SDL.h>
@@ -19,6 +21,7 @@ MainScreen::MainScreen(SimpleConfigModel&& conf_, SDL_Renderer* renderer_, Theme
 	rmm(cr),
 	msnj{cr, {}, {}},
 	mts(rmm),
+	sm(os),
 	tc(save_path, save_password),
 	tpi(tc.getTox()),
 	ad(tc),
@@ -41,7 +44,9 @@ MainScreen::MainScreen(SimpleConfigModel&& conf_, SDL_Renderer* renderer_, Theme
 	sw(conf),
 	osui(os),
 	tuiu(tc, conf),
-	tdch(tpi)
+	tdch(tpi),
+	smui(os, sm),
+	dvt(os, sm, sdlrtu)
 {
 	tel.subscribeAll(tc);
 
@@ -136,9 +141,46 @@ MainScreen::MainScreen(SimpleConfigModel&& conf_, SDL_Renderer* renderer_, Theme
 	}
 
 	conf.dump();
+
+	if (SDL_InitSubSystem(SDL_INIT_AUDIO)) {
+		// add system audio devices
+		{ // audio in
+			ObjectHandle asrc {os.registry(), os.registry().create()};
+			try {
+				asrc.emplace<Components::FrameStream2Source<AudioFrame2>>(
+					std::make_unique<SDLAudio2InputDevice>()
+				);
+
+				asrc.emplace<Components::StreamSource>(Components::StreamSource::create<AudioFrame2>("SDL Audio Default Recording Device"));
+				asrc.emplace<Components::TagDefaultTarget>();
+
+				os.throwEventConstruct(asrc);
+			} catch (...) {
+				os.registry().destroy(asrc);
+			}
+		}
+		{ // audio out
+			ObjectHandle asink {os.registry(), os.registry().create()};
+			try {
+				asink.emplace<Components::FrameStream2Sink<AudioFrame2>>(
+					std::make_unique<SDLAudio2OutputDeviceDefaultSink>()
+				);
+
+				asink.emplace<Components::StreamSink>(Components::StreamSink::create<AudioFrame2>("SDL Audio Default Playback Device"));
+				asink.emplace<Components::TagDefaultTarget>();
+
+				os.throwEventConstruct(asink);
+			} catch (...) {
+				os.registry().destroy(asink);
+			}
+		}
+	} else {
+		std::cerr << "MS warning: no sdl audio: " << SDL_GetError() << "\n";
+	}
 }
 
 MainScreen::~MainScreen(void) {
+	// TODO: quit sdl audio
 }
 
 bool MainScreen::handleEvent(SDL_Event& e) {
@@ -260,6 +302,8 @@ Screen* MainScreen::render(float time_delta, bool&) {
 	osui.render();
 	tuiu.render(); // render
 	tdch.render(); // render
+	smui.render();
+	const float dvt_interval = dvt.render();
 
 	{ // main window menubar injection
 		if (ImGui::Begin("tomato")) {
@@ -442,6 +486,7 @@ Screen* MainScreen::render(float time_delta, bool&) {
 	if (!_window_hidden && _time_since_event < curr_profile.low_delay_window) {
 		_render_interval = std::min<float>(_render_interval, ctc_interval);
 		_render_interval = std::min<float>(_render_interval, msgtc_interval);
+		_render_interval = std::min<float>(_render_interval, dvt_interval);
 
 		_render_interval = std::clamp(
 			_render_interval,
@@ -452,6 +497,7 @@ Screen* MainScreen::render(float time_delta, bool&) {
 	} else if (!_window_hidden && _time_since_event < curr_profile.mid_delay_window) {
 		_render_interval = std::min<float>(_render_interval, ctc_interval);
 		_render_interval = std::min<float>(_render_interval, msgtc_interval);
+		_render_interval = std::min<float>(_render_interval, dvt_interval);
 
 		_render_interval = std::clamp(
 			_render_interval,
@@ -474,8 +520,16 @@ Screen* MainScreen::render(float time_delta, bool&) {
 }
 
 Screen* MainScreen::tick(float time_delta, bool& quit) {
+	const float sm_interval = sm.tick(time_delta);
+
 	quit = !tc.iterate(time_delta); // compute
 
+#if TOMATO_TOX_AV
+	tav.toxavIterate();
+	// HACK: pow by 1.18 to increase 200 -> ~500
+	const float av_interval = std::pow(tav.toxavIterationInterval(), 1.18)/1000.f;
+#endif
+
 	tcm.iterate(time_delta); // compute
 
 	const float fo_interval = tffom.tick(time_delta);
@@ -505,11 +559,22 @@ Screen* MainScreen::tick(float time_delta, bool& quit) {
 		std::pow(tc.toxIterationInterval(), 1.6f)/1000.f,
 		pm_interval
 	);
+	_min_tick_interval = std::min<float>(
+		_min_tick_interval,
+		sm_interval
+	);
 	_min_tick_interval = std::min<float>(
 		_min_tick_interval,
 		fo_interval
 	);
 
+#if TOMATO_TOX_AV
+	_min_tick_interval = std::min<float>(
+		_min_tick_interval,
+		av_interval
+	);
+#endif
+
 	//std::cout << "MS: min tick interval: " << _min_tick_interval << "\n";
 
 	switch (_compute_perf_mode) {

src/main_screen.hpp

@@ -11,6 +11,7 @@
 #include <solanaceae/plugin/plugin_manager.hpp>
 #include <solanaceae/toxcore/tox_event_logger.hpp>
 #include "./tox_private_impl.hpp"
+#include "./frame_streams/stream_manager.hpp"
 
 #include <solanaceae/tox_contacts/tox_contact_model2.hpp>
 #include <solanaceae/tox_messages/tox_message_manager.hpp>
@@ -33,6 +34,8 @@
 #include "./tox_ui_utils.hpp"
 #include "./tox_dht_cap_histo.hpp"
 #include "./tox_friend_faux_offline_messaging.hpp"
+#include "./stream_manager_ui.hpp"
+#include "./debug_video_tap.hpp"
 
 #if TOMATO_TOX_AV
 #include "./tox_av.hpp"
@@ -58,6 +61,8 @@ struct MainScreen final : public Screen {
 	MessageSerializerNJ msnj;
 	MessageTimeSort mts;
 
+	StreamManager sm;
+
 	ToxEventLogger tel{std::cout};
 	ToxClient tc;
 	ToxPrivateImpl tpi;
@@ -88,6 +93,8 @@ struct MainScreen final : public Screen {
 	ObjectStoreUI osui;
 	ToxUIUtils tuiu;
 	ToxDHTCapHisto tdch;
+	StreamManagerUI smui;
+	DebugVideoTap dvt;
 
 	PluginManager pm; // last, so it gets destroyed first
 

src/sdlrenderer_texture_uploader.cpp

@@ -32,7 +32,7 @@ uint64_t SDLRendererTextureUploader::uploadRGBA(const uint8_t* data, uint32_t wi
 	SDL_UpdateTexture(tex, nullptr, surf->pixels, surf->pitch);
 
 	SDL_BlendMode surf_blend_mode = SDL_BLENDMODE_NONE;
-	if (SDL_GetSurfaceBlendMode(surf, &surf_blend_mode) == 0) {
+	if (SDL_GetSurfaceBlendMode(surf, &surf_blend_mode)) {
 		SDL_SetTextureBlendMode(tex, surf_blend_mode);
 	}
 

src/stream_manager_ui.cpp

@@ -0,0 +1,234 @@
+#include "./stream_manager_ui.hpp"
+
+#include <solanaceae/object_store/object_store.hpp>
+
+#include <imgui/imgui.h>
+
+#include <string>
+
+StreamManagerUI::StreamManagerUI(ObjectStore2& os, StreamManager& sm) : _os(os), _sm(sm) {
+}
+
+void StreamManagerUI::render(void) {
+	{ // main window menubar injection
+		// assumes the window "tomato" was rendered already by cg
+		if (ImGui::Begin("tomato")) {
+			if (ImGui::BeginMenuBar()) {
+				// TODO: drop all menu sep?
+				//ImGui::Separator(); // os already exists (very hacky)
+				if (ImGui::BeginMenu("ObjectStore")) {
+					if (ImGui::MenuItem("Stream Manger", nullptr, _show_window)) {
+						_show_window = !_show_window;
+					}
+					ImGui::EndMenu();
+				}
+				ImGui::EndMenuBar();
+			}
+
+		}
+		ImGui::End();
+	}
+
+
+	if (!_show_window) {
+		return;
+	}
+
+	if (ImGui::Begin("StreamManagerUI", &_show_window)) {
+		// TODO: node canvas
+
+		// by fametype ??
+
+		if (ImGui::CollapsingHeader("Sources", ImGuiTreeNodeFlags_DefaultOpen)) {
+			// list sources
+			if (ImGui::BeginTable("sources_and_sinks", 4, ImGuiTableFlags_SizingFixedFit | ImGuiTableFlags_BordersInnerV)) {
+				ImGui::TableSetupColumn("id");
+				ImGui::TableSetupColumn("name");
+				ImGui::TableSetupColumn("##conn");
+				ImGui::TableSetupColumn("type");
+
+				ImGui::TableHeadersRow();
+
+				for (const auto& [oc, ss] : _os.registry().view<Components::StreamSource>().each()) {
+					//ImGui::Text("src  %d (%s)[%s]", entt::to_integral(entt::to_entity(oc)), ss.name.c_str(), ss.frame_type_name.c_str());
+					ImGui::PushID(entt::to_integral(oc));
+
+					ImGui::TableNextColumn();
+					ImGui::Text("%d", entt::to_integral(entt::to_entity(oc)));
+
+					if (_os.registry().all_of<Components::TagDefaultTarget>(oc)) {
+						ImGui::TableSetBgColor(ImGuiTableBgTarget_RowBg1, ImGui::GetColorU32(ImVec4{0.6f, 0.f, 0.6f, 0.25f}));
+					} else if (_os.registry().all_of<Components::TagConnectToDefault>(oc)) {
+						ImGui::TableSetBgColor(ImGuiTableBgTarget_RowBg1, ImGui::GetColorU32(ImVec4{0.6f, 0.6f, 0.f, 0.25f}));
+					}
+
+					const auto *ssrc = _os.registry().try_get<Components::StreamSource>(oc);
+					ImGui::TableNextColumn();
+					ImGui::TextUnformatted(ssrc!=nullptr?ssrc->name.c_str():"none");
+
+					ImGui::TableNextColumn();
+					if (ImGui::SmallButton("->")) {
+						ImGui::OpenPopup("src_connect");
+					}
+					if (ImGui::BeginPopup("src_connect")) {
+						if (ImGui::BeginMenu("connect to")) {
+							for (const auto& [oc_sink, s_sink] : _os.registry().view<Components::StreamSink>().each()) {
+								if (s_sink.frame_type_name != ss.frame_type_name) {
+									continue;
+								}
+
+								ImGui::PushID(entt::to_integral(oc_sink));
+
+								std::string sink_label {"src "};
+								sink_label += std::to_string(entt::to_integral(entt::to_entity(oc_sink)));
+								sink_label += " (";
+								sink_label += s_sink.name;
+								sink_label += ")[";
+								sink_label += s_sink.frame_type_name;
+								sink_label += "]";
+								if (ImGui::MenuItem(sink_label.c_str())) {
+									_sm.connect(oc, oc_sink);
+								}
+
+								ImGui::PopID();
+							}
+
+							ImGui::EndMenu();
+						}
+						ImGui::EndPopup();
+					}
+
+					ImGui::TableNextColumn();
+					ImGui::TextUnformatted(ssrc!=nullptr?ssrc->frame_type_name.c_str():"???");
+
+					ImGui::PopID();
+				}
+
+				ImGui::EndTable();
+			}
+		} // sources header
+
+		if (ImGui::CollapsingHeader("Sinks", ImGuiTreeNodeFlags_DefaultOpen)) {
+			// list sinks
+			if (ImGui::BeginTable("sources_and_sinks", 4, ImGuiTableFlags_SizingFixedFit | ImGuiTableFlags_BordersInnerV)) {
+				ImGui::TableSetupColumn("id");
+				ImGui::TableSetupColumn("name");
+				ImGui::TableSetupColumn("##conn");
+				ImGui::TableSetupColumn("type");
+
+				ImGui::TableHeadersRow();
+
+				for (const auto& [oc, ss] : _os.registry().view<Components::StreamSink>().each()) {
+					ImGui::PushID(entt::to_integral(oc));
+
+					ImGui::TableNextColumn();
+					ImGui::Text("%d", entt::to_integral(entt::to_entity(oc)));
+
+					if (_os.registry().all_of<Components::TagDefaultTarget>(oc)) {
+						ImGui::TableSetBgColor(ImGuiTableBgTarget_RowBg1, ImGui::GetColorU32(ImVec4{0.6f, 0.f, 0.6f, 0.25f}));
+					} else if (_os.registry().all_of<Components::TagConnectToDefault>(oc)) {
+						ImGui::TableSetBgColor(ImGuiTableBgTarget_RowBg1, ImGui::GetColorU32(ImVec4{0.6f, 0.6f, 0.f, 0.25f}));
+					}
+
+					const auto *ssink = _os.registry().try_get<Components::StreamSink>(oc);
+					ImGui::TableNextColumn();
+					ImGui::TextUnformatted(ssink!=nullptr?ssink->name.c_str():"none");
+
+					ImGui::TableNextColumn();
+					if (ImGui::SmallButton("->")) {
+						ImGui::OpenPopup("sink_connect");
+					}
+					// ImGuiWindowFlags_AlwaysAutoResize | ImGuiWindowFlags_NoTitleBar | ImGuiWindowFlags_NoSavedSettings
+					if (ImGui::BeginPopup("sink_connect")) {
+						if (ImGui::BeginMenu("connect to")) {
+							for (const auto& [oc_src, s_src] : _os.registry().view<Components::StreamSource>().each()) {
+								if (s_src.frame_type_name != ss.frame_type_name) {
+									continue;
+								}
+
+								ImGui::PushID(entt::to_integral(oc_src));
+
+								std::string source_label {"src "};
+								source_label += std::to_string(entt::to_integral(entt::to_entity(oc_src)));
+								source_label += " (";
+								source_label += s_src.name;
+								source_label += ")[";
+								source_label += s_src.frame_type_name;
+								source_label += "]";
+								if (ImGui::MenuItem(source_label.c_str())) {
+									_sm.connect(oc_src, oc);
+								}
+
+								ImGui::PopID();
+							}
+
+							ImGui::EndMenu();
+						}
+						ImGui::EndPopup();
+					}
+
+					ImGui::TableNextColumn();
+					ImGui::TextUnformatted(ssink!=nullptr?ssink->frame_type_name.c_str():"???");
+
+					ImGui::PopID();
+				}
+
+				ImGui::EndTable();
+			}
+		} // sink header
+
+		if (ImGui::CollapsingHeader("Connections", ImGuiTreeNodeFlags_DefaultOpen)) {
+			// list connections
+			if (ImGui::BeginTable("connections", 6, ImGuiTableFlags_SizingFixedFit | ImGuiTableFlags_BordersInnerV)) {
+				ImGui::TableSetupColumn("##id"); // TODO: remove?
+				ImGui::TableSetupColumn("##disco");
+				ImGui::TableSetupColumn("##qdesc");
+				ImGui::TableSetupColumn("from");
+				ImGui::TableSetupColumn("to");
+				ImGui::TableSetupColumn("type");
+
+				ImGui::TableHeadersRow();
+
+				for (size_t i = 0; i < _sm._connections.size(); i++) {
+					const auto& con = _sm._connections[i];
+					//ImGui::Text("con %d->%d", entt::to_integral(entt::to_entity(con->src.entity())), entt::to_integral(entt::to_entity(con->sink.entity())));
+
+					ImGui::PushID(i);
+
+					ImGui::TableNextColumn();
+					ImGui::Text("%zu", i); // do connections have ids?
+
+					ImGui::TableNextColumn();
+					if (ImGui::SmallButton("X")) {
+						con->stop = true;
+					}
+
+					ImGui::TableNextColumn();
+					ImGui::Text("%d->%d", entt::to_integral(entt::to_entity(con->src.entity())), entt::to_integral(entt::to_entity(con->sink.entity())));
+
+					const auto *ssrc = con->src.try_get<Components::StreamSource>();
+					ImGui::TableNextColumn();
+					ImGui::TextUnformatted(ssrc!=nullptr?ssrc->name.c_str():"none");
+
+					const auto *ssink = con->sink.try_get<Components::StreamSink>();
+					ImGui::TableNextColumn();
+					ImGui::TextUnformatted(ssink!=nullptr?ssink->name.c_str():"none");
+
+					ImGui::TableNextColumn();
+					ImGui::TextUnformatted(
+						(ssrc!=nullptr)?
+							ssrc->frame_type_name.c_str():
+							(ssink!=nullptr)?
+								ssink->frame_type_name.c_str()
+								:"???"
+					);
+
+					ImGui::PopID();
+				}
+				ImGui::EndTable();
+			}
+		} // con header
+	}
+	ImGui::End();
+}
+

src/stream_manager_ui.hpp

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <solanaceae/object_store/fwd.hpp>
+#include "./frame_streams/stream_manager.hpp"
+
+class StreamManagerUI {
+	ObjectStore2& _os;
+	StreamManager& _sm;
+
+	bool _show_window {false};
+
+	public:
+		StreamManagerUI(ObjectStore2& os, StreamManager& sm);
+
+		void render(void);
+};
+

src/tox_av.cpp

@@ -2,14 +2,85 @@
 
 #include <cassert>
 
+#include <cstdint>
+#include <iostream>
+
 // https://almogfx.bandcamp.com/track/crushed-w-cassade
 
+struct ToxAVFriendCallState final {
+	const uint32_t state {TOXAV_FRIEND_CALL_STATE_NONE};
+
+	[[nodiscard]] bool is_error(void) const { return state & TOXAV_FRIEND_CALL_STATE_ERROR; }
+	[[nodiscard]] bool is_finished(void) const { return state & TOXAV_FRIEND_CALL_STATE_FINISHED; }
+	[[nodiscard]] bool is_sending_a(void) const { return state & TOXAV_FRIEND_CALL_STATE_SENDING_A; }
+	[[nodiscard]] bool is_sending_v(void) const { return state & TOXAV_FRIEND_CALL_STATE_SENDING_V; }
+	[[nodiscard]] bool is_accepting_a(void) const { return state & TOXAV_FRIEND_CALL_STATE_ACCEPTING_A; }
+	[[nodiscard]] bool is_accepting_v(void) const { return state & TOXAV_FRIEND_CALL_STATE_ACCEPTING_V; }
+};
+
 ToxAV::ToxAV(Tox* tox) : _tox(tox) {
 	Toxav_Err_New err_new {TOXAV_ERR_NEW_OK};
 	_tox_av = toxav_new(_tox, &err_new);
 	// TODO: throw
 	assert(err_new == TOXAV_ERR_NEW_OK);
+
+	toxav_callback_call(
+		_tox_av,
+		+[](ToxAV*, uint32_t friend_number, bool audio_enabled, bool video_enabled, void *user_data) {
+			assert(user_data != nullptr);
+			static_cast<ToxAV*>(user_data)->cb_call(friend_number, audio_enabled, video_enabled);
+		},
+		this
+	);
+	toxav_callback_call_state(
+		_tox_av,
+		+[](ToxAV*, uint32_t friend_number, uint32_t state, void *user_data) {
+			assert(user_data != nullptr);
+			static_cast<ToxAV*>(user_data)->cb_call_state(friend_number, state);
+		},
+		this
+	);
+	toxav_callback_audio_bit_rate(
+		_tox_av,
+		+[](ToxAV*, uint32_t friend_number, uint32_t audio_bit_rate, void *user_data) {
+			assert(user_data != nullptr);
+			static_cast<ToxAV*>(user_data)->cb_audio_bit_rate(friend_number, audio_bit_rate);
+		},
+		this
+	);
+	toxav_callback_video_bit_rate(
+		_tox_av,
+		+[](ToxAV*, uint32_t friend_number, uint32_t video_bit_rate, void *user_data) {
+			assert(user_data != nullptr);
+			static_cast<ToxAV*>(user_data)->cb_video_bit_rate(friend_number, video_bit_rate);
+		},
+		this
+	);
+	toxav_callback_audio_receive_frame(
+		_tox_av,
+		+[](ToxAV*, uint32_t friend_number, const int16_t pcm[], size_t sample_count, uint8_t channels, uint32_t sampling_rate, void *user_data) {
+			assert(user_data != nullptr);
+			static_cast<ToxAV*>(user_data)->cb_audio_receive_frame(friend_number, pcm, sample_count, channels, sampling_rate);
+		},
+		this
+	);
+	toxav_callback_video_receive_frame(
+		_tox_av,
+		+[](ToxAV*, uint32_t friend_number,
+			uint16_t width, uint16_t height,
+			const uint8_t y[/*! max(width, abs(ystride)) * height */],
+			const uint8_t u[/*! max(width/2, abs(ustride)) * (height/2) */],
+			const uint8_t v[/*! max(width/2, abs(vstride)) * (height/2) */],
+			int32_t ystride, int32_t ustride, int32_t vstride,
+			void *user_data
+		) {
+			assert(user_data != nullptr);
+			static_cast<ToxAV*>(user_data)->cb_video_receive_frame(friend_number, width, height, y, u, v, ystride, ustride, vstride);
+		},
+		this
+	);
 }
+
 ToxAV::~ToxAV(void) {
 	toxav_kill(_tox_av);
 }
@@ -80,3 +151,101 @@ Toxav_Err_Bit_Rate_Set ToxAV::toxavVideoSetBitRate(uint32_t friend_number, uint3
 	return err;
 }
 
+void ToxAV::cb_call(uint32_t friend_number, bool audio_enabled, bool video_enabled) {
+	std::cerr << "TOXAV: receiving call f:" << friend_number << " a:" << audio_enabled << " v:" << video_enabled << "\n";
+	//Toxav_Err_Answer err_answer { TOXAV_ERR_ANSWER_OK };
+	//toxav_answer(_tox_av, friend_number, 0, 0, &err_answer);
+	//if (err_answer != TOXAV_ERR_ANSWER_OK) {
+	//    std::cerr << "!!!!!!!! answer failed " << err_answer << "\n";
+	//}
+
+	dispatch(
+		ToxAV_Event::friend_call,
+		Events::FriendCall{
+			friend_number,
+			audio_enabled,
+			video_enabled,
+		}
+	);
+}
+
+void ToxAV::cb_call_state(uint32_t friend_number, uint32_t state) {
+	//ToxAVFriendCallState w_state{state};
+
+	//w_state.is_error();
+
+	std::cerr << "TOXAV: call state f:" << friend_number << " s:" << state << "\n";
+
+	dispatch(
+		ToxAV_Event::friend_call_state,
+		Events::FriendCallState{
+			friend_number,
+			state,
+		}
+	);
+}
+
+void ToxAV::cb_audio_bit_rate(uint32_t friend_number, uint32_t audio_bit_rate) {
+	std::cerr << "TOXAV: audio bitrate f:" << friend_number << " abr:" << audio_bit_rate << "\n";
+
+	dispatch(
+		ToxAV_Event::friend_audio_bitrate,
+		Events::FriendAudioBitrate{
+			friend_number,
+			audio_bit_rate,
+		}
+	);
+}
+
+void ToxAV::cb_video_bit_rate(uint32_t friend_number, uint32_t video_bit_rate) {
+	std::cerr << "TOXAV: video bitrate f:" << friend_number << " vbr:" << video_bit_rate << "\n";
+
+	dispatch(
+		ToxAV_Event::friend_video_bitrate,
+		Events::FriendVideoBitrate{
+			friend_number,
+			video_bit_rate,
+		}
+	);
+}
+
+void ToxAV::cb_audio_receive_frame(uint32_t friend_number, const int16_t pcm[], size_t sample_count, uint8_t channels, uint32_t sampling_rate) {
+	//std::cerr << "TOXAV: audio frame f:" <<  friend_number << " sc:" << sample_count << " ch:" << (int)channels << " sr:" << sampling_rate << "\n";
+
+	dispatch(
+		ToxAV_Event::friend_audio_frame,
+		Events::FriendAudioFrame{
+			friend_number,
+			Span<int16_t>(pcm, sample_count*channels), // TODO: is sample count *ch or /ch?
+			channels,
+			sampling_rate,
+		}
+	);
+}
+
+void ToxAV::cb_video_receive_frame(
+	uint32_t friend_number,
+	uint16_t width, uint16_t height,
+	const uint8_t y[/*! max(width, abs(ystride)) * height */],
+	const uint8_t u[/*! max(width/2, abs(ustride)) * (height/2) */],
+	const uint8_t v[/*! max(width/2, abs(vstride)) * (height/2) */],
+	int32_t ystride, int32_t ustride, int32_t vstride
+) {
+	//std::cerr << "TOXAV: video frame f:" <<  friend_number << " w:" << width << " h:" << height << "\n";
+
+	dispatch(
+		ToxAV_Event::friend_video_frame,
+		Events::FriendVideoFrame{
+			friend_number,
+			width,
+			height,
+			Span<uint8_t>(y, std::max<int64_t>(width, std::abs(ystride)) * height),
+			Span<uint8_t>(u, std::max<int64_t>(width/2, std::abs(ustride)) * (height/2)),
+			Span<uint8_t>(v, std::max<int64_t>(width/2, std::abs(vstride)) * (height/2)),
+			ystride,
+			ustride,
+			vstride,
+		}
+	);
+}
+

src/tox_av.hpp

@@ -1,15 +1,98 @@
 #pragma once
 
+#include <solanaceae/util/span.hpp>
+#include <solanaceae/util/event_provider.hpp>
+
 #include <tox/toxav.h>
 
-struct ToxAV {
+namespace /*toxav*/ Events {
+
+	struct FriendCall {
+		uint32_t friend_number;
+		bool audio_enabled;
+		bool video_enabled;
+	};
+
+	struct FriendCallState {
+		uint32_t friend_number;
+		uint32_t state;
+	};
+
+	struct FriendAudioBitrate {
+		uint32_t friend_number;
+		uint32_t audio_bit_rate;
+	};
+
+	struct FriendVideoBitrate {
+		uint32_t friend_number;
+		uint32_t video_bit_rate;
+	};
+
+	struct FriendAudioFrame {
+		uint32_t friend_number;
+
+		Span<int16_t> pcm;
+		//size_t sample_count;
+		uint8_t channels;
+		uint32_t sampling_rate;
+	};
+
+	struct FriendVideoFrame {
+		uint32_t friend_number;
+
+		uint16_t width;
+		uint16_t height;
+		//const uint8_t y[[>! max(width, abs(ystride)) * height <]];
+		//const uint8_t u[[>! max(width/2, abs(ustride)) * (height/2) <]];
+		//const uint8_t v[[>! max(width/2, abs(vstride)) * (height/2) <]];
+		// mdspan would be nice here
+		// bc of the stride, span might be larger than the actual data it contains
+		Span<uint8_t> y;
+		Span<uint8_t> u;
+		Span<uint8_t> v;
+		int32_t ystride;
+		int32_t ustride;
+		int32_t vstride;
+	};
+
+} // Event
+
+enum class ToxAV_Event : uint32_t {
+	friend_call,
+	friend_call_state,
+	friend_audio_bitrate,
+	friend_video_bitrate,
+	friend_audio_frame,
+	friend_video_frame,
+
+	MAX
+};
+
+struct ToxAVEventI {
+	using enumType = ToxAV_Event;
+
+	virtual ~ToxAVEventI(void) {}
+
+	virtual bool onEvent(const Events::FriendCall&) { return false; }
+	virtual bool onEvent(const Events::FriendCallState&) { return false; }
+	virtual bool onEvent(const Events::FriendAudioBitrate&) { return false; }
+	virtual bool onEvent(const Events::FriendVideoBitrate&) { return false; }
+	virtual bool onEvent(const Events::FriendAudioFrame&) { return false; }
+	virtual bool onEvent(const Events::FriendVideoFrame&) { return false; }
+};
+using ToxAVEventProviderI = EventProviderI<ToxAVEventI>;
+
+struct ToxAV : public ToxAVEventProviderI{
 	Tox* _tox = nullptr;
 	ToxAV* _tox_av = nullptr;
 
+	static constexpr const char* version {"0"};
+
 	ToxAV(Tox* tox);
 	virtual ~ToxAV(void);
 
 	// interface
+	// if iterate is called on a different thread, it will fire events there
 	uint32_t toxavIterationInterval(void) const;
 	void toxavIterate(void);
 
@@ -33,5 +116,21 @@ struct ToxAV {
 //int32_t toxav_groupchat_disable_av(Tox *tox, uint32_t groupnumber);
 //bool toxav_groupchat_av_enabled(Tox *tox, uint32_t groupnumber);
 
+
+
+	// toxav callbacks
+	void cb_call(uint32_t friend_number, bool audio_enabled, bool video_enabled);
+	void cb_call_state(uint32_t friend_number, uint32_t state);
+	void cb_audio_bit_rate(uint32_t friend_number, uint32_t audio_bit_rate);
+	void cb_video_bit_rate(uint32_t friend_number, uint32_t video_bit_rate);
+	void cb_audio_receive_frame(uint32_t friend_number, const int16_t pcm[], size_t sample_count, uint8_t channels, uint32_t sampling_rate);
+	void cb_video_receive_frame(
+		uint32_t friend_number,
+		uint16_t width, uint16_t height,
+		const uint8_t y[/*! max(width, abs(ystride)) * height */],
+		const uint8_t u[/*! max(width/2, abs(ustride)) * (height/2) */],
+		const uint8_t v[/*! max(width/2, abs(vstride)) * (height/2) */],
+		int32_t ystride, int32_t ustride, int32_t vstride
+	);
 };