add audio incoming source

src/CMakeLists.txt

@@ -108,6 +108,7 @@ target_sources(tomato PUBLIC
 	./frame_streams/stream_manager.hpp
 	./frame_streams/stream_manager.cpp
 	./frame_streams/locked_frame_stream.hpp
+	./frame_streams/multi_source.hpp
 
 	./frame_streams/voip_model.hpp
 

src/frame_streams/locked_frame_stream.hpp

@@ -34,6 +34,10 @@ struct LockedFrameStream2 : public FrameStream2I<FrameType> {
 	bool push(const FrameType& value) {
 		std::lock_guard lg{_lock};
 
+		if (_frames.size() > 1024) {
+			return false; // hard limit
+		}
+
 		_frames.push_back(value);
 
 		return true;

src/frame_streams/multi_source.hpp

@@ -0,0 +1,62 @@
+#pragma once
+
+#include "./locked_frame_stream.hpp"
+
+#include <cassert>
+
+// implements a stream that pushes to all sub streams
+template<typename FrameType, typename SubStreamType = LockedFrameStream2<FrameType>>
+struct FrameStream2MultiSource : public FrameStream2SourceI<FrameType>, public FrameStream2I<FrameType> {
+	using sub_stream_type_t = SubStreamType;
+
+	// pointer stability
+	std::vector<std::shared_ptr<SubStreamType>> _sub_streams;
+	std::mutex _sub_stream_lock; // accessing the _sub_streams array needs to be exclusive
+	// a simple lock here is ok, since this tends to be a rare operation,
+	// except for the push, which is always on the same thread
+
+	virtual ~FrameStream2MultiSource(void) {}
+
+	// TODO: forward args instead
+	std::shared_ptr<FrameStream2I<FrameType>> subscribe(void) override {
+		std::lock_guard lg{_sub_stream_lock};
+		return _sub_streams.emplace_back(std::make_unique<SubStreamType>());
+	}
+
+	bool unsubscribe(const std::shared_ptr<FrameStream2I<FrameType>>& sub) override {
+		std::lock_guard lg{_sub_stream_lock};
+		for (auto it = _sub_streams.begin(); it != _sub_streams.end(); it++) {
+			if (*it == sub) {
+				_sub_streams.erase(it);
+				return true;
+			}
+		}
+		return false; // ?
+	}
+
+	// stream interface
+
+	int32_t size(void) override {
+		// TODO: return something sensible?
+		return -1;
+	}
+
+	std::optional<FrameType> pop(void) override {
+		// nope
+		assert(false && "this logic is very frame type specific, provide an impl");
+		return std::nullopt;
+	}
+
+	// returns true if there are readers
+	bool push(const FrameType& value) override {
+		std::lock_guard lg{_sub_stream_lock};
+		bool have_readers{false};
+		for (auto& it : _sub_streams) {
+			[[maybe_unused]] auto _ = it->push(value);
+			have_readers = true; // even if queue full, we still continue believing in them
+			// maybe consider push return value?
+		}
+		return have_readers;
+	}
+};
+

src/tox_av_voip_model.cpp

@@ -6,6 +6,7 @@
 #include "./frame_streams/stream_manager.hpp"
 #include "./frame_streams/audio_stream2.hpp"
 #include "./frame_streams/locked_frame_stream.hpp"
+#include "./frame_streams/multi_source.hpp"
 
 #include <iostream>
 
@@ -31,7 +32,7 @@ namespace Components {
 // TODO: make proper adapter
 struct AudioStreamReFramer {
 	FrameStream2I<AudioFrame2>* _stream {nullptr};
-	uint32_t frame_length_ms {10};
+	uint32_t frame_length_ms {20};
 
 	// dequeue?
 	std::vector<int16_t> buffer;
@@ -59,7 +60,7 @@ struct AudioStreamReFramer {
 			}
 
 
-			std::cout << "new incoming frame is " << new_value.getSpan().size/new_value.channels*1000/new_value.sample_rate << "ms\n";
+			//std::cout << "new incoming frame is " << new_value.getSpan().size/new_value.channels*1000/new_value.sample_rate << "ms\n";
 
 			auto new_span = new_value.getSpan();
 
@@ -380,9 +381,36 @@ bool ToxAVVoIPModel::onEvent(const Events::FriendCallState& e) {
 				// video
 
 				// add/update sources
+				auto& stream_source = o.get_or_emplace<Components::VoIP::StreamSources>().streams;
+
 				// audio
-				// video
+				if (s.is_sending_a() && !o.all_of<Components::ToxAVAudioSource>()) {
+					ObjectHandle incoming_audio {_os.registry(), _os.registry().create()};
+
+					auto new_asrc = std::make_unique<FrameStream2MultiSource<AudioFrame2>>();
+					incoming_audio.emplace<FrameStream2MultiSource<AudioFrame2>*>(new_asrc.get());
+					incoming_audio.emplace<Components::FrameStream2Source<AudioFrame2>>(std::move(new_asrc));
+					incoming_audio.emplace<Components::StreamSource>(Components::StreamSource::create<AudioFrame2>("ToxAV Friend Call Incoming Audio"));
+
+					std::cout << "new incoming audio\n";
+					if (
+						const auto* defaults = o.try_get<Components::VoIP::DefaultConfig>();
+						defaults != nullptr && defaults->incoming_audio
+					) {
+						incoming_audio.emplace<Components::TagConnectToDefault>(); // depends on what was specified in enter()
+						std::cout << "with default\n";
+					}
+
+					stream_source.push_back(incoming_audio);
+					o.emplace<Components::ToxAVAudioSource>(incoming_audio);
+					// TODO: tie session to stream
 
+					_os.throwEventConstruct(incoming_audio);
+				} else if (!s.is_sending_a() && o.all_of<Components::ToxAVAudioSource>()) {
+					// remove asrc?
+				}
+
+				// video
 			}
 		}
 	}
@@ -398,8 +426,56 @@ bool ToxAVVoIPModel::onEvent(const Events::FriendVideoBitrate&) {
 	return false;
 }
 
-bool ToxAVVoIPModel::onEvent(const Events::FriendAudioFrame&) {
-	return false;
+bool ToxAVVoIPModel::onEvent(const Events::FriendAudioFrame& e) {
+	//auto& call = _calls[e.friend_number];
+
+	// get session?
+	// get asrc (directly instead?) this is pretty hot
+	const auto session_contact = _tcm.getContactFriend(e.friend_number);
+	if (!_cr.valid(session_contact)) {
+		return false;
+	}
+
+	// jesus this is bad
+	ObjectHandle asrc;
+	for (const auto& [ov, voipmodel] : _os.registry().view<VoIPModelI*>().each()) {
+		if (voipmodel != this) {
+			continue;
+		}
+
+		auto o = _os.objectHandle(ov);
+
+		if (!o.all_of<Components::VoIP::SessionContact>()) {
+			continue;
+		}
+		if (session_contact != o.get<Components::VoIP::SessionContact>().c) {
+			continue;
+		}
+
+		if (!o.all_of<Components::ToxAVAudioSource>()) {
+			continue;
+		}
+		asrc = o.get<Components::ToxAVAudioSource>().o;
+		break;
+	}
+
+	if (!static_cast<bool>(asrc)) {
+		// missing src to put frame into ??
+		return false;
+	}
+
+	//assert(call.incoming_asrc.all_of<AudioFrameStream2MultiSource*>());
+	assert(asrc.all_of<FrameStream2MultiSource<AudioFrame2>*>());
+	//assert(call.incoming_asrc.all_of<Components::FrameStream2Source<AudioFrame>>());
+	assert(asrc.all_of<Components::FrameStream2Source<AudioFrame2>>());
+
+	asrc.get<FrameStream2MultiSource<AudioFrame2>*>()->push(AudioFrame2{
+		e.sampling_rate,
+		e.channels,
+		std::vector<int16_t>(e.pcm.begin(), e.pcm.end()) // copy
+	});
+
+	return true;
 }
 
 bool ToxAVVoIPModel::onEvent(const Events::FriendVideoFrame&) {