Add uri group concept

components/streams/include/ftl/streams/feed.hpp

@@ -94,6 +94,7 @@ private:
 	//ftl::Handler<const ftl::data::FrameSetPtr&> frameset_cb_;
 	std::unordered_map<std::string, uint32_t> fsid_lookup_;
 	std::map<uint32_t, ftl::data::FrameSetPtr> latest_;
+	std::unordered_map<std::string, uint32_t> groups_;
 
 	std::unordered_map<uint32_t, ftl::stream::Stream*> streams_;
 	std::unordered_map<uint32_t, ftl::rgbd::Source*> devices_;
@@ -109,7 +110,7 @@ private:
 
 	uint32_t fs_counter_ = 0;
 
-	uint32_t allocateFrameSetId();
+	uint32_t allocateFrameSetId(const std::string &group);
 
 	void add(uint32_t fsid, const std::string &uri, ftl::stream::Stream *s);
 

components/streams/src/feed.cpp

@@ -239,8 +239,19 @@ Feed::~Feed() {
 }
 
 
-uint32_t Feed::allocateFrameSetId() {
+uint32_t Feed::allocateFrameSetId(const std::string &group) {
+	if (group.size() == 0) {
 		return fs_counter_++;
+	} else {
+		auto i = groups_.find(group);
+		if (i == groups_.end()) {
+			uint32_t id = fs_counter_++;
+			groups_[group] = id;
+			return id;
+		} else {
+			return i->second;
+		}
+	}
 }
 
 void Feed::select() {
@@ -301,8 +312,9 @@ Feed::Filter* Feed::filter(const std::unordered_set<Channel> &channels) {
 Feed::Filter* Feed::filter(const std::unordered_set<std::string> &sources, const std::unordered_set<Channel> &channels) {
 	std::unordered_set<uint32_t> fsids;
 	for (const auto &src : sources) {
+		ftl::URI uri(src);
 		// FIXME: If this map is modified, this could crash here.
-		auto i = fsid_lookup_.find(src);
+		auto i = fsid_lookup_.find(uri.getBaseURI());
 		if (i != fsid_lookup_.end()) {
 			fsids.emplace(i->second);
 		}
@@ -414,15 +426,18 @@ void Feed::updateNetSources() {
 
 	if (value("auto_host_sources", false)) {
 		for (auto s : netcams_) {
-			if (fsid_lookup_.count(s) == 0) {
+			ftl::URI uri(s);
+			const std::string group = uri.getAttribute<std::string>("group");
+
+			if (fsid_lookup_.count(uri.getBaseURI()) == 0) {
 				auto *stream = ftl::create<ftl::stream::Net>
 					(this, std::string("netstream")
 					+std::to_string(fsid_lookup_.size()), net_);
 
-				int fsid = allocateFrameSetId();
+				int fsid = allocateFrameSetId(group);
 
 				stream->set("uri", s);
-				add(fsid, s, stream);
+				add(fsid, uri.getBaseURI(), stream);
 
 				LOG(INFO)	<< "Add Stream: "
 							<< stream->value("uri", std::string("NONE"))
@@ -489,8 +504,9 @@ bool Feed::sourceAvailable(const std::string &uri) {
 	return false;
 }
 
-bool Feed::sourceActive(const std::string &uri) {
-	return fsid_lookup_.count(uri) > 0;
+bool Feed::sourceActive(const std::string &suri) {
+	ftl::URI uri(suri);
+	return fsid_lookup_.count(uri.getBaseURI()) > 0;
 }
 
 std::string Feed::getName(const std::string &puri) {
@@ -539,23 +555,24 @@ void Feed::add(uint32_t fsid, const std::string &uri, ftl::stream::Stream* strea
 
 uint32_t Feed::add(const std::string &path) {
 	UNIQUE_LOCK(mtx_, lk);
+	ftl::URI uri(path);
 
-	if (fsid_lookup_.count(path) > 0) return fsid_lookup_[path];
+	if (fsid_lookup_.count(uri.getBaseURI()) > 0) return fsid_lookup_[uri.getBaseURI()];
 
-	ftl::URI uri(path);
 	const auto scheme = uri.getScheme();
+	const std::string group = uri.getAttribute<std::string>("group");
 
 	if ((scheme == ftl::URI::SCHEME_OTHER) || // assumes relative path
 		(scheme == ftl::URI::SCHEME_FILE)) {
 
-		auto eix = path.find_last_of('.');
-		auto ext = path.substr(eix+1);
+		auto eix = uri.getPath().find_last_of('.');
+		auto ext = uri.getPath().substr(eix+1);
 
 		if (ext != "ftl") {
 			throw ftl::exception("bad filename (expects .ftl)");
 		}
 
-		const int fsid = allocateFrameSetId();
+		const int fsid = allocateFrameSetId(group);
 		auto* fstream = ftl::create<ftl::stream::File>
 			(this, std::string("ftlfile-") + std::to_string(fsid));
 
@@ -583,12 +600,12 @@ uint32_t Feed::add(const std::string &path) {
 		// TODO: URI normalization; should happen in add(,,) or add(,,,) take
 		// ftl::URI instead of std::string as argument. Note the bug above.
 		// TODO: write unit test for uri parsing
-		add(fsid, path, fstream);
+		add(fsid, uri.getBaseURI(), fstream);
 		return fsid;
 	}
 	else if (scheme == ftl::URI::SCHEME_DEVICE) {
-		int fsid = allocateFrameSetId();
-		fsid_lookup_[path] = fsid; // Manually add mapping
+		int fsid = allocateFrameSetId(group);
+		fsid_lookup_[uri.getBaseURI()] = fsid; // Manually add mapping
 
 		std::string srcname = std::string("source") + std::to_string(fsid);
 		uri.to_json(getConfig()[srcname]);
@@ -652,10 +669,10 @@ uint32_t Feed::add(const std::string &path) {
 			(this, std::string("netstream")
 			+std::to_string(fsid_lookup_.size()), net_);
 
-		int fsid = allocateFrameSetId();
+		int fsid = allocateFrameSetId(group);
 
 		stream->set("uri", path);
-		add(fsid, path, stream);
+		add(fsid, uri.getBaseURI(), stream);
 
 		LOG(INFO)	<< "Add Stream: "
 					<< stream->value("uri", std::string("NONE"))