Fix reset dataSource by changing properties, disconnect correctly, removing from workerPool, removing handler from worker and connect again

Author: mdhsl

source/core/datasource/DataSource.datasource.js

@@ -66,8 +66,8 @@ class DataSource {
  }
 
  terminate() {
- if (this.dataSourceWorker !== null) {
- this.dataSourceWorker.terminate();
+ if (this.getWorker() !== null) {
+ this.getWorker().terminate();
  }
  }
 
@@ -107,7 +107,7 @@ class DataSource {
  ...this.properties,
  ...properties
  };
- return this.dataSourceWorker.postMessageWithAck({
+ return this.getWorker().postMessageWithAck({
  message: 'update-properties',
  data: properties,
  dsId: this.id
@@ -136,8 +136,7 @@ class DataSource {
  }
 
  async initDataSource(properties = this.properties) {
- this.dataSourceWorker = this.getWorker();
- return this.dataSourceWorker.postMessageWithAck({
+ return this.getWorker().postMessageWithAck({
  message: 'init',
  id: this.id,
  properties: properties,
@@ -176,7 +175,7 @@ class DataSource {
  }
 
  async doConnect() {
- return this.dataSourceWorker.postMessageWithAck({
+ return this.getWorker().postMessageWithAck({
  message: 'connect',
  dsId: this.id
  });
@@ -187,7 +186,7 @@ class DataSource {
  return false;
  } else {
  return this.checkInit().then(() => {
- return this.dataSourceWorker.postMessageWithAck({
+ return this.getWorker().postMessageWithAck({
  message: 'is-connected',
  dsId: this.id
  });
@@ -200,7 +199,7 @@ class DataSource {
  */
  async disconnect() {
  await this.checkInit();
- return this.dataSourceWorker.postMessageWithAck({
+ return this.getWorker().postMessageWithAck({
  message: 'disconnect',
  dsId: this.id
  });
@@ -209,8 +208,21 @@ class DataSource {
  async onDisconnect() {
  }
 
- reset() {
- this.init = undefined;
+ async reset() {
+ await this.disconnect();
+ this.resetInit();
+ return this.removeWorker();
+ }
+
+ async removeWorker() {
+ if (this.id in dataSourceWorkers) {
+ return this.getWorker().postMessageWithAck({
+ message: 'remove-handler',
+ dsId: this.id
+ }).then(() => {
+ delete dataSourceWorkers[this.id]; // delete index from pool
+ });
+ }
  }
 
  onRemovedDataSource(dataSourceId) {

source/core/datasource/TimeSeries.datasource.js

@@ -35,11 +35,11 @@ class TimeSeriesDatasource {
  ...properties
  });
 
- this.setMode(properties.mode);
+ this.setMode(properties.mode, false);
  }
 
- async setMode(mode) {
- if (this.timeSeriesDataSource) {
+ async setMode(mode, disconnect = true) {
+ if (disconnect && this.timeSeriesDataSource) {
  await this.timeSeriesDataSource.disconnect();
  }
 
@@ -320,11 +320,10 @@ class TimeSeriesDatasource {
  return this.timeSeriesDataSource.isConnected();
  }
 
- reset() {
- this.timeSeriesDataSource.reset();
+ async reset() {
+ return this.timeSeriesDataSource.reset();
  }
 
-
  onTimeChanged(min, max, start, end) {
  }
 

source/core/datasource/TimeSeries.realtime.datasource.js

@@ -18,6 +18,7 @@ import {DATA_SYNCHRONIZER_TOPIC, DATASOURCE_TIME_TOPIC} from "../Constants";
 import DataSource from "./DataSource.datasource";
 import {Mode} from './Mode';
 import MqttConnector from "../connector/MqttConnector";
+import {isDefined} from "../utils/Utils";
 
 /**
  * The DataSource is the abstract class used to create different datasources.
@@ -186,7 +187,7 @@ class TimeSeriesRealtimeDatasource extends DataSource {
  await this.checkInit();
  const topic = DATA_SYNCHRONIZER_TOPIC + this.dataSynchronizer.id;
  this.properties.version = this.dataSynchronizer.version();
- return this.dataSourceWorker.postMessageWithAck({
+ return this.getWorker().postMessageWithAck({
  message: 'topics',
  topics: {
  data: topic,
@@ -201,29 +202,32 @@ class TimeSeriesRealtimeDatasource extends DataSource {
  }
 
  async removeDataSynchronizer() {
- if (this.dataSourceWorker) {
- this.dataSourceWorker.terminate();
- this.dataSynchronizer = undefined;
- }
+ await this.removeWorker();
+ this.dataSynchronizer = undefined;
  // this.init = undefined;
  return this.checkInit();
-
  }
 
  /**
  * Disconnect the dataSource then the protocol will be closed as well.
  */
  async disconnect() {
- await this.checkInit();
- return this.dataSourceWorker.postMessageWithAck({
- message: 'disconnect',
- dsId: this.id,
- mode: Mode.REAL_TIME,
- });
+ if (isDefined(this.init)) {
+ try {
+ return this.getWorker().postMessageWithAck({
+ message: 'disconnect',
+ dsId: this.id,
+ mode: Mode.REAL_TIME,
+ });
+ } catch (ex) {
+ console.error(ex);
+ }
+ }
  }
 
  async doConnect() {
- return this.dataSourceWorker.postMessageWithAck({
+ await this.checkInit();
+ return this.getWorker().postMessageWithAck({
  message: 'connect',
  startTime: 'now',
  version: this.version(),
@@ -247,7 +251,7 @@ class TimeSeriesRealtimeDatasource extends DataSource {
  topics.sync = this.dataSynchronizer.getTimeTopicId()
  }
 
- return this.dataSourceWorker.postMessageWithAck({
+ return this.getWorker().postMessageWithAck({
  message: 'topics',
  topics: topics,
  dsId: this.id,
@@ -273,6 +277,12 @@ class TimeSeriesRealtimeDatasource extends DataSource {
  version() {
  return this.properties.version;
  }
+
+ async reset() {
+ console.warn(`dataSource ${this.id} has been reset`);
+ await super.reset();
+ return this.doConnect();
+ }
 }
 
 export default TimeSeriesRealtimeDatasource;

source/core/datasource/TimeSeries.replay.datasource.js

@@ -333,6 +333,11 @@ class TimeSeriesReplayDatasource extends DataSource {
  version: version
  });
  }
+ async reset() {
+ console.warn(`dataSource ${this.id} has been reset`);
+ await super.reset();
+ return this.doConnect();
+ }
 }
 
 export default TimeSeriesReplayDatasource;

source/core/datasource/worker/DataSourceWorker.js

@@ -30,6 +30,8 @@ class DataSourceWorker {
  this.handleIsConnected(eventData, resp);
  } else if (eventData.message === 'is-init') {
  this.handleIsInit(eventData, resp);
+ } else if (eventData.message === 'remove-handler') {
+ this.handleRemoveHandler(eventData, resp);
  }
  } catch (ex) {
  console.error(ex);
@@ -82,6 +84,12 @@ class DataSourceWorker {
  this.postMessage(resp);
  }
 
+ handleRemoveHandler(eventData, resp) {
+ const dsId = eventData.dsId;
+ delete this.dataSourceHandlers[dsId];
+ this.postMessage(resp);
+ }
+
  handleIsInit(eventData, resp) {
  const dsId = eventData.dsId;
  resp.data = this.dataSourceHandlers[dsId].isInitialized();

source/core/timesync/DataSynchronizer.js

@@ -44,7 +44,7 @@ class DataSynchronizer {
  id: id+'-realtime'
  }, this)
  this.broadcastChannels = [];
- this.setMode(properties.mode || Mode.REPLAY).then(() => {
+ this.setMode(properties.mode || Mode.REPLAY, false).then(() => {
  this.dataSynchronizer.onTimeChanged = (min, max, start, end) => this.onTimeChanged(min, max, start, end);
  this.dataSynchronizer.onAddedDataSource = (dataSourceId) => this.onAddedDataSource(dataSourceId);
  this.dataSynchronizer.onRemovedDataSource = (dataSourceId) => this.onRemovedDataSource(dataSourceId);
@@ -55,8 +55,8 @@ class DataSynchronizer {
  return this.id;
  }
 
- async setMode(mode) {
- if (this.dataSynchronizer) {
+ async setMode(mode, disconnect = true) {
+ if (this.dataSynchronizer && disconnect) {
  await this.dataSynchronizer.disconnect();
  }
  if (mode === Mode.REPLAY) {
@@ -73,7 +73,7 @@ class DataSynchronizer {
  this.broadcastChannels = [];
  const promises=[];
  for(let ds of this.dataSynchronizer.getDataSources()) {
- promises.push(ds.setMode(mode));
+ promises.push(ds.setMode(mode, disconnect));
  }
  this.dataSynchronizer.onTimeChanged = (min, max, start, end) => this.onTimeChanged(min, max, start, end);
  this.dataSynchronizer.onAddedDataSource = (dataSourceId) => this.onAddedDataSource(dataSourceId);

source/core/timesync/rt/DataSynchronizer.realtime.js

@@ -214,6 +214,7 @@ class DataSynchronizerRealtime {
  * Disconnects all dataSources
  */
  async disconnect() {
+ console.log('disconnect')
  await this.reset();
  const promises = [];
  for (let dataSource of this.dataSources) {

source/core/ui/view/View.js

@@ -242,7 +242,6 @@ class View {
  if (event.data.type === EventType.STATUS && event.data.status === Status.CLOSED_ERROR) {
  self.reset();
  } else if (event.data.type === EventType.DATA) {
-
  const that = this;
 
  // transform the data