Fixed subscription issues.

openeo/files.js

@@ -35,7 +35,7 @@ module.exports = class FilesAPI {
 		.then(files => {
 			var output = files.map(file => {
 				return {
-					name: path.relative(this.getUserFolder(req.user._id), file.path),
+					name: this.getFileName(req.user._id, file.path),
 					size: file.stat.size,
 					modified: file.stat.mtime.toISOString()
 				}
@@ -107,10 +107,10 @@ module.exports = class FilesAPI {
 				stream.end();
 				const payload = {
 					user_id: req.user._id,
-					path: p.replace('storage/user_files/' + req.user._id + '/', ''),
+					path: this.getFileName(req.user._id, p),
 					action: fileExists ? 'updated' : 'created'
 				};
-				req.api.subscriptions.publish(req, 'openeo.files', payload, payload);
+				req.api.subscriptions.publish(req.user._id, 'openeo.files', payload, payload);
 				res.send(204);
 				return next();
 			});
@@ -137,10 +137,10 @@ module.exports = class FilesAPI {
 		.then(() => {
 			const payload = {
 				user_id: req.user._id,
-				path: p.replace('storage/user_files/'+req.user._id+'/', ''),
+				path: this.getFileName(req.user._id, p),
 				action: 'deleted'
 			};
-			req.api.subscriptions.publish(req, 'openeo.files', payload, payload);
+			req.api.subscriptions.publish(req.user._id, 'openeo.files', payload, payload);
 			res.send(204);
 			return next();
 		})
@@ -192,6 +192,10 @@ module.exports = class FilesAPI {
 		return null;
 	}
 
+	getFileName(user_id, p) {
+		return path.relative(this.getUserFolder(user_id), p);
+	}
+
 	isFile(path) {
 		return fse.stat(path).then(stat => {
 			if (stat.isFile()) {

openeo/jobs.js

@@ -229,7 +229,7 @@ module.exports = class JobsAPI {
 					parameters: processParams
 				};
 			}
-			req.api.subscriptions.publish(req, "openeo.jobs.debug", params, payload);
+			req.api.subscriptions.publish(req.user._id, "openeo.jobs.debug", params, payload);
 		} catch (e) {
 			console.log(e);
 		}

openeo/subscriptions.js

@@ -31,7 +31,7 @@ class SubscriptionConnection {
 	isSubscribed(topic, params) {
 		params.topic = topic;
 		var hash = Utils.hashJSON(params);
-		return (topics.get(hash) !== 'undefined');
+		return (this.topics.get(hash) !== 'undefined');
 	}
 
 	sendWelcomeMessage() {
@@ -50,7 +50,9 @@ class SubscriptionConnection {
 			payload: payload
 		};
 
-		this.connection.send(JSON.stringify(message));
+		if (this.connection.readyState === this.connection.OPEN) {
+			this.connection.send(JSON.stringify(message));
+		}
 	}
 
 	sendMessageIfSubscribed(topic, params, payload) {
@@ -80,14 +82,6 @@ module.exports = class SubscriptionsAPI {
 	beforeServerStart(server) {
 		server.addEndpoint('get', '/subscription', this.getSubscription.bind(this));
 
-		// ToDo: Remove test topic once this is a bit more stable
-		if (global.server.config.debug) {
-			this.registerTopic("openeo.test");
-			setInterval(() => {
-				this.broadcast("openeo.test", {}, {message: 'test'})
-			}, 1000);
-		}
-
 		return Promise.resolve();
 	}