Add muxer, broadcast and tests

CMakeLists.txt

@@ -171,6 +171,10 @@ add_library(beyond-protocol STATIC
 	src/protocol/websocket.cpp
 	src/base64.cpp
 	src/protocol.cpp
+	src/streams.cpp
+	src/channelSet.cpp
+	src/muxer.cpp
+	src/broadcaster.cpp
 )
 
 target_include_directories(beyond-protocol PUBLIC

include/ftl/protocol/broadcaster.hpp

@@ -16,8 +16,8 @@ class Broadcast : public Stream {
 	explicit Broadcast();
 	virtual ~Broadcast();
 
-	void add(Stream *);
-	void remove(Stream *);
+	void add(const std::shared_ptr<Stream> &);
+	void remove(const std::shared_ptr<Stream> &);
 	void clear();
 
 	bool post(const ftl::protocol::StreamPacket &, const ftl::protocol::Packet &) override;
@@ -28,10 +28,10 @@ class Broadcast : public Stream {
 
 	void reset() override;
 
-	const std::list<Stream*> &streams() const { return streams_; }
+	const std::list<std::shared_ptr<Stream>> &streams() const { return streams_; }
 
 	private:
-	std::list<Stream*> streams_;
+	std::list<std::shared_ptr<Stream>> streams_;
 	std::list<ftl::Handle> handles_;
 	//StreamCallback cb_;
 	SHARED_MUTEX mutex_;

include/ftl/protocol/muxer.hpp

@@ -3,6 +3,7 @@
 #include <ftl/protocol/streams.hpp>
 
 #include <map>
+#include <list>
 
 namespace ftl {
 namespace protocol {
@@ -20,8 +21,8 @@ class Muxer : public Stream {
 	explicit Muxer();
 	virtual ~Muxer();
 
-	void add(Stream *, size_t fsid=0, const std::function<int()> &cb=nullptr);
-	void remove(Stream *);
+	void add(const std::shared_ptr<Stream> &, size_t fsid=0, const std::function<int()> &cb=nullptr);
+	void remove(const std::shared_ptr<Stream> &);
 
 	//bool onPacket(const StreamCallback &) override;
 
@@ -33,11 +34,11 @@ class Muxer : public Stream {
 
 	void reset() override;
 
-	ftl::protocol::Stream *originStream(size_t fsid, int fid);
+	std::shared_ptr<Stream> originStream(size_t fsid, int fid);
 
 	private:
 	struct StreamEntry {
-		Stream *stream;
+		std::shared_ptr<Stream> stream;
 		std::unordered_map<int, std::vector<int>> maps;
 		uint32_t original_fsid = 0;
 		ftl::Handle handle;

include/ftl/protocol/packet.hpp

@@ -67,7 +67,7 @@ struct StreamPacketV4 {
 	int64_t timestamp;
 	uint8_t streamID;  		// Source number [or v4 frameset id]
 	uint8_t frame_number;	// v4+ First frame number (packet may include multiple frames)
-	ftl::codecs::Channel channel;		// Actual channel of this current set of packets
+	ftl::protocol::Channel channel;		// Actual channel of this current set of packets
 
 	inline int frameNumber() const { return (version >= 4) ? frame_number : streamID; }
 	inline size_t frameSetID() const { return (version >= 4) ? streamID : 0; }
@@ -92,7 +92,7 @@ struct StreamPacket {
 	int64_t timestamp;
 	uint8_t streamID;  		// Source number [or v4 frameset id]
 	uint8_t frame_number;	// v4+ First frame number (packet may include multiple frames)
-	ftl::codecs::Channel channel;		// Actual channel of this current set of packets
+	ftl::protocol::Channel channel;		// Actual channel of this current set of packets
 	uint8_t flags=0;
 
 	inline int frameNumber() const { return (version >= 4) ? frame_number : streamID; }

src/broadcaster.cpp

+#include <ftl/protocol/broadcaster.hpp>
+
+using ftl::protocol::Broadcast;
+using ftl::protocol::StreamPacket;
+using ftl::protocol::Packet;
+
+Broadcast::Broadcast() {
+
+}
+
+Broadcast::~Broadcast() {
+
+}
+
+void Broadcast::add(const std::shared_ptr<Stream> &s) {
+	UNIQUE_LOCK(mutex_,lk);
+
+	streams_.push_back(s);
+	handles_.push_back(std::move(s->onPacket([this,s](const StreamPacket &spkt, const Packet &pkt) {
+		//LOG(INFO) << "BCAST Request: " << (int)spkt.streamID << " " << (int)spkt.channel << " " << spkt.timestamp;
+		SHARED_LOCK(mutex_, lk);
+		if (spkt.frameSetID() < 255) available(spkt.frameSetID()) += spkt.channel;
+		cb_.trigger(spkt, pkt);
+		if (spkt.streamID < 255) s->select(spkt.streamID, selected(spkt.streamID));
+		return true;
+	})));
+}
+
+void Broadcast::remove(const std::shared_ptr<Stream> &s) {
+	UNIQUE_LOCK(mutex_,lk);
+	// TODO: Find and remove handle also
+	streams_.remove(s);
+}
+
+void Broadcast::clear() {
+	UNIQUE_LOCK(mutex_,lk);
+	handles_.clear();
+	streams_.clear();
+}
+
+bool Broadcast::post(const StreamPacket &spkt, const Packet &pkt) {
+	SHARED_LOCK(mutex_, lk);
+	if (spkt.frameSetID() < 255) available(spkt.frameSetID()) += spkt.channel;
+
+	bool status = true;
+	for (auto s : streams_) {
+		//s->select(spkt.frameSetID(), selected(spkt.frameSetID()));
+		status = status && s->post(spkt, pkt);
+	}
+	return status;
+}
+
+bool Broadcast::begin() {
+	bool r = true;
+	for (auto &s : streams_) {
+		r = r && s->begin();
+	}
+	return r;
+}
+
+bool Broadcast::end() {
+	bool r = true;
+	for (auto &s : streams_) {
+		r = r && s->end();
+	}
+	return r;
+}
+
+bool Broadcast::active() {
+	if (streams_.size() == 0) return false;
+	bool r = true;
+	for (auto &s : streams_) {
+		r = r && s->active();
+	}
+	return r;
+}
+
+void Broadcast::reset() {
+	for (auto &s : streams_) {
+		s->reset();
+	}
+}

src/channelSet.cpp

+#include <ftl/protocol/channelSet.hpp>
+
+using ftl::protocol::ChannelSet;
+
+ChannelSet operator&(const ChannelSet &a, const ChannelSet &b) {
+	ChannelSet result;
+	for (auto &i : a) {
+		if (b.find(i) != b.end()) result.insert(i);
+	}
+	return result;
+}
+
+ChannelSet operator-(const ChannelSet &a, const ChannelSet &b) {
+	ChannelSet result;
+	for (auto &i : a) {
+		if (b.find(i) == b.end()) result.insert(i);
+	}
+	return result;
+}
+
+bool operator!=(const ChannelSet &a, const ChannelSet &b) {
+	if (a.size() != b.size()) return true;
+	for (auto &i : a) {
+		if (b.count(i) == 0) return true;
+	}
+	return false;
+}

src/muxer.cpp

+#include <ftl/protocol/muxer.hpp>
+#include <ftl/lib/loguru.hpp>
+
+using ftl::protocol::Muxer;
+using ftl::protocol::Stream;
+using ftl::protocol::StreamPacket;
+
+Muxer::Muxer() : nid_{0} {
+	//value("paused", false);
+	//_forward("paused");
+}
+
+Muxer::~Muxer() {
+	UNIQUE_LOCK(mutex_,lk);
+	for (auto &se : streams_) {
+		se.handle.cancel();
+	}
+}
+
+void Muxer::_forward(const std::string &name) {
+	/*on(name, [this,name]() {
+		auto val = getConfig()[name];
+		UNIQUE_LOCK(mutex_,lk);
+		for (auto &se : streams_) {
+			se.stream->set(name, val);
+		}
+	});*/
+}
+
+
+void Muxer::add(const std::shared_ptr<Stream> &s, size_t fsid, const std::function<int()> &cb) {
+	UNIQUE_LOCK(mutex_,lk);
+	if (fsid < 0u || fsid >= ftl::protocol::kMaxStreams) return;
+
+	auto &se = streams_.emplace_back();
+	//int i = streams_.size()-1;
+	se.stream = s;
+	se.ids.push_back(fsid);
+	Muxer::StreamEntry *ptr = &se;
+
+	se.handle = std::move(s->onPacket([this,s,ptr,cb](const StreamPacket &spkt, const Packet &pkt) {
+		//TODO: Allow input streams to have other streamIDs
+		// Same fsid means same streamIDs map together in the end
+
+		/*ftl::stream::Muxer::StreamEntry *ptr = nullptr;
+		{
+			SHARED_LOCK(mutex_,lk);
+			ptr = &streams_[i];
+		}*/
+
+		if (!cb && spkt.streamID > 0) {
+			LOG(WARNING) << "Multiple framesets in stream";
+			return true;
+		}
+
+		if (ptr->ids.size() <= spkt.streamID) {
+			UNIQUE_LOCK(mutex_,lk);
+			if (ptr->ids.size() <= spkt.streamID) {
+				ptr->ids.resize(spkt.streamID + 1);
+				ptr->ids[spkt.streamID] = cb();
+			}
+		}
+
+		int fsid;
+		{
+			SHARED_LOCK(mutex_, lk);
+			fsid = ptr->ids[spkt.streamID];
+		}
+
+		StreamPacket spkt2 = spkt;
+		ptr->original_fsid = spkt.streamID;  // FIXME: Multiple originals needed
+		spkt2.streamID = fsid;
+
+		if (spkt2.frame_number < 255) {
+			int id = _lookup(fsid, ptr, spkt.frame_number, pkt.frame_count);
+			spkt2.frame_number = id;
+		}
+
+		_notify(spkt2, pkt);
+		s->select(spkt.streamID, selected(fsid), true);
+		return true;
+	}));
+}
+
+void Muxer::remove(const std::shared_ptr<Stream> &s) {
+	UNIQUE_LOCK(mutex_,lk);
+	for (auto i = streams_.begin(); i != streams_.end(); ++i) {
+		if (i->stream == s) {
+			i->handle.cancel();
+			auto *se = &(*i);
+
+			for (size_t j=0; j<kMaxStreams; ++j) {
+				for (auto &k : revmap_[j]) {
+					if (k.first == se) {
+						k.first = nullptr;
+					}
+				}
+			}
+
+			streams_.erase(i);
+			return;
+		}
+	}
+}
+
+std::shared_ptr<Stream> Muxer::originStream(size_t fsid, int fid) {
+	if (fsid < ftl::protocol::kMaxStreams && static_cast<uint32_t>(fid) < revmap_[fsid].size()) {
+		return std::get<0>(revmap_[fsid][fid])->stream;
+	}
+	return nullptr;
+}
+
+bool Muxer::post(const StreamPacket &spkt, const Packet &pkt) {
+	SHARED_LOCK(mutex_, lk);
+	if (pkt.data.size() > 0 || !(spkt.flags & ftl::protocol::kFlagRequest)) available(spkt.frameSetID()) += spkt.channel;
+
+	if (spkt.streamID < ftl::protocol::kMaxStreams && spkt.frame_number < revmap_[spkt.streamID].size()) {
+		auto [se, ssid] = revmap_[spkt.streamID][spkt.frame_number];
+		//auto &se = streams_[sid];
+
+		if (!se) return false;
+
+		//LOG(INFO) << "POST " << spkt.frame_number;
+
+		StreamPacket spkt2 = spkt;
+		spkt2.streamID = se->original_fsid;  // FIXME: Multiple possible originals
+		spkt2.frame_number = ssid;
+		se->stream->select(spkt2.streamID, selected(spkt.frameSetID()));
+		return se->stream->post(spkt2, pkt);
+	} else {
+		return false;
+	}
+}
+
+bool Muxer::begin() {
+	bool r = true;
+	for (auto &s : streams_) {
+		r = r && s.stream->begin();
+	}
+	return r;
+}
+
+bool Muxer::end() {
+	bool r = true;
+	for (auto &s : streams_) {
+		r = r && s.stream->end();
+	}
+	return r;
+}
+
+bool Muxer::active() {
+	bool r = true;
+	for (auto &s : streams_) {
+		r = r && s.stream->active();
+	}
+	return r;
+}
+
+void Muxer::reset() {
+	for (auto &s : streams_) {
+		s.stream->reset();
+	}
+}
+
+int Muxer::_lookup(size_t fsid, Muxer::StreamEntry *se, int ssid, int count) {
+	SHARED_LOCK(mutex_, lk);
+	
+	auto i = se->maps.find(fsid);
+	if (i == se->maps.end()) {
+		lk.unlock();
+		{
+			UNIQUE_LOCK(mutex_, lk2);
+			if (se->maps.count(fsid) == 0) {
+				se->maps[fsid] = {};
+			}
+			i = se->maps.find(fsid);
+		}
+		lk.lock();
+	}
+
+	auto &map = i->second;
+
+	if (static_cast<uint32_t>(ssid) >= map.size()) {
+		lk.unlock();
+		{
+			UNIQUE_LOCK(mutex_, lk2);
+			while (static_cast<uint32_t>(ssid) >= map.size()) {
+				int nid = nid_[fsid]++;
+				revmap_[fsid].push_back({se, static_cast<uint32_t>(map.size())});
+				map.push_back(nid);
+				for (int j=1; j<count; ++j) {
+					int nid2 = nid_[fsid]++;
+					revmap_[fsid].push_back({se, static_cast<uint32_t>(map.size())});
+					map.push_back(nid2);
+				}
+			}
+		}
+		lk.lock();
+	}
+	return map[ssid];
+}
+
+void Muxer::_notify(const StreamPacket &spkt, const Packet &pkt) {
+	SHARED_LOCK(mutex_, lk);
+	available(spkt.frameSetID()) += spkt.channel;
+
+	try {
+		cb_.trigger(spkt, pkt);  // spkt.frame_number < 255 &&
+	} catch (std::exception &e) {
+		LOG(ERROR) << "Exception in packet handler (" << int(spkt.channel) << "): " << e.what();
+		//reset();  // Force stream reset here to get new i-frames
+	}
+}

src/streams.cpp

+#include <ftl/protocol/streams.hpp>
+
+using ftl::protocol::Stream;
+using ftl::protocol::Channel;
+using ftl::protocol::ChannelSet;
+
+const ChannelSet &Stream::available(int fs) const {
+	SHARED_LOCK(mtx_, lk);
+	if (fs < 0 || static_cast<uint32_t>(fs) >= state_.size()) throw FTL_Error("Frameset index out-of-bounds: " << fs);
+	return state_[fs].available;
+}
+
+ChannelSet Stream::selected(int fs) const {
+	SHARED_LOCK(mtx_, lk);
+	if (fs < 0 || static_cast<uint32_t>(fs) >= state_.size()) throw FTL_Error("Frameset index out-of-bounds: " << fs);
+	return state_[fs].selected;
+}
+
+ChannelSet Stream::selectedNoExcept(int fs) const {
+	if (fs == 255) return {};
+
+	SHARED_LOCK(mtx_, lk);
+	if (fs < 0 || static_cast<uint32_t>(fs) >= state_.size()) return {};
+	return state_[fs].selected;
+}
+
+void Stream::select(int fs, const ChannelSet &s, bool make) {
+	if (fs == 255) return;
+
+	UNIQUE_LOCK(mtx_, lk);
+	if (fs < 0 || (!make && static_cast<uint32_t>(fs) >= state_.size())) throw FTL_Error("Frameset index out-of-bounds: " << fs);
+	if (static_cast<uint32_t>(fs) >= state_.size()) state_.resize(fs+1);
+	state_[fs].selected = s;
+}
+
+ChannelSet &Stream::available(int fs) {
+	UNIQUE_LOCK(mtx_, lk);
+	if (fs < 0) throw FTL_Error("Frameset index out-of-bounds: " << fs);
+	if (static_cast<uint32_t>(fs) >= state_.size()) state_.resize(fs+1);
+	return state_[fs].available;
+}
+
+void Stream::reset() {
+	// Clear available and selected?
+}

test/CMakeLists.txt

@@ -49,3 +49,14 @@ target_link_libraries(net_performance beyond-protocol
 	GnuTLS::GnuTLS Threads::Threads ${UUID_LIBRARIES} ${URIPARSER_LIBRARIES} ${OS_LIBS})
 
 add_test(NetPerformance net_performance)
+
+### Stream Unit ################################################################
+add_executable(stream_unit
+	$<TARGET_OBJECTS:CatchTestFTL>
+	./stream_unit.cpp
+)
+target_include_directories(stream_unit PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/../include")
+target_link_libraries(stream_unit
+	beyond-protocol ${URIPARSER_LIBRARIES} ${OS_LIBS})
+
+add_test(StreamUnitTest stream_unit)
\ No newline at end of file

test/stream_unit.cpp

+#include "catch.hpp"
+
+#include <ftl/protocol/streams.hpp>
+#include <ftl/protocol/muxer.hpp>
+#include <ftl/protocol/broadcaster.hpp>
+#include <nlohmann/json.hpp>
+
+using ftl::protocol::Muxer;
+using ftl::protocol::Broadcast;
+using ftl::protocol::Stream;
+using ftl::protocol::StreamPacket;
+using ftl::protocol::Packet;
+using ftl::protocol::Channel;
+
+class TestStream : public ftl::protocol::Stream {
+	public:
+	TestStream() {};
+	~TestStream() {};
+
+	bool post(const ftl::protocol::StreamPacket &spkt, const ftl::protocol::Packet &pkt) {
+		available(spkt.streamID) += spkt.channel;
+		cb_.trigger(spkt, pkt);
+		return true;
+	}
+
+	bool begin() override { return true; }
+	bool end() override { return true; }
+	bool active() override { return true; }
+
+	private:
+	//std::function<void(const StreamPacket &, const Packet &)> cb_;
+};
+
+TEST_CASE("ftl::stream::Muxer()::write", "[stream]") {
+
+	std::unique_ptr<Muxer> mux = std::make_unique<Muxer>();
+	REQUIRE(mux);
+
+	SECTION("write with one stream") {
+		std::shared_ptr<Stream> s = std::make_shared<TestStream>();
+		REQUIRE(s);
+
+		mux->add(s);
+
+		ftl::protocol::StreamPacket tspkt = {4,0,0,1, Channel::kColour};
+
+		auto h = s->onPacket([&tspkt](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt = spkt;
+			return true;
+		});
+
+		REQUIRE( !mux->post({4,100,0,1,ftl::protocol::Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 0 );
+	}
+
+	SECTION("write to previously read") {
+
+		std::shared_ptr<Stream> s1 = std::make_shared<TestStream>();
+		REQUIRE(s1);
+		std::shared_ptr<Stream> s2 = std::make_shared<TestStream>();
+		REQUIRE(s2);
+
+		mux->add(s1);
+		mux->add(s2);
+
+		ftl::protocol::StreamPacket tspkt = {4,0,0,1,Channel::kColour};
+		auto h = mux->onPacket([&tspkt](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt = spkt;
+			return true;
+		});
+
+		REQUIRE( s1->post({4,100,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 100 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		REQUIRE( s2->post({4,101,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 101 );
+		REQUIRE( tspkt.frame_number == 1 );
+
+		StreamPacket tspkt2 = {4,0,0,1,Channel::kColour};
+		StreamPacket tspkt3 = {4,0,0,1,Channel::kColour};
+		auto h2 = s1->onPacket([&tspkt2](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt2 = spkt;
+			return true;
+		});
+		auto h3 = s2->onPacket([&tspkt3](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt3 = spkt;
+			return true;
+		});
+
+		REQUIRE( mux->post({4,200,0,1,Channel::kColour},{}) );
+		REQUIRE( tspkt3.timestamp == 200 );
+		REQUIRE( tspkt3.frame_number == 0 );
+	}
+}
+
+TEST_CASE("ftl::stream::Muxer()::post multi-frameset", "[stream]") {
+
+	std::unique_ptr<Muxer> mux = std::make_unique<Muxer>();
+	REQUIRE(mux);
+
+	SECTION("write to previously read") {
+		std::shared_ptr<Stream> s1 = std::make_shared<TestStream>();
+		REQUIRE(s1);
+		std::shared_ptr<Stream> s2 = std::make_shared<TestStream>();
+		REQUIRE(s2);
+
+		mux->add(s1);
+		mux->add(s2,1);
+
+		StreamPacket tspkt = {4,0,0,1,Channel::kColour};
+		auto h = mux->onPacket([&tspkt](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt = spkt;
+			return true;
+		});
+
+		REQUIRE( s1->post({4,100,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.streamID == 0 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		REQUIRE( s2->post({4,101,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.streamID == 1 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		StreamPacket tspkt2 = {4,0,0,1,Channel::kColour};
+		StreamPacket tspkt3 = {4,0,0,1,Channel::kColour};
+		auto h2 = s1->onPacket([&tspkt2](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt2 = spkt;
+			return true;
+		});
+		auto h3 = s2->onPacket([&tspkt3](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt3 = spkt;
+			return true;
+		});
+
+		REQUIRE( mux->post({4,200,1,0,Channel::kColour},{}) );
+		REQUIRE( tspkt3.streamID == 0 );
+		REQUIRE( tspkt3.frame_number == 0 );
+	}
+}
+
+TEST_CASE("ftl::stream::Muxer()::read", "[stream]") {
+	std::unique_ptr<Muxer> mux = std::make_unique<Muxer>();
+	REQUIRE(mux);
+
+	SECTION("read with two writing streams") {
+		std::shared_ptr<Stream> s1 = std::make_shared<TestStream>();
+		REQUIRE(s1);
+		std::shared_ptr<Stream> s2 = std::make_shared<TestStream>();
+		REQUIRE(s2);
+
+		mux->add(s1);
+		mux->add(s2);
+
+		StreamPacket tspkt = {4,0,0,1,Channel::kColour};
+		auto h = mux->onPacket([&tspkt](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt = spkt;
+			return true;
+		});
+
+		REQUIRE( s1->post({4,100,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 100 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		REQUIRE( s2->post({4,101,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 101 );
+		REQUIRE( tspkt.frame_number == 1 );
+
+		REQUIRE( s1->post({4,102,0,1,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 102 );
+		REQUIRE( tspkt.frame_number == 2 );
+
+		REQUIRE( s2->post({4,103,0,1,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 103 );
+		REQUIRE( tspkt.frame_number == 3 );
+	}
+
+	SECTION("read consistency with two writing streams") {
+		std::shared_ptr<Stream> s1 = std::make_shared<TestStream>();
+		REQUIRE(s1);
+		std::shared_ptr<Stream> s2 = std::make_shared<TestStream>();
+		REQUIRE(s2);
+
+		mux->add(s1);
+		mux->add(s2);
+
+		StreamPacket tspkt = {4,0,0,1,Channel::kColour};
+		auto h = mux->onPacket([&tspkt](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt = spkt;
+			return true;
+		});
+
+		REQUIRE( s1->post({4,100,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 100 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		REQUIRE( s2->post({4,101,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 101 );
+		REQUIRE( tspkt.frame_number == 1 );
+
+		REQUIRE( s1->post({4,102,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 102 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		REQUIRE( s2->post({4,103,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.timestamp == 103 );
+		REQUIRE( tspkt.frame_number == 1 );
+	}
+}
+
+TEST_CASE("ftl::stream::Muxer()::read multi-frameset", "[stream]") {
+	std::unique_ptr<Muxer> mux = std::make_unique<Muxer>();
+	REQUIRE(mux);
+
+	//SECTION("read with two writing streams") {
+
+		std::shared_ptr<Stream> s1 = std::make_shared<TestStream>();
+		REQUIRE(s1);
+		std::shared_ptr<Stream> s2 = std::make_shared<TestStream>();
+		REQUIRE(s2);
+		std::shared_ptr<Stream> s3 = std::make_shared<TestStream>();
+		REQUIRE(s3);
+		std::shared_ptr<Stream> s4 = std::make_shared<TestStream>();
+		REQUIRE(s4);
+
+		mux->add(s1,0);
+		mux->add(s2,1);
+		mux->add(s3,0);
+		mux->add(s4,1);
+
+		StreamPacket tspkt = {4,0,0,1,Channel::kColour};
+		auto h = mux->onPacket([&tspkt](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt = spkt;
+			return true;
+		});
+
+		REQUIRE( s1->post({4,100,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.streamID == 0 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		REQUIRE( s2->post({4,101,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.streamID == 1 );
+		REQUIRE( tspkt.frame_number == 0 );
+
+		REQUIRE( s3->post({4,102,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.streamID == 0 );
+		REQUIRE( tspkt.frame_number == 1 );
+
+		REQUIRE( s4->post({4,103,0,0,Channel::kColour},{}) );
+		REQUIRE( tspkt.streamID == 1 );
+		REQUIRE( tspkt.frame_number == 1 );
+	//}
+}
+
+TEST_CASE("ftl::stream::Broadcast()::write", "[stream]") {
+	std::unique_ptr<Broadcast> mux = std::make_unique<Broadcast>();
+	REQUIRE(mux);
+
+	SECTION("write with two streams") {
+		std::shared_ptr<Stream> s1 = std::make_shared<TestStream>();
+		REQUIRE(s1);
+		std::shared_ptr<Stream> s2 = std::make_shared<TestStream>();
+		REQUIRE(s2);
+
+		mux->add(s1);
+		mux->add(s2);
+
+		StreamPacket tspkt1 = {4,0,0,1,Channel::kColour};
+		StreamPacket tspkt2 = {4,0,0,1,Channel::kColour};
+
+		auto h1 = s1->onPacket([&tspkt1](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt1 = spkt;
+			return true;
+		});
+		auto h2 = s2->onPacket([&tspkt2](const StreamPacket &spkt, const Packet &pkt) {
+			tspkt2 = spkt;
+			return true;
+		});
+
+		REQUIRE( mux->post({4,100,0,1,Channel::kColour},{}) );
+		REQUIRE( tspkt1.timestamp == 100 );
+		REQUIRE( tspkt2.timestamp == 100 );
+	}
+
+}