-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: resubscribe on reconnect to different core node (#222)
Signed-off-by: Tristan Bastian <[email protected]>
- Loading branch information
Showing
5 changed files
with
210 additions
and
160 deletions.
There are no files selected for viewing
74 changes: 42 additions & 32 deletions
74
node-red-contrib-c8y-client/src/nodes/realtime-alarms/realtime-alarms.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,32 +1,42 @@ | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeAlarmsNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/alarms/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-alarms", RealtimeAlarmsNode); | ||
| } | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeAlarmsNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/alarms/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| const handle = node.client.realtime.addHandshakeListener((msg) => { | ||
| if (msg.successful && msg.reestablish) { | ||
| this.log(`Resubscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| node.client.realtime.resubscribe(subscription); | ||
| } | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && handle) { | ||
| node.client.realtime.removeListener(handle); | ||
| } | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-alarms", RealtimeAlarmsNode); | ||
| } |
74 changes: 42 additions & 32 deletions
74
node-red-contrib-c8y-client/src/nodes/realtime-device/realtime-device.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,32 +1,42 @@ | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeDeviceNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/inventory/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-device", RealtimeDeviceNode); | ||
| } | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeDeviceNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/inventory/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| const handle = node.client.realtime.addHandshakeListener((msg) => { | ||
| if (msg.successful && msg.reestablish) { | ||
| this.log(`Resubscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| node.client.realtime.resubscribe(subscription); | ||
| } | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && handle) { | ||
| node.client.realtime.removeListener(handle); | ||
| } | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-device", RealtimeDeviceNode); | ||
| } |
74 changes: 42 additions & 32 deletions
74
node-red-contrib-c8y-client/src/nodes/realtime-events/realtime-events.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,32 +1,42 @@ | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeEventNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/events/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-events", RealtimeEventNode); | ||
| } | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeEventNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/events/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| const handle = node.client.realtime.addHandshakeListener((msg) => { | ||
| if (msg.successful && msg.reestablish) { | ||
| this.log(`Resubscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| node.client.realtime.resubscribe(subscription); | ||
| } | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && handle) { | ||
| node.client.realtime.removeListener(handle); | ||
| } | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-events", RealtimeEventNode); | ||
| } |
74 changes: 42 additions & 32 deletions
74
node-red-contrib-c8y-client/src/nodes/realtime-measurements/realtime-measurements.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,32 +1,42 @@ | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeMeasurementNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/measurements/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-measurements", RealtimeMeasurementNode); | ||
| } | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeMeasurementNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/measurements/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| const handle = node.client.realtime.addHandshakeListener((msg) => { | ||
| if (msg.successful && msg.reestablish) { | ||
| this.log(`Resubscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| node.client.realtime.resubscribe(subscription); | ||
| } | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && handle) { | ||
| node.client.realtime.removeListener(handle); | ||
| } | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-measurements", RealtimeMeasurementNode); | ||
| } |
74 changes: 42 additions & 32 deletions
74
node-red-contrib-c8y-client/src/nodes/realtime-operations/realtime-operations.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,32 +1,42 @@ | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeOperationsNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/operations/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-operations", RealtimeOperationsNode); | ||
| } | ||
| const c8yClientLib = require('@c8y/client'); | ||
|
|
||
| module.exports = function(RED) { | ||
| function RealtimeOperationsNode(config) { | ||
| RED.nodes.createNode(this,config); | ||
| var node = this; | ||
| const tenant = process.env.C8Y_TENANT; | ||
| const baseUrl = process.env.C8Y_BASEURL; | ||
| const user = process.env.C8Y_USER; | ||
| const password = process.env.C8Y_PASSWORD; | ||
| const auth = new c8yClientLib.BasicAuth({tenant, user, password}); | ||
|
|
||
| node.client = new c8yClientLib.Client(auth, baseUrl); | ||
| node.client.core.tenant = tenant; | ||
|
|
||
| const topic = '/operations/' + (config.deviceId || '*'); | ||
| this.log(`Subscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| const subscription = node.client.realtime.subscribe(topic, (evt) => { | ||
| const msg = { | ||
| payload: evt | ||
| }; | ||
| node.send(msg); | ||
| }); | ||
|
|
||
| const handle = node.client.realtime.addHandshakeListener((msg) => { | ||
| if (msg.successful && msg.reestablish) { | ||
| this.log(`Resubscribing to: ${topic} on tenant: ${tenant} and url: ${baseUrl}`); | ||
| node.client.realtime.resubscribe(subscription); | ||
| } | ||
| }); | ||
|
|
||
| node.on('close', function() { | ||
| if (node.client && handle) { | ||
| node.client.realtime.removeListener(handle); | ||
| } | ||
| if (node.client && subscription) { | ||
| node.client.realtime.unsubscribe(subscription); | ||
| } | ||
| }); | ||
| } | ||
| RED.nodes.registerType("realtime-operations", RealtimeOperationsNode); | ||
| } |