Use Shared MQTT Connector (#746)

* add MqttTopicConnector to shared MqttConnector between workers * Fix issue while switching between REPLAY & RT, the callback functions were not set back * Fix issue depending on new promise architecture * Add new SweApiDataSource updater to update min/max of a dataSource * Fix issue declaring Handler into dataSource worker, after switching the mode, the same handler was use (a map was caching the object). The handler is now automatically re-init after 'init' function * Refactor DataSourceWorker to use internal object * recreate the handler object into DataSourceWorker after switching to REPLAY or RT mode * add MISB live demo * pass mode into DataSource worker to differentiate kind of worker (because dataSource is still the same in RT or REPLAY) * rename misb-live to misb and remove old demo

Author: mdhsl

demos/misb/src/App.vue

@@ -71,15 +71,15 @@ export default {
  beforeMount() {
  const START_TIME = '2012-06-29T14:32:34.099333251Z';
  const END_TIME = '2012-06-29T14:36:54.033333251Z';
- const MODE = Mode.REPLAY;
+ const MODE = Mode.REAL_TIME;
  //
 
  // const START_TIME = 'now';
  // const END_TIME = '2055-01-01'
  // const MODE = Mode.REAL_TIME;
 
  const MIN_TIME = '2012-06-29T14:32:34.099333251Z';
- const MAX_TIME = '2012-06-29T14:36:54.033333251Z';
+ const MAX_TIME = '2055-06-29T14:36:54.033333251Z';
  const tls = true;
 
  const dsReplaySpeed = 2.6;
@@ -96,15 +96,15 @@ export default {
  endTime: END_TIME,
  minTime: MIN_TIME,
  maxTime: MAX_TIME,
- mode: MODE,
+ mode: Mode.REAL_TIME,
  replaySpeed: dsReplaySpeed,
  prefetchBatchDuration: 10000,
  prefetchBatchSize: 250
  };
 
  const droneVideoDataSource = new SweApiDatasource('MISB Drone - Video', {
  ...commonDatasourceOpts,
- resource: '/datastreams/8ni90dbu4uf0g/observations',
+ resource: '/datastreams/h225hesual08g/observations',
  responseFormat: 'application/swe+binary',
  });
 
@@ -113,37 +113,37 @@ export default {
 
  const droneLocationDataSource = new SweApiDatasource('MISB UAS - Platform Location', {
  ...commonDatasourceOpts,
- resource: '/datastreams/fled6eics1cl4/observations',
+ resource: '/datastreams/o7pce3e60s0ie/observations',
  responseFormat: 'application/swe+json',
  });
 
  const droneOrientationDataSource = new SweApiDatasource('MISB UAS - Platform Attitude', {
  ...commonDatasourceOpts,
- resource: '/datastreams/adheadf9nghts/observations',
+ resource: '/datastreams/mlme3gtdfepvc/observations',
  responseFormat: 'application/swe+json',
  });
 
  const droneCameraOrientationDataSource = new SweApiDatasource('MISB UAS - Gimbal Attitude', {
  ...commonDatasourceOpts,
- resource: '/datastreams/hrpo1u6r5096i/observations',
+ resource: '/datastreams/h75dmug8e3sae/observations',
  responseFormat: 'application/swe+binary',
  });
 
  const droneHFovDataSource = new SweApiDatasource('MISB UAS - Horizontal FoV', {
  ...commonDatasourceOpts,
- resource: '/datastreams/d962edate9okm/observations',
+ resource: '/datastreams/nhcs9rp6713ka/observations',
  responseFormat: 'application/swe+binary',
  });
 
  const droneVFovDataSource = new SweApiDatasource('MISB UAS - Vertical FoV', {
  ...commonDatasourceOpts,
- resource: '/datastreams/d962edate9okm/observations',
+ resource: '/datastreams/nhcs9rp6713ka/observations',
  responseFormat: 'application/swe+binary',
  });
 
  const geoRefImageFrameDataSource = new SweApiDatasource('MISB UAS - GeoReferenced Image Frame', {
  ...commonDatasourceOpts,
- resource: '/datastreams/p3mp2peibksl4/observations',
+ resource: '/datastreams/iabpf1ivua1qm/observations',
  responseFormat: 'application/swe+csv',
  });
 
@@ -166,13 +166,14 @@ export default {
  masterTimeRefreshRate: 250,
  startTime: START_TIME,
  endTime: END_TIME,
+ mode: Mode.REAL_TIME,
  dataSources: [
  droneLocationDataSource,
  droneVideoDataSource,
  droneOrientationDataSource,
  droneCameraOrientationDataSource,
  geoRefImageFrameDataSource,
- targetLocationDataSource,
+ // targetLocationDataSource,
  droneVFovDataSource,
  droneHFovDataSource
  ]

demos/misb/src/components/CollapseTimeController.vue

@@ -9,6 +9,7 @@
  <TimeController
  :dataSynchronizer="dataSynchronizer"
  @event='onControlEvent'
+ trackRealtime=""
  :skipTimeStep="'5%'"
  ></TimeController>
  <v-btn

showcase/examples/leaflet-location/leaflet-location.js

@@ -12,7 +12,7 @@ let gpsDataSource = new SosGetResult("android-GPS", {
  endpointUrl: 'sensiasoft.net/sensorhub/sos',
  offeringID: 'urn:android:device:060693280a28e015-sos',
  observedProperty: 'http://sensorml.com/ont/swe/property/Location',
- startTime: '2015-02-16T07:58:15.447Z',
+ startTime: '2015-02-16T08:01:15.447Z',
  endTime: '2015-02-16T08:09:00Z',
  mode: Mode.REPLAY,
  tls: true,

source/core/connector/MqttConnector.js

@@ -46,6 +46,7 @@ const mqttProviders = {};
 class MqttConnector extends DataConnector {
  /**
  *
+ * @param url
  * @param properties -
  */
  constructor(url, properties) {
@@ -54,6 +55,26 @@ class MqttConnector extends DataConnector {
  ...properties
  });
  this.interval = -1;
+ this.id = `mqtt-connector-${randomUUID()}`;
+ }
+
+ initBc() {
+ this.onMessage = (data, topic) => {
+ this.broadcastChannel.postMessage({
+ data: data,
+ topic: topic
+ }, [data]);
+ }
+ this.broadcastChannel = new BroadcastChannel(this.id);
+ let queryString;
+ this.broadcastChannel.onmessage = (message) => {
+ if(message.data.message === 'subscribe') {
+ queryString = message.data.queryString;
+ this.doRequest(message.data.topic, message.data.queryString);
+ } else if(message.data.message === 'unsubscribe') {
+ this.disconnect(message.data.topic+'?'+queryString);
+ }
+ }
  }
 
  getMqttProvider() {
@@ -99,6 +120,7 @@ class MqttConnector extends DataConnector {
  */
  doRequest(topic = '',queryString= undefined) {
  const mqttProvider = this.getMqttProvider();
+
  mqttProvider.subscribe(`${topic}?${queryString}`, this.onMessage).then(() => {
  this.onChangeStatus(Status.CONNECTED);
  });
@@ -112,7 +134,7 @@ class MqttConnector extends DataConnector {
  /**
  * Disconnects and close the mqtt client.
  */
- async disconnect() {
+ async disconnect(topic) {
  // does not call super to avoid reconnection logic and use the one of the mqtt.js lib
  // this.checkStatus(Status.DISCONNECTED);
  // this.init = false;
@@ -121,9 +143,13 @@ class MqttConnector extends DataConnector {
  const client = mqttProviders[this.getUrl()];
 
  if (isDefined(client) && client.isConnected()) {
- // unsubscribe all topics
- await client.unsubscribeAll();
- // client.disconnect();
+ if(!topic) {
+ // unsubscribe all topics
+ return client.unsubscribeAll();
+ // client.disconnect();
+ } else {
+ return client.unsubscribe(topic);
+ }
  }
  //delete mqttProviders[this.getUrl()];
  //console.warn(`Disconnected from ${this.getUrl()}`);

source/core/connector/MqttTopicConnector.js

@@ -0,0 +1,104 @@
+import DataConnector from "./DataConnector";
+import {Status} from "./Status";
+
+/**
+ * Defines the MqttTopicConnector to communicate with Shared MqttConnector using broadcast channel
+ * @extends DataConnector
+ */
+class MqttTopicConnector extends DataConnector {
+ /**
+ *
+ * @param properties -
+ */
+ constructor(url, properties) {
+ super(url, properties);
+ this.lastReceiveTime = -1;
+ this.interval = -1;
+ this.broadcastChannel = new BroadcastChannel(url);
+ this.broadcastChannel.onmessage = (message) => {
+ if(message.data.topic === this.fullTopic) {
+ this.onMessage(message.data.data);
+ }
+ }
+ this.topics = [];
+ }
+
+ doRequest(topic = '',queryString= undefined) {
+ this.fullTopic = topic + '?'+queryString;
+ this.broadcastChannel.postMessage({
+ message: 'subscribe',
+ connectorId: this.id,
+ topic: topic,
+ queryString: queryString
+ });
+ this.topics.push(topic);
+ this.onChangeStatus(Status.CONNECTED);
+ }
+ /**
+ * Disconnects.
+ */
+ disconnect() {
+ this.fullDisconnect(true);
+ }
+
+ /**
+ * Fully disconnect the websocket connection by sending a close message to the webWorker.
+ * @param {Boolean} removeInterval - force removing the interval
+ */
+ fullDisconnect(removeInterval) {
+ if (this.broadcastChannel != null) {
+ for(let topic of this.topics) {
+ this.broadcastChannel.postMessage({
+ message: 'unsubscribe',
+ connectorId: this.id,
+ topic: topic,
+ });
+ }
+ this.broadcastChannel.close();
+ this.broadcastChannel = null;
+ this.topics = [];
+ }
+ if (removeInterval) {
+ clearInterval(this.interval);
+ }
+ this.opened = false;
+ }
+
+ /**
+ * Try to reconnect if the connexion if closed
+ */
+ reconnect() {
+ this.onReconnect();
+ if (this.init) {
+ this.fullDisconnect(false);
+ }
+ this.connect();
+
+ }
+
+ /**
+ * The onMessage method used by the websocket to callback the data
+ * @param data the callback data
+ * @event
+ */
+ onMessage(data) {
+ }
+
+ /**
+ * Closes the webSocket.
+ */
+ close() {
+ this.disconnect();
+ }
+
+ isConnected() {
+ return this.broadcastChannel !== null && this.opened;
+ }
+
+ checkStatus(status) {
+ this.onChangeStatus(status);
+ this.status = status;
+ }
+}
+
+export default MqttTopicConnector;