You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2022/02/06 18:38:40 UTC
[incubator-streampipes] 05/07: [STREAMPIPES-509] Refactor asset dashboard to receive data from data lake
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch experimental-module-federation-494
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 7781208d87147b51291076f1ca2a21410bfbfbc4
Author: Dominik Riemer <do...@gmail.com>
AuthorDate: Sun Feb 6 19:19:21 2022 +0100
[STREAMPIPES-509] Refactor asset dashboard to receive data from data lake
---
.../app-asset-monitoring.module.ts | 51 ++++++++-----------
.../components/view-asset/view-asset.component.ts | 57 +++++++++++++++++-----
.../add-pipeline-dialog.component.html | 6 +--
.../add-pipeline/add-pipeline-dialog.component.ts | 32 ++++++++----
.../model/selected-visualization-data.model.ts | 3 +-
.../app-asset-monitoring/services/shape.service.ts | 3 +-
.../services/websocket.service.ts | 40 ---------------
7 files changed, 90 insertions(+), 102 deletions(-)
diff --git a/ui/src/app/app-asset-monitoring/app-asset-monitoring.module.ts b/ui/src/app/app-asset-monitoring/app-asset-monitoring.module.ts
index f4de74d..c3fd643 100644
--- a/ui/src/app/app-asset-monitoring/app-asset-monitoring.module.ts
+++ b/ui/src/app/app-asset-monitoring/app-asset-monitoring.module.ts
@@ -16,32 +16,29 @@
*
*/
-import {NgModule} from '@angular/core';
-import {FlexLayoutModule} from '@angular/flex-layout';
-import {CommonModule} from '@angular/common';
+import { NgModule } from '@angular/core';
+import { FlexLayoutModule } from '@angular/flex-layout';
+import { CommonModule } from '@angular/common';
-import {AppAssetMonitoringComponent} from './app-asset-monitoring.component';
+import { AppAssetMonitoringComponent } from './app-asset-monitoring.component';
-import {CustomMaterialModule} from '../CustomMaterial/custom-material.module';
+import { CustomMaterialModule } from '../CustomMaterial/custom-material.module';
-import {ViewAssetComponent} from "./components/view-asset/view-asset.component";
-import {CreateAssetComponent} from "./components/create-asset/create-asset.component";
-import {WebsocketService} from "./services/websocket.service";
-import {AddPipelineDialogComponent} from "./dialog/add-pipeline/add-pipeline-dialog.component";
-import {RestService} from './services/rest.service';
-import {MatFormFieldModule} from "@angular/material/form-field";
-import {MatGridListModule} from "@angular/material/grid-list";
-import {MatInputModule} from "@angular/material/input";
-import {ElementIconText} from "../services/get-element-icon-text.service";
-import {FormsModule} from "@angular/forms";
-import {ColorPickerModule} from "ngx-color-picker";
-import {ShapeService} from "./services/shape.service";
-import {SaveDashboardDialogComponent} from "./dialog/save-dashboard/save-dashboard-dialog.component";
-import {AssetDashboardOverviewComponent} from "./components/dashboard-overview/dashboard-overview.component";
-import {InjectableRxStompConfig, RxStompService, rxStompServiceFactory} from "@stomp/ng2-stompjs";
-import {streamPipesStompConfig} from "../dashboard/services/websocket.config";
+import { ViewAssetComponent } from './components/view-asset/view-asset.component';
+import { CreateAssetComponent } from './components/create-asset/create-asset.component';
+import { AddPipelineDialogComponent } from './dialog/add-pipeline/add-pipeline-dialog.component';
+import { RestService } from './services/rest.service';
+import { MatFormFieldModule } from '@angular/material/form-field';
+import { MatGridListModule } from '@angular/material/grid-list';
+import { MatInputModule } from '@angular/material/input';
+import { ElementIconText } from '../services/get-element-icon-text.service';
+import { FormsModule } from '@angular/forms';
+import { ColorPickerModule } from 'ngx-color-picker';
+import { ShapeService } from './services/shape.service';
+import { SaveDashboardDialogComponent } from './dialog/save-dashboard/save-dashboard-dialog.component';
+import { AssetDashboardOverviewComponent } from './components/dashboard-overview/dashboard-overview.component';
import { AddLinkDialogComponent } from './dialog/add-link/add-link-dialog.component';
-import {DashboardModule} from "../dashboard/dashboard.module";
+import { DashboardModule } from '../dashboard/dashboard.module';
@NgModule({
imports: [
@@ -65,18 +62,8 @@ import {DashboardModule} from "../dashboard/dashboard.module";
AssetDashboardOverviewComponent
],
providers: [
- WebsocketService,
RestService,
ShapeService,
- {
- provide: InjectableRxStompConfig,
- useValue: streamPipesStompConfig
- },
- {
- provide: RxStompService,
- useFactory: rxStompServiceFactory,
- deps: [InjectableRxStompConfig]
- },
ElementIconText
],
entryComponents: [
diff --git a/ui/src/app/app-asset-monitoring/components/view-asset/view-asset.component.ts b/ui/src/app/app-asset-monitoring/components/view-asset/view-asset.component.ts
index 1fe5319..8a88fbf 100644
--- a/ui/src/app/app-asset-monitoring/components/view-asset/view-asset.component.ts
+++ b/ui/src/app/app-asset-monitoring/components/view-asset/view-asset.component.ts
@@ -16,14 +16,15 @@
*
*/
-import { AfterViewInit, Component, EventEmitter, Input, OnInit, Output } from '@angular/core';
+import { AfterViewInit, Component, EventEmitter, Input, OnDestroy, OnInit, Output } from '@angular/core';
import Konva from 'konva';
-import { WebsocketService } from '../../services/websocket.service';
import { DashboardConfiguration } from '../../model/dashboard-configuration.model';
import { RestService } from '../../services/rest.service';
import { DashboardService } from '../../../dashboard/services/dashboard.service';
-import { DashboardItem } from '@streampipes/platform-services';
+import { DatalakeQueryParameterBuilder, DatalakeRestService, SpQueryResult } from '@streampipes/platform-services';
+import { Subscription, timer } from 'rxjs';
+import { switchMap } from 'rxjs/operators';
interface Window {
Image: any;
@@ -36,7 +37,7 @@ declare const window: Window;
templateUrl: './view-asset.component.html',
styleUrls: ['./view-asset.component.css']
})
-export class ViewAssetComponent implements OnInit, AfterViewInit {
+export class ViewAssetComponent implements OnInit, AfterViewInit, OnDestroy {
@Input() dashboardConfig: DashboardConfiguration;
@Output() dashboardClosed = new EventEmitter<boolean>();
@@ -46,12 +47,11 @@ export class ViewAssetComponent implements OnInit, AfterViewInit {
mainLayer: any;
backgroundImageLayer: any;
- dashboardItem: DashboardItem;
- widgetLoaded = false;
+ subscriptions: Subscription[] = [];
- constructor(private websocketService: WebsocketService,
- private restService: RestService,
- private dashboardService: DashboardService) {
+ constructor(private restService: RestService,
+ private dashboardService: DashboardService,
+ private dataLakeRestService: DatalakeRestService) {
}
@@ -97,13 +97,40 @@ export class ViewAssetComponent implements OnInit, AfterViewInit {
const dynamicShapes = this.mainCanvasStage.find('.dynamic-text');
dynamicShapes.forEach(ds => {
const monitoredField = ds.text();
- this.websocketService.connect(ds.attrs.brokerUrl, ds.attrs.topic).subscribe(msg => {
- ds.text(msg[monitoredField]);
- this.mainCanvasStage.draw();
- });
+ const measurement = ds.attrs.dataLakeMeasure;
+ const subscription = timer(0, 2000).pipe(
+ switchMap(() => this.dataLakeRestService.getData(measurement, this.buildQuery())))
+ .subscribe(queryResult => {
+ this.handleResponse(ds, monitoredField, queryResult);
+ });
+ this.subscriptions.push(subscription);
});
}
+ handleResponse(ds: any,
+ monitoredField: string,
+ queryResult: SpQueryResult): void {
+ if (queryResult.total > 0) {
+ if (queryResult.allDataSeries.length === 1) {
+ const series = queryResult.allDataSeries[0];
+ if (series.rows.length > 0) {
+ const row = series.rows[0];
+ const event = {};
+ series.headers.forEach((fieldName, index) => {
+ event[fieldName] = row[index];
+ });
+ ds.text(event[monitoredField]);
+ this.mainCanvasStage.draw();
+ }
+ }
+ }
+ }
+
+ buildQuery() {
+ const queryBuilder = DatalakeQueryParameterBuilder.create();
+ return queryBuilder.withLimit(1).withOrdering('DESC').build();
+ }
+
showImage() {
const image = new window.Image();
image.src = this.restService.getImageUrl(this.dashboardConfig.imageInfo.imageName);
@@ -123,4 +150,8 @@ export class ViewAssetComponent implements OnInit, AfterViewInit {
this.editDashboardEmitter.emit(this.dashboardConfig);
}
+ ngOnDestroy(): void {
+ this.subscriptions.forEach(s => s.unsubscribe());
+ }
+
}
diff --git a/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.html b/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.html
index e5748ce..4df8b47 100644
--- a/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.html
+++ b/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.html
@@ -27,10 +27,10 @@
<mat-list-item *ngFor="let pipeline of visualizablePipelines"
(click)="selectPipeline(pipeline)" class="list-item">
<div mat-list-avatar
- class="pipeline-avatar sp-accent-bg">{{iconText(pipeline.visualizationName)}}
+ class="pipeline-avatar sp-accent-bg">{{iconText(pipeline.pipelineName)}}
</div>
<h4 mat-line>{{pipeline.pipelineName}}</h4>
- <p mat-line>{{pipeline.visualizationName}} </p>
+ <p mat-line>{{pipeline.measureName}} </p>
</mat-list-item>
</mat-list>
</div>
@@ -43,7 +43,7 @@
<mat-radio-group fxLayout="column" class="example-radio-group"
[(ngModel)]="selectedMeasurement">
<mat-radio-button class="example-radio-button"
- *ngFor="let prop of selectedVisualization.schema.eventProperties"
+ *ngFor="let prop of selectedVisualization.eventSchema.eventProperties"
[value]="prop.runtimeName">
{{prop.runtimeName}}
</mat-radio-button>
diff --git a/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.ts b/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.ts
index 294f79d..5803c06 100644
--- a/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.ts
+++ b/ui/src/app/app-asset-monitoring/dialog/add-pipeline/add-pipeline-dialog.component.ts
@@ -21,9 +21,9 @@ import { RestApi } from '../../../services/rest-api.service';
import { RestService } from '../../services/rest.service';
import { ElementIconText } from '../../../services/get-element-icon-text.service';
import { SelectedVisualizationData } from '../../model/selected-visualization-data.model';
-import { DashboardService } from '../../../dashboard/services/dashboard.service';
import { DialogRef } from '../../../core-ui/dialog/base-dialog/dialog-ref';
-import { VisualizablePipeline } from '@streampipes/platform-services';
+import { DataLakeMeasure, DatalakeRestService, DataViewDataExplorerService } from '@streampipes/platform-services';
+import { zip } from 'rxjs';
@Component({
selector: 'sp-add-pipeline-dialog-component',
@@ -46,9 +46,9 @@ export class AddPipelineDialogComponent implements OnInit {
description: 'Choose label'
}];
- visualizablePipelines: VisualizablePipeline[] = [];
+ visualizablePipelines: DataLakeMeasure[] = [];
- selectedVisualization: VisualizablePipeline;
+ selectedVisualization: DataLakeMeasure;
selectedType: any;
selectedMeasurement: any;
page: any = 'select-pipeline';
@@ -64,14 +64,27 @@ export class AddPipelineDialogComponent implements OnInit {
private dialogRef: DialogRef<AddPipelineDialogComponent>,
private restApi: RestApi,
private restService: RestService,
- private dashboardService: DashboardService,
+ private dataLakeRestService: DatalakeRestService,
+ private dataExplorerService: DataViewDataExplorerService,
public elementIconText: ElementIconText) {
}
ngOnInit() {
- this.dashboardService.getVisualizablePipelines().subscribe(visualizations => {
- this.visualizablePipelines = visualizations;
- });
+ this.loadVisualizablePipelines();
+ }
+
+ loadVisualizablePipelines() {
+ zip(this.dataExplorerService.getAllPersistedDataStreams(), this.dataLakeRestService.getAllMeasurementSeries())
+ .subscribe(res => {
+ const visualizablePipelines = res[0];
+ visualizablePipelines.forEach(p => {
+ const measurement = res[1].find(m => {
+ return m.measureName === p.measureName;
+ });
+ p.eventSchema = measurement.eventSchema;
+ });
+ this.visualizablePipelines = visualizablePipelines;
+ });
}
onCancel(): void {
@@ -122,8 +135,7 @@ export class AddPipelineDialogComponent implements OnInit {
selectedConfig.measurement = this.selectedMeasurement;
selectedConfig.visualizationId = this.selectedVisualization.pipelineId;
selectedConfig.label = this.selectedLabel;
- selectedConfig.brokerUrl = (this.selectedVisualization as any).broker;
- selectedConfig.topic = this.selectedVisualization.topic;
+ selectedConfig.dataLakeMeasure = this.selectedVisualization.measureName;
this.dialogRef.close(selectedConfig);
}
diff --git a/ui/src/app/app-asset-monitoring/model/selected-visualization-data.model.ts b/ui/src/app/app-asset-monitoring/model/selected-visualization-data.model.ts
index b79cb68..611bc0d 100644
--- a/ui/src/app/app-asset-monitoring/model/selected-visualization-data.model.ts
+++ b/ui/src/app/app-asset-monitoring/model/selected-visualization-data.model.ts
@@ -24,8 +24,7 @@ export interface SelectedVisualizationData {
visualizationId: string;
measurement: string;
label: string;
- brokerUrl: string;
- topic: string;
+ dataLakeMeasure: string;
}
export interface HyperlinkConfig {
diff --git a/ui/src/app/app-asset-monitoring/services/shape.service.ts b/ui/src/app/app-asset-monitoring/services/shape.service.ts
index 3e61250..01c2c16 100644
--- a/ui/src/app/app-asset-monitoring/services/shape.service.ts
+++ b/ui/src/app/app-asset-monitoring/services/shape.service.ts
@@ -125,8 +125,7 @@ export class ShapeService {
if (dynamicContent) {
textSettings.name = 'dynamic-text';
- textSettings.brokerUrl = config.brokerUrl;
- textSettings.topic = config.topic;
+ textSettings.dataLakeMeasure = config.dataLakeMeasure;
textSettings.fontSize = '30';
}
diff --git a/ui/src/app/app-asset-monitoring/services/websocket.service.ts b/ui/src/app/app-asset-monitoring/services/websocket.service.ts
deleted file mode 100644
index 4a3de23..0000000
--- a/ui/src/app/app-asset-monitoring/services/websocket.service.ts
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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 {Injectable} from "@angular/core";
-import {Observable, Subscription} from "rxjs";
-import {Message} from "@stomp/stompjs";
-import {RxStompService} from "@stomp/ng2-stompjs";
-
-@Injectable()
-export class WebsocketService {
-
- subscription: Subscription;
-
- constructor(private rxStompService: RxStompService) {
- }
-
- connect(url, topic): Observable<any> {
- return new Observable<any>(observable => {
- this.subscription = this.rxStompService.watch("/topic/" +topic).subscribe((message: Message) => {
- observable.next(JSON.parse(message.body));
- });
- });
- }
-
-}
\ No newline at end of file