Allow data packets in frameset frame

components/streams/src/builder.cpp

@@ -80,6 +80,10 @@ std::shared_ptr<ftl::data::FrameSet> Builder::_get(int64_t timestamp) {
 }
 
 ftl::data::Frame &Builder::get(int64_t timestamp, size_t ix) {
+	if (ix == 255) {
+		auto fs = _get(timestamp);
+		return *fs;
+	} else {
 		if (timestamp <= 0 || ix >= 32) throw FTL_Error("Invalid frame timestamp or index (" << timestamp << ", " << ix << ")");
 
 		UNIQUE_LOCK(mutex_, lk);
@@ -99,16 +103,9 @@ ftl::data::Frame &Builder::get(int64_t timestamp, size_t ix) {
 			}
 		}
 
-	//if (fs->frames.size() < size_) fs->frames.resize(size_);
-
-	//lk.unlock();
-	//SHARED_LOCK(fs->mtx, lk2);
-
-	//frame.swapTo(ftl::codecs::kAllChannels, fs->frames[ix]);
-
-	//fs->frames[ix].id = ix;
 		return fs->frames[ix];
 	}
+}
 
 void Builder::completed(int64_t ts, size_t ix) {
 	std::shared_ptr<ftl::data::FrameSet> fs;
@@ -118,7 +115,13 @@ void Builder::completed(int64_t ts, size_t ix) {
 		fs = _findFrameset(ts);
 	}
 	
-	if (fs && ix < fs->frames.size()) {
+	if (fs && ix == 255) {
+		// Note: Frameset can't complete without frames
+		//if (bufferSize_ == 0 && !fs->test(ftl::data::FSFlag::STALE) && static_cast<unsigned int>(fs->count) >= size_) {
+		//	UNIQUE_LOCK(mutex_, lk);
+		//	_schedule();
+		//}
+	} else if (fs && ix < fs->frames.size()) {
 		{
 			UNIQUE_LOCK(fs->mutex(), lk2);
 

components/streams/src/receiver.cpp

@@ -140,6 +140,7 @@ Receiver::InternalAudioStates &Receiver::_getAudioFrame(const StreamPacket &spkt
 	return f;
 }
 
+// TODO: Remove, not used
 void Receiver::_processState(const StreamPacket &spkt, const Packet &pkt) {
 	for (int i=0; i<pkt.frame_count; ++i) {
 		InternalVideoStates &frame = _getVideoFrame(spkt,i);

components/streams/test/receiver_unit.cpp

@@ -486,6 +486,37 @@ TEST_CASE( "Receiver for data channels" ) {
 		REQUIRE( count == 1 );
 	}
 
+	SECTION("a single data packet in overall frameset") {
+
+		spkt.frame_number = 255;
+		pkt.data.resize(0);
+		ftl::util::FTLVectorBuffer buf(pkt.data);
+		msgpack::pack(buf, 5.0f);
+
+		stream.post(spkt, pkt);
+
+		// Need to have at least one frame for this to work
+		spkt.frame_number = 0;
+		stream.post(spkt, pkt);
+
+		int count = 0;
+		auto h = receiver->onFrameSet([&count](const std::shared_ptr<ftl::data::FrameSet>& fs) {
+			++count;
+
+			REQUIRE( fs->timestamp() == 10 );
+			REQUIRE( fs->frames.size() == 1 );
+			REQUIRE( fs->hasChannel(Channel::Data) );
+			REQUIRE( fs->get<float>(Channel::Data) == 5.0f );
+
+			return true;
+		});
+
+		int i=10;
+		while (i-- > 0 && count < 1) std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+		REQUIRE( count == 1 );
+	}
+
 	SECTION("a single calibration packet") {
 
 		pkt.data.resize(0);
@@ -514,6 +545,100 @@ TEST_CASE( "Receiver for data channels" ) {
 		REQUIRE( count == 1 );
 	}
 
+	SECTION("a single pose packet") {
+
+		pkt.data.resize(0);
+		ftl::util::FTLVectorBuffer buf(pkt.data);
+		Eigen::Matrix4d pose;
+		msgpack::pack(buf, pose);
+
+		stream.post(spkt, pkt);
+
+		int count = 0;
+		auto h = receiver->onFrameSet([&count](const std::shared_ptr<ftl::data::FrameSet>& fs) {
+			++count;
+
+			REQUIRE( fs->timestamp() == 10 );
+			REQUIRE( fs->frames.size() == 1 );
+			REQUIRE( fs->frames[0].hasChannel(Channel::Data) );
+			fs->frames[0].get<Eigen::Matrix4d>(Channel::Data);
+
+			return true;
+		});
+
+		int i=10;
+		while (i-- > 0 && count < 1) std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+		REQUIRE( count == 1 );
+	}
+
 	ftl::timer::stop(true);
 	delete receiver;
 }
+
+// TODO: Annoying to test because I need to create valid audio encoding
+/*TEST_CASE( "Receiver for audio channels" ) {
+	//ftl::data::make_channel<ftl::rgbd::Camera>(Channel::Calibration, "calibration", ftl::data::StorageMode::PERSISTENT);
+	json_t global = json_t{{"$id","ftl://test"}};
+	ftl::config::configure(global);
+
+	ftl::data::Pool pool(5,7);
+
+	json_t cfg = json_t{
+		{"$id","ftl://test/1"}
+	};
+	auto *receiver = ftl::create<Receiver>(cfg, &pool);
+
+	json_t cfg2 = json_t{
+		{"$id","ftl://test/2"}
+	};
+	TestStream stream(cfg2);
+	receiver->setStream(&stream);
+	receiver->set("frameset_buffer_size", 0);
+
+	ftl::codecs::Packet pkt;
+	pkt.codec = codec_t::OPUS;
+	pkt.bitrate = 255;
+	pkt.flags = 0;
+	pkt.frame_count = 1;
+
+	ftl::codecs::StreamPacket spkt;
+	spkt.version = 4;
+	spkt.timestamp = 10;
+	spkt.frame_number = 0;
+	spkt.channel = Channel::AudioMono;
+	spkt.streamID = 0;
+
+	ftl::timer::start(false);
+
+	SECTION("a single data packet") {
+
+		pkt.data.resize(0);
+		ftl::util::FTLVectorBuffer buf(pkt.data);
+		msgpack::pack(buf, 5.0f);
+
+		stream.post(spkt, pkt);
+
+		int count = 0;
+		auto h = receiver->onFrameSet([&count](const std::shared_ptr<ftl::data::FrameSet>& fs) {
+			++count;
+
+			REQUIRE( fs->timestamp() == 10 );
+			REQUIRE( fs->frames.size() == 1 );
+			REQUIRE( fs->frames[0].hasChannel(Channel::Data) );
+			REQUIRE( fs->frames[0].get<float>(Channel::Data) == 5.0f );
+
+			return true;
+		});
+
+		int i=10;
+		while (i-- > 0 && count < 1) std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+		REQUIRE( count == 1 );
+	}
+
+	
+
+	ftl::timer::stop(true);
+	delete receiver;
+}*/