Merge branch 'master' into feature/gui2

web-service/server/src/index.js

@@ -32,6 +32,8 @@ let uri_data = {};
 
 let peer_data = [];
 
+let cfg_to_peer = {};
+
 /**
  * A client stream request object. Each source maintains a list of clients who
  * are wanting frames from that source. Clients can only request N frames at a
@@ -244,6 +246,7 @@ app.get('/stream', (req, res) => {
 
 function checkStreams(peer) {
 	if (!peer.master) {
+		setTimeout(() => {
 			peer.rpc("list_streams", (streams) => {
 				console.log("STREAMS", streams);
 				for (let i=0; i<streams.length; i++) {
@@ -254,6 +257,14 @@ function checkStreams(peer) {
 					uri_data[streams[i]] = new RGBDStream(streams[i], peer);
 				}
 			});
+
+			peer.rpc("list_configurables", (cfgs) => {
+				console.log("CONFIGS", cfgs);
+				for (let i=0; i<cfgs.length; i++) {
+					if (!cfg_to_peer.hasOwnProperty(cfgs[i])) cfg_to_peer[cfgs[i]] = peer;
+				}
+			});
+		}, 500);  // Give a delay to allow startup
 	}
 }
 
@@ -265,6 +276,14 @@ function broadcastExcept(exc, name, ...args) {
 	}
 }
 
+function locateConfigPeer(uri) {
+	let cur_uri = uri;
+	while (cur_uri.length > 0 && !cfg_to_peer.hasOwnProperty(cur_uri)) {
+		cur_uri = cur_uri.substring(0, cur_uri.lastIndexOf('/'));
+	}
+	return (cur_uri.length > 0) ? cfg_to_peer[cur_uri] : null;
+}
+
 
 app.ws('/', (ws, req) => {
 	console.log("New web socket request");
@@ -305,6 +324,11 @@ app.ws('/', (ws, req) => {
 			delete peer_uris[peer.string_id];
 		}
 		if (peer_by_id.hasOwnProperty(peer.string_id)) delete peer_by_id[peer.string_id];
+
+		// Clear configurables
+		for (let c in cfg_to_peer) {
+			if (cfg_to_peer[c] === p) delete cfg_to_peer[c];
+		}
 	});
 
 	p.bind("new_peer", (id) => {
@@ -324,6 +348,25 @@ app.ws('/', (ws, req) => {
 		return Object.keys(uri_data);
 	});
 
+	p.bind("list_configurables", () => {
+		let result = [];
+		for (let c in cfg_to_peer) {
+			if (cfg_to_peer[c] !== p) result.push(c);
+		}
+		console.log("List Configs: ", result);
+		return result;
+	});
+
+	p.proxy("get_configurable", (cb, uri) => {
+		if (cfg_to_peer.hasOwnProperty(uri)) {
+			let peer = cfg_to_peer[uri];
+			peer.rpc("get_configurable", cb, uri);
+		} else {
+			console.log("Failed to get configurable ", uri);
+			return "{}";
+		}
+	});
+
 	p.bind("find_stream", (uri) => {
 		const parsedURI = stringSplitter(uri)
 		if (uri_to_peer.hasOwnProperty(parsedURI)) {
@@ -332,7 +375,7 @@ app.ws('/', (ws, req) => {
 			if (!p.isBound(uri)) {
 				console.log("Adding local stream binding");
 				p.bind(uri, (ttimeoff, spkt, pkt) => {
-					console.log("STREAM: ", ttimeoff, spkt, pkt);
+					//console.log("STREAM: ", ttimeoff, spkt, pkt);
 					let speer = uri_to_peer[parsedURI];
 					if (speer) {
 						try {
@@ -436,11 +479,9 @@ app.ws('/', (ws, req) => {
 	 * Update certain URIs values
 	 */
 	 p.bind("update_cfg", (uri, json) => {
-		const parsedURI = stringSplitter(uri)
-		console.log("URI", uri)
-		console.log("JSON", json)
-		if(uri_to_peer[parsedURI]){
-			let peer = uri_to_peer[parsedURI]
+		let peer = locateConfigPeer(uri);
+
+		if (peer) {
 			peer.send("update_cfg", uri, json)
 		}else{
 			console.log("Failed to update the configuration uri", uri)

web-service/server/src/peer.js

@@ -71,8 +71,8 @@ function Peer(ws) {
 		this._notify("disconnect", this);
 	}
 
-	let error = () => {
-		console.error("Socket error");
+	let error = (e) => {
+		console.error("Socket error: ", e);
 		this.sock.close();
 		this.status = kDisconnected;
 	}