Implement telemetry websocket service

ui-ngx/proxy.conf.json

@@ -2,5 +2,9 @@
   "/api": {
     "target": "http://localhost:8080",
     "secure": false
+  },
+  "/api/ws": {
+    "target": "ws://localhost:8080",
+    "ws": true
   }
 }

ui-ngx/src/app/core/ws/telemetry-websocket.service.ts

@@ -0,0 +1,309 @@
+///
+/// Copyright © 2016-2019 The Thingsboard Authors
+///
+/// Licensed under the Apache License, Version 2.0 (the "License");
+/// you may not use this file except in compliance with the License.
+/// You may obtain a copy of the License at
+///
+///     http://www.apache.org/licenses/LICENSE-2.0
+///
+/// Unless required by applicable law or agreed to in writing, software
+/// distributed under the License is distributed on an "AS IS" BASIS,
+/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/// See the License for the specific language governing permissions and
+/// limitations under the License.
+///
+
+import { Inject, Injectable } from '@angular/core';
+import {
+  AttributesSubscriptionCmd,
+  GetHistoryCmd,
+  SubscriptionCmd,
+  SubscriptionUpdate,
+  SubscriptionUpdateMsg,
+  TelemetryFeature,
+  TelemetryPluginCmdsWrapper,
+  TelemetryService,
+  TelemetrySubscriber,
+  TimeseriesSubscriptionCmd
+} from '@app/shared/models/telemetry/telemetry.models';
+import { select, Store } from '@ngrx/store';
+import { AppState } from '@core/core.state';
+import { AuthService } from '@core/auth/auth.service';
+import { selectIsAuthenticated } from '@core/auth/auth.selectors';
+import { WINDOW } from '@core/services/window.service';
+import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
+import { ActionNotificationShow } from '@core/notification/notification.actions';
+import Timeout = NodeJS.Timeout;
+
+const RECONNECT_INTERVAL = 2000;
+const WS_IDLE_TIMEOUT = 90000;
+const MAX_PUBLISH_COMMANDS = 10;
+
+@Injectable({
+  providedIn: 'root'
+})
+export class TelemetryWebsocketService implements TelemetryService {
+
+  isActive = false;
+  isOpening = false;
+  isOpened = false;
+  isReconnect = false;
+
+  socketCloseTimer: Timeout;
+  reconnectTimer: Timeout;
+
+  lastCmdId = 0;
+  subscribersCount = 0;
+  subscribersMap = new Map<number, TelemetrySubscriber>();
+
+  reconnectSubscribers = new Set<TelemetrySubscriber>();
+
+  cmdsWrapper = new TelemetryPluginCmdsWrapper();
+  telemetryUri: string;
+
+  dataStream: WebSocketSubject<TelemetryPluginCmdsWrapper | SubscriptionUpdateMsg>;
+
+  constructor(private store: Store<AppState>,
+              private authService: AuthService,
+              @Inject(WINDOW) private window: Window) {
+    this.store.pipe(select(selectIsAuthenticated)).subscribe(
+      (authenticated: boolean) => {
+        if (!authenticated) {
+          this.reset(true);
+        }
+      }
+    );
+
+    let port = this.window.location.port;
+    if (this.window.location.protocol === 'https:') {
+      if (!port) {
+        port = '443';
+      }
+      this.telemetryUri = 'wss:';
+    } else {
+      if (!port) {
+        port = '80';
+      }
+      this.telemetryUri = 'ws:';
+    }
+    this.telemetryUri += `//${this.window.location.hostname}:${port}/api/ws/plugins/telemetry`;
+  }
+
+  public subscribe(subscriber: TelemetrySubscriber) {
+    this.isActive = true;
+    subscriber.subscriptionCommands.forEach(
+      (subscriptionCommand) => {
+        const cmdId = this.nextCmdId();
+        this.subscribersMap.set(cmdId, subscriber);
+        subscriptionCommand.cmdId = cmdId;
+        if (subscriptionCommand instanceof SubscriptionCmd) {
+          if (subscriptionCommand.getType() === TelemetryFeature.TIMESERIES) {
+            this.cmdsWrapper.tsSubCmds.push(subscriptionCommand as TimeseriesSubscriptionCmd);
+          } else {
+            this.cmdsWrapper.attrSubCmds.push(subscriptionCommand as AttributesSubscriptionCmd);
+          }
+        } else if (subscriptionCommand instanceof GetHistoryCmd) {
+          this.cmdsWrapper.historyCmds.push(subscriptionCommand);
+        }
+      }
+    );
+    this.subscribersCount++;
+    this.publishCommands();
+  }
+
+  public unsubscribe(subscriber: TelemetrySubscriber) {
+    if (this.isActive) {
+      subscriber.subscriptionCommands.forEach(
+        (subscriptionCommand) => {
+          if (subscriptionCommand instanceof SubscriptionCmd) {
+            subscriptionCommand.unsubscribe = true;
+            if (subscriptionCommand.getType() === TelemetryFeature.TIMESERIES) {
+              this.cmdsWrapper.tsSubCmds.push(subscriptionCommand as TimeseriesSubscriptionCmd);
+            } else {
+              this.cmdsWrapper.attrSubCmds.push(subscriptionCommand as AttributesSubscriptionCmd);
+            }
+          }
+          const cmdId = subscriptionCommand.cmdId;
+          if (cmdId) {
+            this.subscribersMap.delete(cmdId);
+          }
+        }
+      );
+      this.reconnectSubscribers.delete(subscriber);
+      this.subscribersCount--;
+      this.publishCommands();
+    }
+  }
+
+  private nextCmdId(): number {
+    this.lastCmdId++;
+    return this.lastCmdId;
+  }
+
+  private publishCommands() {
+    while (this.isOpened && this.cmdsWrapper.hasCommands()) {
+      this.dataStream.next(this.cmdsWrapper.preparePublishCommands(MAX_PUBLISH_COMMANDS));
+      this.checkToClose();
+    }
+    this.tryOpenSocket();
+  }
+
+  private checkToClose() {
+    if (this.subscribersCount === 0 && this.isOpened) {
+      if (!this.socketCloseTimer) {
+        this.socketCloseTimer = setTimeout(
+          () => this.closeSocket(), WS_IDLE_TIMEOUT);
+      }
+    }
+  }
+
+  private reset(close: boolean) {
+    if (this.socketCloseTimer) {
+      clearTimeout(this.socketCloseTimer);
+      this.socketCloseTimer = null;
+    }
+    this.lastCmdId = 0;
+    this.subscribersMap.clear();
+    this.subscribersCount = 0;
+    this.cmdsWrapper.clear();
+    if (close) {
+      this.closeSocket();
+    }
+  }
+
+  private closeSocket() {
+    this.isActive = false;
+    if (this.isOpened) {
+      this.dataStream.unsubscribe();
+    }
+  }
+
+  private tryOpenSocket() {
+    if (this.isActive) {
+      if (!this.isOpened && !this.isOpening) {
+        this.isOpening = true;
+        if (AuthService.isJwtTokenValid()) {
+          this.openSocket(AuthService.getJwtToken());
+        } else {
+          this.authService.refreshJwtToken().subscribe(() => {
+              this.openSocket(AuthService.getJwtToken());
+            },
+            () => {
+              this.isOpening = false;
+              this.authService.logout(true);
+            }
+          );
+        }
+      }
+      if (this.socketCloseTimer) {
+        clearTimeout(this.socketCloseTimer);
+        this.socketCloseTimer = null;
+      }
+    }
+  }
+
+  private openSocket(token: string) {
+    const uri = `${this.telemetryUri}?token=${token}`;
+    this.dataStream = webSocket(
+      {
+        url: uri,
+        openObserver: {
+          next: (e: Event) => {
+            this.onOpen();
+          }
+        },
+        closeObserver: {
+          next: (e: CloseEvent) => {
+            this.onClose(e);
+          }
+        }
+      }
+    );
+
+    this.dataStream.subscribe((message) => {
+      this.onMessage(message as SubscriptionUpdateMsg);
+    },
+    (error) => {
+      this.onError(error);
+    });
+  }
+
+  private onOpen() {
+    this.isOpening = false;
+    this.isOpened = true;
+    if (this.reconnectTimer) {
+      clearTimeout(this.reconnectTimer);
+      this.reconnectTimer = null;
+    }
+    if (this.isReconnect) {
+      this.isReconnect = false;
+      this.reconnectSubscribers.forEach(
+        (reconnectSubscriber) => {
+          reconnectSubscriber.onReconnected();
+          this.subscribe(reconnectSubscriber);
+        }
+      );
+      this.reconnectSubscribers.clear();
+    } else {
+      this.publishCommands();
+    }
+  }
+
+  private onMessage(message: SubscriptionUpdateMsg) {
+    if (message.errorCode) {
+      this.showWsError(message.errorCode, message.errorMsg);
+    } else if (message.subscriptionId) {
+      const subscriber = this.subscribersMap.get(message.subscriptionId);
+      if (subscriber) {
+        subscriber.onData(new SubscriptionUpdate(message));
+      }
+    }
+    this.checkToClose();
+  }
+
+  private onError(errorEvent) {
+    if (errorEvent) {
+      console.warn('WebSocket error event', errorEvent);
+    }
+    this.isOpening = false;
+  }
+
+  private onClose(closeEvent: CloseEvent) {
+    if (closeEvent && closeEvent.code > 1000 && closeEvent.code !== 1006) {
+      this.showWsError(closeEvent.code, closeEvent.reason);
+    }
+    this.isOpening = false;
+    this.isOpened = false;
+    if (this.isActive) {
+      if (!this.isReconnect) {
+        this.reconnectSubscribers.clear();
+        this.subscribersMap.forEach(
+          (subscriber) => {
+            this.reconnectSubscribers.add(subscriber);
+          }
+        );
+        this.reset(false);
+        this.isReconnect = true;
+      }
+      if (this.reconnectTimer) {
+        clearTimeout(this.reconnectTimer);
+      }
+      this.reconnectTimer = setTimeout(() => this.tryOpenSocket(), RECONNECT_INTERVAL);
+    }
+  }
+
+  private showWsError(errorCode: number, errorMsg: string) {
+    let message = 'WebSocket Error: ';
+    if (errorMsg) {
+      message += errorMsg;
+    } else {
+      message += `error code - ${errorCode}.`;
+    }
+    this.store.dispatch(new ActionNotificationShow(
+      {
+        message, type: 'error'
+      }));
+  }
+
+}

ui-ngx/src/app/modules/home/components/attribute/attribute-table.component.html

@@ -85,7 +85,8 @@
                     {count: dataSource.selection.selected.length}) | async }}
                 </span>
         <span fxFlex></span>
-        <button mat-button mat-icon-button [disabled]="isLoading$ | async"
+        <button [fxShow]="!isClientSideTelemetryTypeMap.get(attributeScope)"
+                mat-button mat-icon-button [disabled]="isLoading$ | async"
                 matTooltip="{{ 'action.delete' | translate }}"
                 matTooltipPosition="above"
                 (click)="deleteAttributes($event)">

ui-ngx/src/app/modules/home/components/attribute/attribute-table.component.ts

@@ -61,6 +61,7 @@ import {
   EditAttributeValuePanelData
 } from './edit-attribute-value-panel.component';
 import { ComponentPortal, PortalInjector } from '@angular/cdk/portal';
+import { TelemetryWebsocketService } from '@core/ws/telemetry-websocket.service';
 
 
 @Component({
@@ -137,6 +138,7 @@ export class AttributeTableComponent extends PageComponent implements AfterViewI
 
   constructor(protected store: Store<AppState>,
               private attributeService: AttributeService,
+              private telemetryWsService: TelemetryWebsocketService,
               public translate: TranslateService,
               public dialog: MatDialog,
               private overlay: Overlay,
@@ -146,7 +148,7 @@ export class AttributeTableComponent extends PageComponent implements AfterViewI
     this.dirtyValue = !this.activeValue;
     const sortOrder: SortOrder = { property: 'key', direction: Direction.ASC };
     this.pageLink = new PageLink(10, 0, null, sortOrder);
-    this.dataSource = new AttributeDatasource(this.attributeService, this.translate);
+    this.dataSource = new AttributeDatasource(this.attributeService, this.telemetryWsService, this.translate);
   }
 
   ngOnInit() {
@@ -333,7 +335,7 @@ export class AttributeTableComponent extends PageComponent implements AfterViewI
 
   exitWidgetMode() {
     this.mode = 'default';
-    this.reloadAttributes();
+    // this.reloadAttributes();
 
     // TODO:
   }

ui-ngx/src/app/modules/home/models/datasource/attribute-datasource.ts

@@ -26,9 +26,11 @@ import {
   AttributeData,
   AttributeScope,
   isClientSideTelemetryType,
-  TelemetryType
+  TelemetryType,
+  TelemetrySubscriber
 } from '@shared/models/telemetry/telemetry.models';
 import { AttributeService } from '@core/http/attribute.service';
+import { TelemetryWebsocketService } from '@core/ws/telemetry-websocket.service';
 
 export class AttributeDatasource implements DataSource<AttributeData> {
 
@@ -40,8 +42,10 @@ export class AttributeDatasource implements DataSource<AttributeData> {
   public selection = new SelectionModel<AttributeData>(true, []);
 
   private allAttributes: Observable<Array<AttributeData>>;
+  private telemetrySubscriber: TelemetrySubscriber;
 
   constructor(private attributeService: AttributeService,
+              private telemetryWsService: TelemetryWebsocketService,
               private translate: TranslateService) {}
 
   connect(collectionViewer: CollectionViewer): Observable<AttributeData[] | ReadonlyArray<AttributeData>> {
@@ -51,18 +55,24 @@ export class AttributeDatasource implements DataSource<AttributeData> {
   disconnect(collectionViewer: CollectionViewer): void {
     this.attributesSubject.complete();
     this.pageDataSubject.complete();
+    if (this.telemetrySubscriber) {
+      this.telemetrySubscriber.unsubscribe();
+      this.telemetrySubscriber = null;
+    }
   }
 
   loadAttributes(entityId: EntityId, attributesScope: TelemetryType,
                  pageLink: PageLink, reload: boolean = false): Observable<PageData<AttributeData>> {
     if (reload) {
       this.allAttributes = null;
+      if (this.telemetrySubscriber) {
+        this.telemetrySubscriber.unsubscribe();
+        this.telemetrySubscriber = null;
+      }
     }
+    this.selection.clear();
     const result = new ReplaySubject<PageData<AttributeData>>();
     this.fetchAttributes(entityId, attributesScope, pageLink).pipe(
-      tap(() => {
-        this.selection.clear();
-      }),
       catchError(() => of(emptyPageData<AttributeData>())),
     ).subscribe(
       (pageData) => {
@@ -85,8 +95,10 @@ export class AttributeDatasource implements DataSource<AttributeData> {
     if (!this.allAttributes) {
       let attributesObservable: Observable<Array<AttributeData>>;
       if (isClientSideTelemetryType.get(attributesScope)) {
-        attributesObservable = of([]);
+        this.telemetrySubscriber = TelemetrySubscriber.createEntityAttributesSubscription(
-        // TODO:
+          this.telemetryWsService, entityId, attributesScope);
+        this.telemetrySubscriber.subscribe();
+        attributesObservable = this.telemetrySubscriber.attributeData$();
       } else {
         attributesObservable = this.attributeService.getEntityAttributes(entityId, attributesScope as AttributeScope);
       }

ui-ngx/src/app/modules/home/pages/asset/asset-tabs.component.html

@@ -15,6 +15,21 @@
     limitations under the License.
 
 -->
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.attributes' | translate }}" #attributesTab="matTab">
+  <tb-attribute-table [active]="attributesTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="attributeScopes.SERVER_SCOPE">
+  </tb-attribute-table>
+</mat-tab>
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.latest-telemetry' | translate }}" #telemetryTab="matTab">
+  <tb-attribute-table [active]="telemetryTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="latestTelemetryTypes.LATEST_TELEMETRY"
+                      disableAttributeScopeSelection>
+  </tb-attribute-table>
+</mat-tab>
 <mat-tab *ngIf="entity"
          label="{{ 'alarm.alarms' | translate }}" #alarmsTab="matTab">
   <tb-alarm-table [active]="alarmsTab.isActive" [entityId]="entity.id"></tb-alarm-table>

ui-ngx/src/app/modules/home/pages/customer/customer-tabs.component.html

@@ -15,6 +15,21 @@
     limitations under the License.
 
 -->
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.attributes' | translate }}" #attributesTab="matTab">
+  <tb-attribute-table [active]="attributesTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="attributeScopes.SERVER_SCOPE">
+  </tb-attribute-table>
+</mat-tab>
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.latest-telemetry' | translate }}" #telemetryTab="matTab">
+  <tb-attribute-table [active]="telemetryTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="latestTelemetryTypes.LATEST_TELEMETRY"
+                      disableAttributeScopeSelection>
+  </tb-attribute-table>
+</mat-tab>
 <mat-tab *ngIf="entity"
          label="{{ 'alarm.alarms' | translate }}" #alarmsTab="matTab">
   <tb-alarm-table [active]="alarmsTab.isActive" [entityId]="entity.id"></tb-alarm-table>

ui-ngx/src/app/modules/home/pages/entity-view/entity-view-tabs.component.html

@@ -15,6 +15,21 @@
     limitations under the License.
 
 -->
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.attributes' | translate }}" #attributesTab="matTab">
+  <tb-attribute-table [active]="attributesTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="attributeScopes.CLIENT_SCOPE">
+  </tb-attribute-table>
+</mat-tab>
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.latest-telemetry' | translate }}" #telemetryTab="matTab">
+  <tb-attribute-table [active]="telemetryTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="latestTelemetryTypes.LATEST_TELEMETRY"
+                      disableAttributeScopeSelection>
+  </tb-attribute-table>
+</mat-tab>
 <mat-tab *ngIf="entity"
          label="{{ 'alarm.alarms' | translate }}" #alarmsTab="matTab">
   <tb-alarm-table [active]="alarmsTab.isActive" [entityId]="entity.id"></tb-alarm-table>

ui-ngx/src/app/modules/home/pages/rulechain/rulechain-tabs.component.html

@@ -15,6 +15,21 @@
     limitations under the License.
 
 -->
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.attributes' | translate }}" #attributesTab="matTab">
+  <tb-attribute-table [active]="attributesTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="attributeScopes.SERVER_SCOPE">
+  </tb-attribute-table>
+</mat-tab>
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.latest-telemetry' | translate }}" #telemetryTab="matTab">
+  <tb-attribute-table [active]="telemetryTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="latestTelemetryTypes.LATEST_TELEMETRY"
+                      disableAttributeScopeSelection>
+  </tb-attribute-table>
+</mat-tab>
 <mat-tab *ngIf="entity"
          label="{{ 'alarm.alarms' | translate }}" #alarmsTab="matTab">
   <tb-alarm-table [active]="alarmsTab.isActive" [entityId]="entity.id"></tb-alarm-table>

ui-ngx/src/app/modules/home/pages/tenant/tenant-tabs.component.html

@@ -15,6 +15,21 @@
     limitations under the License.
 
 -->
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.attributes' | translate }}" #attributesTab="matTab">
+  <tb-attribute-table [active]="attributesTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="attributeScopes.SERVER_SCOPE">
+  </tb-attribute-table>
+</mat-tab>
+<mat-tab *ngIf="entity"
+         label="{{ 'attribute.latest-telemetry' | translate }}" #telemetryTab="matTab">
+  <tb-attribute-table [active]="telemetryTab.isActive"
+                      [entityId]="entity.id"
+                      [defaultAttributeScope]="latestTelemetryTypes.LATEST_TELEMETRY"
+                      disableAttributeScopeSelection>
+  </tb-attribute-table>
+</mat-tab>
 <mat-tab *ngIf="entity"
          label="{{ 'alarm.alarms' | translate }}" #alarmsTab="matTab">
   <tb-alarm-table [active]="alarmsTab.isActive" [entityId]="entity.id"></tb-alarm-table>

ui-ngx/src/app/shared/models/telemetry/telemetry.models.ts

@@ -15,7 +15,11 @@
 ///
 
 
-import { AlarmSeverity } from '@shared/models/alarm.models';
+import { EntityType } from '@shared/models/entity-type.models';
+import { AggregationType } from '../time/time.models';
+import { Observable, ReplaySubject, Subject } from 'rxjs';
+import { EntityId } from '@shared/models/id/entity-id';
+import { map } from 'rxjs/operators';
 
 export enum DataKeyType {
   timeseries = 'timeseries',
@@ -34,6 +38,11 @@ export enum AttributeScope {
   SHARED_SCOPE = 'SHARED_SCOPE'
 }
 
+export enum TelemetryFeature {
+  ATTRIBUTES = 'ATTRIBUTES',
+  TIMESERIES = 'TIMESERIES'
+}
+
 export type TelemetryType = LatestTelemetry | AttributeScope;
 
 export const telemetryTypeTranslations = new Map<TelemetryType, string>(
@@ -59,3 +68,222 @@ export interface AttributeData {
   key: string;
   value: any;
 }
+
+export interface TelemetryPluginCmd {
+  cmdId: number;
+  keys: string;
+}
+
+export abstract class SubscriptionCmd implements TelemetryPluginCmd {
+  cmdId: number;
+  keys: string;
+  entityType: EntityType;
+  entityId: string;
+  scope?: AttributeScope;
+  unsubscribe: boolean;
+  abstract getType(): TelemetryFeature;
+}
+
+export class AttributesSubscriptionCmd extends SubscriptionCmd {
+  getType() {
+    return TelemetryFeature.ATTRIBUTES;
+  }
+}
+
+export class TimeseriesSubscriptionCmd extends SubscriptionCmd {
+  startTs: number;
+  timeWindow: number;
+  interval: number;
+  limit: number;
+  agg: AggregationType;
+
+  getType() {
+    return TelemetryFeature.TIMESERIES;
+  }
+}
+
+export class GetHistoryCmd implements TelemetryPluginCmd {
+  cmdId: number;
+  keys: string;
+  entityType: EntityType;
+  entityId: string;
+  startTs: number;
+  endTs: number;
+  interval: number;
+  limit: number;
+  agg: AggregationType;
+}
+
+export class TelemetryPluginCmdsWrapper {
+  attrSubCmds: Array<AttributesSubscriptionCmd>;
+  tsSubCmds: Array<TimeseriesSubscriptionCmd>;
+  historyCmds: Array<GetHistoryCmd>;
+
+  constructor() {
+    this.attrSubCmds = [];
+    this.tsSubCmds = [];
+    this.historyCmds = [];
+  }
+
+  public hasCommands(): boolean {
+    return this.tsSubCmds.length > 0 ||
+      this.historyCmds.length > 0 ||
+      this.attrSubCmds.length > 0;
+  }
+
+  public clear() {
+    this.attrSubCmds.length = 0;
+    this.tsSubCmds.length = 0;
+    this.historyCmds.length = 0;
+  }
+
+  public preparePublishCommands(maxCommands: number): TelemetryPluginCmdsWrapper {
+    const preparedWrapper = new TelemetryPluginCmdsWrapper();
+    let leftCount = maxCommands;
+    preparedWrapper.tsSubCmds = this.popCmds(this.tsSubCmds, leftCount);
+    leftCount -= preparedWrapper.tsSubCmds.length;
+    preparedWrapper.historyCmds = this.popCmds(this.historyCmds, leftCount);
+    leftCount -= preparedWrapper.historyCmds.length;
+    preparedWrapper.attrSubCmds = this.popCmds(this.attrSubCmds, leftCount);
+    return preparedWrapper;
+  }
+
+  private popCmds<T extends TelemetryPluginCmd>(cmds: Array<T>, leftCount: number): Array<T> {
+    const toPublish = Math.min(cmds.length, leftCount);
+    if (toPublish > 0) {
+      return cmds.splice(0, toPublish);
+    } else {
+      return [];
+    }
+  }
+}
+
+export interface SubscriptionUpdateMsg {
+  subscriptionId: number;
+  errorCode: number;
+  errorMsg: string;
+  data: {[key: string]: [number, string][]};
+}
+
+export class SubscriptionUpdate implements SubscriptionUpdateMsg {
+  subscriptionId: number;
+  errorCode: number;
+  errorMsg: string;
+  data: {[key: string]: [number, string][]};
+
+  constructor(msg: SubscriptionUpdateMsg) {
+    this.subscriptionId = msg.subscriptionId;
+    this.errorCode = msg.errorCode;
+    this.errorMsg = msg.errorMsg;
+    this.data = msg.data;
+  }
+
+  public prepareData(keys: string[]) {
+    if (!this.data) {
+      this.data = {};
+    }
+    if (keys) {
+      keys.forEach((key) => {
+        if (!this.data[key]) {
+          this.data[key] = [];
+        }
+      });
+    }
+  }
+
+  public updateAttributeData(origData: Array<AttributeData>): Array<AttributeData> {
+    for (const key of Object.keys(this.data)) {
+      const keyData = this.data[key];
+      if (keyData.length) {
+        const existing = origData.find((data) => data.key === key);
+        if (existing) {
+          existing.lastUpdateTs = keyData[0][0];
+          existing.value = keyData[0][1];
+        } else {
+          origData.push(
+            {
+              key,
+              lastUpdateTs: keyData[0][0],
+              value: keyData[0][1]
+            }
+          );
+        }
+      }
+    }
+    return origData;
+  }
+}
+
+export interface TelemetryService {
+  subscribe(subscriber: TelemetrySubscriber);
+  unsubscribe(subscriber: TelemetrySubscriber);
+}
+
+export class TelemetrySubscriber {
+
+  private dataSubject = new ReplaySubject<SubscriptionUpdate>();
+  private reconnectSubject = new Subject();
+
+  public subscriptionCommands: Array<TelemetryPluginCmd>;
+
+  public data$ = this.dataSubject.asObservable();
+  public reconnect$ = this.reconnectSubject.asObservable();
+
+  public static createEntityAttributesSubscription(telemetryService: TelemetryService,
+                                                   entityId: EntityId, attributeScope: TelemetryType,
+                                                   keys: string[] = null): TelemetrySubscriber {
+    let subscriptionCommand: SubscriptionCmd;
+    if (attributeScope === LatestTelemetry.LATEST_TELEMETRY) {
+      subscriptionCommand = new TimeseriesSubscriptionCmd();
+    } else {
+      subscriptionCommand = new AttributesSubscriptionCmd();
+    }
+    subscriptionCommand.entityType = entityId.entityType as EntityType;
+    subscriptionCommand.entityId = entityId.id;
+    subscriptionCommand.scope = attributeScope as AttributeScope;
+    if (keys) {
+      subscriptionCommand.keys = keys.join(',');
+    }
+    const subscriber = new TelemetrySubscriber(telemetryService);
+    subscriber.subscriptionCommands.push(subscriptionCommand);
+    return subscriber;
+  }
+
+  constructor(private telemetryService: TelemetryService) {
+    this.subscriptionCommands = [];
+  }
+
+  public subscribe() {
+    this.telemetryService.subscribe(this);
+  }
+
+  public unsubscribe() {
+    this.telemetryService.unsubscribe(this);
+    this.dataSubject.complete();
+    this.reconnectSubject.complete();
+  }
+
+  public onData(message: SubscriptionUpdate) {
+    const cmdId = message.subscriptionId;
+    let keys: string[];
+    const cmd = this.subscriptionCommands.find((command) => command.cmdId === cmdId);
+    if (cmd) {
+      if (cmd.keys && cmd.keys.length) {
+        keys = cmd.keys.split(',');
+      }
+    }
+    message.prepareData(keys);
+    this.dataSubject.next(message);
+  }
+
+  public onReconnected() {
+    this.reconnectSubject.next();
+  }
+
+  public attributeData$(): Observable<Array<AttributeData>> {
+    const attributeData = new Array<AttributeData>();
+    return this.data$.pipe(
+      map((message) => message.updateAttributeData(attributeData))
+    );
+  }
+}