This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch improve-adapter-pipeline-preview in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 329ea4ce0532af47eaa0bc4a6141eddaf82e1a9c Author: Dominik Riemer <[email protected]> AuthorDate: Sat Aug 3 12:28:49 2024 +0200 chore: Provide adapter and pipeline preview as stream --- .../messaging/kafka/SpKafkaConsumer.java | 1 - .../streampipes/manager/operations/Operations.java | 6 -- .../manager/preview/PipelinePreview.java | 23 +++++--- ...her.java => DataStreamRuntimeInfoProvider.java} | 53 ++++++------------ .../runtime/RateLimitedRuntimeInfoProvider.java | 60 ++++++++++++++++++++ .../rest/impl/PipelineElementPreview.java | 18 ++++-- .../rest/impl/PipelineElementRuntimeInfo.java | 19 +++---- .../adapter-started-preview.component.html | 1 - .../adapter-started-preview.component.ts | 9 +-- ui/src/app/connect/services/rest.service.ts | 11 +++- .../pipeline-element-runtime-info.component.ts | 48 +++++----------- .../pipeline-element-preview.component.html | 5 +- .../pipeline-element-preview.component.ts | 64 ++++++++++++++++------ .../components/pipeline/pipeline.component.ts | 6 ++ ui/src/app/editor/services/editor.service.ts | 17 +++--- 15 files changed, 202 insertions(+), 139 deletions(-) diff --git a/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaConsumer.java b/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaConsumer.java index e999cfbc4a..fcbd38224f 100644 --- a/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaConsumer.java +++ b/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaConsumer.java @@ -104,7 +104,6 @@ public class SpKafkaConsumer implements EventConsumer, Runnable, ConsumerRecords<byte[], byte[]> records = consumer.poll(duration); records.forEach(record -> eventProcessor.onEvent(record.value())); } - LOG.info("Closing Kafka Consumer."); consumer.close(); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java index 81d6215c88..f2addd2b20 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java @@ -20,12 +20,10 @@ package org.apache.streampipes.manager.operations; import org.apache.streampipes.commons.exceptions.NoSuitableSepasAvailableException; import org.apache.streampipes.commons.exceptions.SepaParseException; -import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.manager.execution.PipelineExecutor; import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2; import org.apache.streampipes.manager.recommender.ElementRecommender; import org.apache.streampipes.manager.remote.ContainerProvidedOptionsHandler; -import org.apache.streampipes.manager.runtime.PipelineElementRuntimeInfoFetcher; import org.apache.streampipes.manager.storage.PipelineStorageService; import org.apache.streampipes.manager.template.PipelineTemplateGenerator; import org.apache.streampipes.manager.template.PipelineTemplateInvocationGenerator; @@ -116,10 +114,6 @@ public class Operations { return new ContainerProvidedOptionsHandler().fetchRemoteOptions(request); } - public static String getRuntimeInfo(SpDataStream spDataStream) throws SpRuntimeException { - return PipelineElementRuntimeInfoFetcher.INSTANCE.getCurrentData(spDataStream); - } - public static List<PipelineTemplateDescription> getAllPipelineTemplates() { return new PipelineTemplateGenerator().getAllPipelineTemplates(); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java index 903ecbeb75..23aedaa8f5 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java @@ -23,7 +23,6 @@ import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpoi import org.apache.streampipes.manager.execution.http.DetachHttpRequest; import org.apache.streampipes.manager.execution.http.InvokeHttpRequest; import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2; -import org.apache.streampipes.manager.operations.Operations; import org.apache.streampipes.model.SpDataStream; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.base.NamedStreamPipesEntity; @@ -31,6 +30,9 @@ import org.apache.streampipes.model.graph.DataProcessorInvocation; import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.model.preview.PipelinePreviewModel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.ArrayList; import java.util.List; import java.util.Optional; @@ -39,17 +41,21 @@ import java.util.stream.Collectors; public class PipelinePreview { + private static final Logger LOG = LoggerFactory.getLogger(PipelinePreview.class); + public PipelinePreviewModel initiatePreview(Pipeline pipeline) { String previewId = generatePreviewId(); pipeline.setActions(new ArrayList<>()); - List<NamedStreamPipesEntity> pipelineElements = new PipelineVerificationHandlerV2(pipeline) + List<NamedStreamPipesEntity> pipelineElements = new ArrayList<>( + new PipelineVerificationHandlerV2(pipeline) .verifyAndBuildGraphs(true) - .stream() - .collect(Collectors.toList()); + ); invokeGraphs(filter(pipelineElements)); storeGraphs(previewId, pipelineElements); + LOG.info("Preview pipeline {} started", previewId); + return makePreviewModel(previewId, pipelineElements); } @@ -57,9 +63,10 @@ public class PipelinePreview { List<NamedStreamPipesEntity> graphs = ActivePipelinePreviews.INSTANCE.getInvocationGraphs(previewId); detachGraphs(filter(graphs)); deleteGraphs(previewId); + LOG.info("Preview pipeline {} stopped", previewId); } - public String getPipelineElementPreview(String previewId, + public SpDataStream getPipelineElementPreviewStream(String previewId, String pipelineElementDomId) throws IllegalArgumentException { Optional<NamedStreamPipesEntity> graphOpt = ActivePipelinePreviews .INSTANCE @@ -68,9 +75,9 @@ public class PipelinePreview { if (graphOpt.isPresent()) { NamedStreamPipesEntity graph = graphOpt.get(); if (graph instanceof DataProcessorInvocation) { - return Operations.getRuntimeInfo(((DataProcessorInvocation) graph).getOutputStream()); + return ((DataProcessorInvocation) graph).getOutputStream(); } else if (graph instanceof SpDataStream) { - return Operations.getRuntimeInfo((SpDataStream) graph); + return ((SpDataStream) graph); } else { throw new IllegalArgumentException("Requested pipeline element is not a data processor"); } @@ -93,7 +100,7 @@ public class PipelinePreview { g.setSelectedEndpointUrl(findSelectedEndpoint(g)); new InvokeHttpRequest().execute(g, g.getSelectedEndpointUrl(), null); } catch (NoServiceEndpointsAvailableException e) { - e.printStackTrace(); + LOG.warn("No endpoint found for pipeline element {}", g.getAppId()); } }); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/PipelineElementRuntimeInfoFetcher.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/DataStreamRuntimeInfoProvider.java similarity index 67% rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/PipelineElementRuntimeInfoFetcher.java rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/DataStreamRuntimeInfoProvider.java index 8c522477ac..d54e5f7a35 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/PipelineElementRuntimeInfoFetcher.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/DataStreamRuntimeInfoProvider.java @@ -31,25 +31,26 @@ import org.slf4j.LoggerFactory; import java.util.HashMap; import java.util.Map; -import java.util.concurrent.TimeUnit; -public enum PipelineElementRuntimeInfoFetcher { - INSTANCE; +public class DataStreamRuntimeInfoProvider { - private static final Logger LOG = LoggerFactory.getLogger(PipelineElementRuntimeInfoFetcher.class); + private static final Logger LOG = LoggerFactory.getLogger(DataStreamRuntimeInfoProvider.class); - private static final int FETCH_INTERVAL_MS = 300; private final Map<String, SpDataFormatConverter> converterMap; private final Environment env; + private final SpDataStream dataStream; + private EventConsumer consumer; + private String latestEvent; - PipelineElementRuntimeInfoFetcher() { + public DataStreamRuntimeInfoProvider(SpDataStream dataStream) { + this.dataStream = dataStream; this.converterMap = new HashMap<>(); this.env = Environments.getEnvironment(); } - public String getCurrentData(SpDataStream spDataStream) throws SpRuntimeException { - var topic = getOutputTopic(spDataStream); - var protocol = spDataStream.getEventGrounding().getTransportProtocol(); + public void startConsuming() throws SpRuntimeException { + var topic = getOutputTopic(dataStream); + var protocol = dataStream.getEventGrounding().getTransportProtocol(); if (env.getSpDebug().getValueOrDefault()) { protocol.setBrokerHostname("localhost"); if (protocol instanceof KafkaTransportProtocol) { @@ -59,19 +60,18 @@ public enum PipelineElementRuntimeInfoFetcher { if (!converterMap.containsKey(topic)) { this.converterMap.put(topic, - new SpDataFormatConverterGenerator(getTransportFormat(spDataStream)).makeConverter()); + new SpDataFormatConverterGenerator(getTransportFormat(dataStream)).makeConverter()); } var converter = converterMap.get(topic); var protocolDefinitionOpt = SpProtocolManager .INSTANCE - .findDefinition(spDataStream.getEventGrounding().getTransportProtocol()); + .findDefinition(dataStream.getEventGrounding().getTransportProtocol()); if (protocolDefinitionOpt.isPresent()) { - var consumer = protocolDefinitionOpt.get().getConsumer(protocol); - return getLatestEvent(consumer, converter); - + consumer = protocolDefinitionOpt.get().getConsumer(protocol); + consumer.connect(event -> this.latestEvent = converter.convert(event)); } else { LOG.error("Error while fetching data for preview - protocol {} not found - did you register the protocol? ", protocol.getClass().getCanonicalName()); @@ -91,28 +91,11 @@ public enum PipelineElementRuntimeInfoFetcher { .getActualTopicName(); } - private void waitForEvent(String[] result) { - long timeout = 0; - while (result[0] == null && timeout < 6000) { - try { - TimeUnit.MILLISECONDS.sleep(FETCH_INTERVAL_MS); - timeout = timeout + 300; - } catch (InterruptedException e) { - e.printStackTrace(); - } - } + public String getLatestEvent() { + return latestEvent; } - private String getLatestEvent(EventConsumer consumer, - SpDataFormatConverter converter) { - final String[] result = {null}; - consumer.connect(event -> { - result[0] = converter.convert(event); - consumer.disconnect(); - }); - - waitForEvent(result); - - return result[0]; + public void close() { + consumer.disconnect(); } } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/RateLimitedRuntimeInfoProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/RateLimitedRuntimeInfoProvider.java new file mode 100644 index 0000000000..a7474d74cb --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/RateLimitedRuntimeInfoProvider.java @@ -0,0 +1,60 @@ +/* + * 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. + * + */ + +package org.apache.streampipes.manager.runtime; + +import java.io.IOException; +import java.io.OutputStream; +import java.util.concurrent.TimeUnit; + +public class RateLimitedRuntimeInfoProvider { + + private static final int MAX_PREVIEW_TIME_CYCLES = 360; + private static final int MAX_FREQUENCY = 500; + + private final DataStreamRuntimeInfoProvider runtimeInfoProvider; + + public RateLimitedRuntimeInfoProvider(DataStreamRuntimeInfoProvider runtimeInfoProvider) { + this.runtimeInfoProvider = runtimeInfoProvider; + } + + public void streamOutput(OutputStream outputStream) { + runtimeInfoProvider.startConsuming(); + try { + for (int i = 0; i < MAX_PREVIEW_TIME_CYCLES; i++) { + String message = runtimeInfoProvider.getLatestEvent(); + if (message != null) { + try { + outputStream.write((message + "\n").getBytes()); + outputStream.flush(); + } catch (IOException ignored) { + } + } + TimeUnit.MILLISECONDS.sleep(MAX_FREQUENCY); + } + } catch (InterruptedException e) { + throw new RuntimeException(e); + } finally { + closeResources(runtimeInfoProvider); + } + } + + private void closeResources(DataStreamRuntimeInfoProvider fetcher) { + fetcher.close(); + } +} diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementPreview.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementPreview.java index 47105afc24..3a2a341463 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementPreview.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementPreview.java @@ -18,9 +18,12 @@ package org.apache.streampipes.rest.impl; import org.apache.streampipes.manager.preview.PipelinePreview; +import org.apache.streampipes.manager.runtime.DataStreamRuntimeInfoProvider; +import org.apache.streampipes.manager.runtime.RateLimitedRuntimeInfoProvider; import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.model.preview.PipelinePreviewModel; import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource; +import org.apache.streampipes.rest.shared.exception.BadRequestException; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; @@ -31,6 +34,7 @@ import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody; @RestController @RequestMapping("/api/v2/pipeline-element-preview") @@ -48,14 +52,16 @@ public class PipelineElementPreview extends AbstractAuthGuardedRestResource { } @GetMapping(path = "{previewId}/{pipelineElementDomId}", - produces = MediaType.APPLICATION_JSON_VALUE) - public ResponseEntity<?> getPipelinePreviewResult(@PathVariable("previewId") String previewId, - @PathVariable("pipelineElementDomId") String pipelineElementDomId) { + produces = MediaType.APPLICATION_OCTET_STREAM_VALUE) + public StreamingResponseBody getPipelinePreviewResult(@PathVariable("previewId") String previewId, + @PathVariable("pipelineElementDomId") String pipelineElementDomId) { try { - String runtimeInfo = new PipelinePreview().getPipelineElementPreview(previewId, pipelineElementDomId); - return ok(runtimeInfo); + var spDataStream = new PipelinePreview().getPipelineElementPreviewStream(previewId, pipelineElementDomId); + var runtimeInfoFetcher = new DataStreamRuntimeInfoProvider(spDataStream); + var runtimeInfoProvider = new RateLimitedRuntimeInfoProvider(runtimeInfoFetcher); + return runtimeInfoProvider::streamOutput; } catch (IllegalArgumentException e) { - return badRequest(); + throw new BadRequestException("Could not generate preview", e); } } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementRuntimeInfo.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementRuntimeInfo.java index 806bbbfce3..bc6cbf711d 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementRuntimeInfo.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementRuntimeInfo.java @@ -17,32 +17,29 @@ */ package org.apache.streampipes.rest.impl; -import org.apache.streampipes.commons.exceptions.SpRuntimeException; -import org.apache.streampipes.manager.operations.Operations; +import org.apache.streampipes.manager.runtime.DataStreamRuntimeInfoProvider; +import org.apache.streampipes.manager.runtime.RateLimitedRuntimeInfoProvider; import org.apache.streampipes.model.SpDataStream; -import org.apache.streampipes.model.message.Notifications; import org.apache.streampipes.rest.core.base.impl.AbstractRestResource; import org.springframework.http.MediaType; -import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody; @RestController @RequestMapping("/api/v2/pipeline-element/runtime") public class PipelineElementRuntimeInfo extends AbstractRestResource { @PostMapping( - produces = MediaType.APPLICATION_JSON_VALUE, + produces = MediaType.APPLICATION_OCTET_STREAM_VALUE, consumes = MediaType.APPLICATION_JSON_VALUE ) - public ResponseEntity<?> getRuntimeInfo(@RequestBody SpDataStream spDataStream) { - try { - return ok(Operations.getRuntimeInfo(spDataStream)); - } catch (SpRuntimeException e) { - return statusMessage(Notifications.error("Could not get runtime data")); - } + public StreamingResponseBody getRuntimeInfo(@RequestBody SpDataStream spDataStream) { + var runtimeInfoFetcher = new DataStreamRuntimeInfoProvider(spDataStream); + var runtimeInfoProvider = new RateLimitedRuntimeInfoProvider(runtimeInfoFetcher); + return runtimeInfoProvider::streamOutput; } } diff --git a/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.html b/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.html index 26ed16172e..f0e5d04b15 100644 --- a/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.html +++ b/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.html @@ -19,6 +19,5 @@ <div fxFlex="100" fxLayout="column"> <sp-pipeline-element-runtime-info [streamDescription]="streamDescription" - [pollingActive]="pollingActive" ></sp-pipeline-element-runtime-info> </div> diff --git a/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.ts b/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.ts index 2b8fd3b27a..5ea85c950a 100644 --- a/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.ts +++ b/ui/src/app/connect/dialog/adapter-started/adapter-started-preview/adapter-started-preview.component.ts @@ -24,15 +24,13 @@ import { RestService } from '../../../services/rest.service'; selector: 'sp-adapter-started-preview', templateUrl: './adapter-started-preview.component.html', }) -export class SpAdapterStartedPreviewComponent implements OnInit, OnDestroy { +export class SpAdapterStartedPreviewComponent implements OnInit { @Input() streamDescription: SpDataStream; @Input() adapterElementId: string; - pollingActive = false; - constructor( private adapterService: AdapterService, private restService: RestService, @@ -50,12 +48,7 @@ export class SpAdapterStartedPreviewComponent implements OnInit, OnDestroy { .getSourceDetails(adapter.correspondingDataStreamElementId) .subscribe(st => { this.streamDescription = st; - this.pollingActive = true; }); }); } - - ngOnDestroy(): void { - this.pollingActive = false; - } } diff --git a/ui/src/app/connect/services/rest.service.ts b/ui/src/app/connect/services/rest.service.ts index cede76c54c..279a151a37 100644 --- a/ui/src/app/connect/services/rest.service.ts +++ b/ui/src/app/connect/services/rest.service.ts @@ -18,9 +18,9 @@ import { Injectable } from '@angular/core'; -import { HttpClient, HttpContext } from '@angular/common/http'; +import { HttpClient, HttpContext, HttpEvent } from '@angular/common/http'; -import { Observable } from 'rxjs'; +import { Observable, share } from 'rxjs'; import { map } from 'rxjs/operators'; import { UnitDescription } from '../model/UnitDescription'; import { @@ -78,11 +78,16 @@ export class RestService { ); } - getRuntimeInfo(sourceDescription): Observable<Record<string, any>> { + getRuntimeInfo( + sourceDescription: SpDataStream, + ): Observable<HttpEvent<string>> { return this.http.post( `${this.platformServicesCommons.apiBasePath}/pipeline-element/runtime`, sourceDescription, { + responseType: 'text', + observe: 'events', + reportProgress: true, context: new HttpContext().set(NGX_LOADING_BAR_IGNORED, true), }, ); diff --git a/ui/src/app/core-ui/pipeline-element-runtime-info/pipeline-element-runtime-info.component.ts b/ui/src/app/core-ui/pipeline-element-runtime-info/pipeline-element-runtime-info.component.ts index 3fd5977784..301bb3a0c1 100644 --- a/ui/src/app/core-ui/pipeline-element-runtime-info/pipeline-element-runtime-info.component.ts +++ b/ui/src/app/core-ui/pipeline-element-runtime-info/pipeline-element-runtime-info.component.ts @@ -19,6 +19,8 @@ import { Component, Input, OnDestroy, OnInit } from '@angular/core'; import { SpDataStream } from '@streampipes/platform-services'; import { RestService } from '../../connect/services/rest.service'; +import { Subscription } from 'rxjs'; +import { HttpDownloadProgressEvent, HttpEventType } from '@angular/common/http'; @Component({ selector: 'sp-pipeline-element-runtime-info', @@ -29,55 +31,35 @@ export class PipelineElementRuntimeInfoComponent implements OnInit, OnDestroy { @Input() streamDescription: SpDataStream; - _pollingActive: boolean; - runtimeData: { runtimeName: string; value: any }[]; timer: any; runtimeDataError = false; + runtimeSub: Subscription; constructor(private restService: RestService) {} ngOnInit(): void { - this.checkPollingStart(); - } - - checkPollingStart() { - if (this._pollingActive) { - this.getLatestRuntimeInfo(); - } + this.getLatestRuntimeInfo(); } getLatestRuntimeInfo() { - this.restService + this.runtimeSub = this.restService .getRuntimeInfo(this.streamDescription) - .subscribe(data => { - this.runtimeDataError = !data; - - this.runtimeData = Object.entries(data).map( - ([runtimeName, value]) => ({ runtimeName, value }), - ); - - if (this._pollingActive) { - this.timer = setTimeout( - () => this.getLatestRuntimeInfo(), - 1000, + .subscribe(event => { + if (event.type === HttpEventType.DownloadProgress) { + const chunks = ( + event as HttpDownloadProgressEvent + ).partialText.split('\n'); + const json = JSON.parse(chunks[chunks.length - 2]); + this.runtimeDataError = !json; + this.runtimeData = Object.entries(json).map( + ([runtimeName, value]) => ({ runtimeName, value }), ); } }); } - @Input() - set pollingActive(pollingActive: boolean) { - this._pollingActive = pollingActive; - this.checkPollingStart(); - } - - get pollingActive(): boolean { - return this._pollingActive; - } - ngOnDestroy(): void { - this.pollingActive = false; - clearTimeout(this.timer); + this.runtimeSub?.unsubscribe(); } } diff --git a/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.html b/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.html index 306b9f5829..a58235feaa 100644 --- a/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.html +++ b/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.html @@ -28,7 +28,10 @@ </div> <table class="row-border hover preview-table" *ngIf="runtimeData"> <tbody id="preview-data-rows-id"> - <tr *ngFor="let item of runtimeData | keyvalue" class="preview-row"> + <tr + *ngFor="let item of runtimeData | keyvalue: keyValueCompareFn" + class="preview-row" + > <td> <b>{{ item.key }}</b> </td> diff --git a/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.ts b/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.ts index 8e57fc405e..64786a050d 100644 --- a/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.ts +++ b/ui/src/app/editor/components/pipeline-element-preview/pipeline-element-preview.component.ts @@ -16,25 +16,29 @@ * */ -import { Component, Input, OnInit } from '@angular/core'; +import { Component, Input, OnDestroy, OnInit } from '@angular/core'; import { EditorService } from '../../services/editor.service'; +import { Subscription } from 'rxjs'; +import { HttpDownloadProgressEvent, HttpEventType } from '@angular/common/http'; +import { KeyValue } from '@angular/common'; @Component({ selector: 'sp-pipeline-element-preview', templateUrl: './pipeline-element-preview.component.html', styleUrls: ['./pipeline-element-preview.component.scss'], }) -export class PipelineElementPreviewComponent implements OnInit { +export class PipelineElementPreviewComponent implements OnInit, OnDestroy { @Input() previewId: string; @Input() pipelineElementDomId: string; - runtimeData: ReadonlyMap<string, unknown>; + runtimeData: Record<string, any>; runtimeDataError = false; timer: any; + previewSub: Subscription; constructor(private editorService: EditorService) {} @@ -42,27 +46,51 @@ export class PipelineElementPreviewComponent implements OnInit { this.getLatestRuntimeInfo(); } + keyValueCompareFn = ( + a: KeyValue<string, any>, + b: KeyValue<string, any>, + ): number => { + return a.key.localeCompare(b.key); + }; + getLatestRuntimeInfo() { - this.editorService + this.previewSub = this.editorService .getPipelinePreviewResult(this.previewId, this.pipelineElementDomId) - .subscribe(data => { - if (data) { - this.runtimeDataError = false; - if ( - !( - Object.keys(data).length === 0 && - data.constructor === Object - ) - ) { - this.runtimeData = data; + .subscribe(event => { + if (event) { + if (event.type === HttpEventType.DownloadProgress) { + const chunks = ( + event as HttpDownloadProgressEvent + ).partialText.split('\n'); + this.runtimeData = JSON.parse( + chunks[chunks.length - 2], + ); + // this.runtimeDataError = !json; + // this.runtimeData = Object.entries(json).map( + // ([runtimeName, value]) => ({ runtimeName, value }), + // ); + console.log(this.runtimeData); } - - this.timer = setTimeout(() => { - this.getLatestRuntimeInfo(); - }, 1000); + // this.runtimeDataError = false; + // if ( + // !( + // Object.keys(data).length === 0 && + // data.constructor === Object + // ) + // ) { + // this.runtimeData = data; + // } + // + // this.timer = setTimeout(() => { + // this.getLatestRuntimeInfo(); + // }, 1000); } else { this.runtimeDataError = true; } }); } + + ngOnDestroy() { + this.previewSub?.unsubscribe(); + } } diff --git a/ui/src/app/editor/components/pipeline/pipeline.component.ts b/ui/src/app/editor/components/pipeline/pipeline.component.ts index 40044192e9..48a3a5dcad 100644 --- a/ui/src/app/editor/components/pipeline/pipeline.component.ts +++ b/ui/src/app/editor/components/pipeline/pipeline.component.ts @@ -24,6 +24,7 @@ import { ShepherdService } from '../../../services/tour/shepherd.service'; import { Component, EventEmitter, + HostListener, Input, NgZone, OnDestroy, @@ -170,6 +171,11 @@ export class PipelineComponent implements OnInit, OnDestroy { this.jsplumbFactoryService.destroy(this.preview); } + @HostListener('window:beforeunload') + onWindowClose() { + this.ngOnDestroy(); + } + updateMouseover(elementId: string) { this.currentMouseOverElement = elementId; } diff --git a/ui/src/app/editor/services/editor.service.ts b/ui/src/app/editor/services/editor.service.ts index a07f6bbee9..65256b2ae3 100644 --- a/ui/src/app/editor/services/editor.service.ts +++ b/ui/src/app/editor/services/editor.service.ts @@ -17,7 +17,7 @@ */ import { Injectable } from '@angular/core'; -import { HttpClient, HttpContext } from '@angular/common/http'; +import { HttpClient, HttpContext, HttpEvent } from '@angular/common/http'; import { DataProcessorInvocation, DataSinkInvocation, @@ -192,14 +192,15 @@ export class EditorService { getPipelinePreviewResult( previewId: string, pipelineElementDomId: string, - ): Observable<any> { + ): Observable<HttpEvent<string>> { return this.http.get( - this.pipelinePreviewBasePath + - '/' + - previewId + - '/' + - pipelineElementDomId, - { context: new HttpContext().set(NGX_LOADING_BAR_IGNORED, true) }, + `${this.pipelinePreviewBasePath}/${previewId}/${pipelineElementDomId}`, + { + responseType: 'text', + observe: 'events', + reportProgress: true, + context: new HttpContext().set(NGX_LOADING_BAR_IGNORED, true), + }, ); }
