This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new cb5d5739e5 fix(#3837): Remove delete preview rest endpoint (#3838)
cb5d5739e5 is described below
commit cb5d5739e554aaef5813633b7d7f5be03c312726
Author: Philipp Zehnder <[email protected]>
AuthorDate: Wed Oct 15 17:01:19 2025 +0200
fix(#3837): Remove delete preview rest endpoint (#3838)
---
.../manager/runtime/ClientDisconnectCallback.java | 24 +++++++++++
.../runtime/RateLimitedRuntimeInfoProvider.java | 7 +++-
.../PipelinePreviewResource.java} | 23 +++++------
.../StreamPreviewResource.java} | 9 ++--
.../apis/pipeline-element-runtime-info.service.ts | 2 +-
.../pipeline-assembly-drawing-area.component.ts | 48 +++++++++++-----------
ui/src/app/editor/services/editor.service.ts | 6 +--
7 files changed, 72 insertions(+), 47 deletions(-)
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/ClientDisconnectCallback.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/ClientDisconnectCallback.java
new file mode 100644
index 0000000000..c95f0e152f
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/ClientDisconnectCallback.java
@@ -0,0 +1,24 @@
+/*
+ * 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;
+
+@FunctionalInterface
+public interface ClientDisconnectCallback {
+ void onClientDisconnect();
+}
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
index ee763a40f6..7ef0949521 100644
---
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
@@ -31,10 +31,14 @@ public class RateLimitedRuntimeInfoProvider {
private final DataStreamRuntimeInfoProvider runtimeInfoProvider;
private final ObjectMapper objectMapper;
+ private final ClientDisconnectCallback callback;
- public RateLimitedRuntimeInfoProvider(DataStreamRuntimeInfoProvider
runtimeInfoProvider) {
+
+ public RateLimitedRuntimeInfoProvider(DataStreamRuntimeInfoProvider
runtimeInfoProvider,
+ ClientDisconnectCallback callback) {
this.runtimeInfoProvider = runtimeInfoProvider;
this.objectMapper = new ObjectMapper();
+ this.callback = callback;
}
public void streamOutput(OutputStream outputStream) {
@@ -47,6 +51,7 @@ public class RateLimitedRuntimeInfoProvider {
outputStream.write((objectMapper.writeValueAsString(messages) +
"\n").getBytes());
outputStream.flush();
} catch (IOException ignored) {
+ callback.onClientDisconnect();
}
}
TimeUnit.MILLISECONDS.sleep(MAX_FREQUENCY);
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/runtime/PipelinePreviewResource.java
similarity index 82%
rename from
streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementPreview.java
rename to
streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/runtime/PipelinePreviewResource.java
index 59fb2aa053..3192f17a6e 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/runtime/PipelinePreviewResource.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.rest.impl;
+package org.apache.streampipes.rest.impl.runtime;
import org.apache.streampipes.manager.preview.PipelinePreview;
import org.apache.streampipes.manager.runtime.DataStreamRuntimeInfoProvider;
@@ -27,7 +27,6 @@ import
org.apache.streampipes.rest.shared.exception.BadRequestException;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
-import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
@@ -39,8 +38,8 @@ import
org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBo
import jakarta.servlet.http.HttpServletResponse;
@RestController
-@RequestMapping("/api/v2/pipeline-element-preview")
-public class PipelineElementPreview extends AbstractAuthGuardedRestResource {
+@RequestMapping("/api/v2/pipeline-preview")
+public class PipelinePreviewResource extends AbstractAuthGuardedRestResource {
@PostMapping(
@@ -54,26 +53,24 @@ public class PipelineElementPreview extends
AbstractAuthGuardedRestResource {
}
@GetMapping(path = "{previewId}",
- produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
+ produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
public StreamingResponseBody getPipelinePreviewResult(
HttpServletResponse response,
- @PathVariable("previewId") String previewId) {
+ @PathVariable("previewId") String previewId
+ ) {
try {
// deactivate nginx proxy buffering for better performance of streaming
output
response.addHeader("X-Accel-Buffering", "no");
var spDataStreams = new
PipelinePreview().getPipelineElementPreviewStreams(previewId);
var runtimeInfoFetcher = new
DataStreamRuntimeInfoProvider(spDataStreams);
- var runtimeInfoProvider = new
RateLimitedRuntimeInfoProvider(runtimeInfoFetcher);
+ var runtimeInfoProvider = new RateLimitedRuntimeInfoProvider(
+ runtimeInfoFetcher,
+ () -> new PipelinePreview().deletePreview(previewId)
+ );
return runtimeInfoProvider::streamOutput;
} catch (IllegalArgumentException e) {
throw new BadRequestException("Could not generate preview", e);
}
}
- @DeleteMapping(path = "{previewId}", produces =
MediaType.APPLICATION_JSON_VALUE)
- public ResponseEntity<Void>
deletePipelinePreviewRequest(@PathVariable("previewId") String previewId) {
- new PipelinePreview().deletePreview(previewId);
- return ok();
- }
-
}
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/runtime/StreamPreviewResource.java
similarity index 91%
rename from
streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementRuntimeInfo.java
rename to
streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/runtime/StreamPreviewResource.java
index 17a11bfbfc..5988a96d1d 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/runtime/StreamPreviewResource.java
@@ -15,7 +15,8 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.rest.impl;
+
+package org.apache.streampipes.rest.impl.runtime;
import org.apache.streampipes.manager.runtime.DataStreamRuntimeInfoProvider;
import org.apache.streampipes.manager.runtime.RateLimitedRuntimeInfoProvider;
@@ -34,8 +35,8 @@ import jakarta.servlet.http.HttpServletResponse;
import java.util.Map;
@RestController
-@RequestMapping("/api/v2/pipeline-element/runtime")
-public class PipelineElementRuntimeInfo extends AbstractRestResource {
+@RequestMapping("/api/v2/stream-preview")
+public class StreamPreviewResource extends AbstractRestResource {
@PostMapping(
produces = MediaType.APPLICATION_OCTET_STREAM_VALUE,
@@ -46,7 +47,7 @@ public class PipelineElementRuntimeInfo extends
AbstractRestResource {
// deactivate nginx proxy buffering for better performance of streaming
output
response.addHeader("X-Accel-Buffering", "no");
var runtimeInfoFetcher = new
DataStreamRuntimeInfoProvider(Map.of("adapter", spDataStream));
- var runtimeInfoProvider = new
RateLimitedRuntimeInfoProvider(runtimeInfoFetcher);
+ var runtimeInfoProvider = new
RateLimitedRuntimeInfoProvider(runtimeInfoFetcher, () -> {});
return runtimeInfoProvider::streamOutput;
}
}
diff --git
a/ui/projects/streampipes/platform-services/src/lib/apis/pipeline-element-runtime-info.service.ts
b/ui/projects/streampipes/platform-services/src/lib/apis/pipeline-element-runtime-info.service.ts
index e9d53b8c3b..b88df44ce7 100644
---
a/ui/projects/streampipes/platform-services/src/lib/apis/pipeline-element-runtime-info.service.ts
+++
b/ui/projects/streampipes/platform-services/src/lib/apis/pipeline-element-runtime-info.service.ts
@@ -34,7 +34,7 @@ export class PipelineElementRuntimeInfoService {
sourceDescription: SpDataStream,
): Observable<HttpEvent<string>> {
return this.http.post(
-
`${this.platformServicesCommons.apiBasePath}/pipeline-element/runtime`,
+ `${this.platformServicesCommons.apiBasePath}/stream-preview`,
sourceDescription,
{
responseType: 'text',
diff --git
a/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly-drawing-area/pipeline-assembly-drawing-area.component.ts
b/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly-drawing-area/pipeline-assembly-drawing-area.component.ts
index f78a8d0fec..bf1f8b2d35 100644
---
a/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly-drawing-area/pipeline-assembly-drawing-area.component.ts
+++
b/ui/src/app/editor/components/pipeline-assembly/pipeline-assembly-drawing-area/pipeline-assembly-drawing-area.component.ts
@@ -22,6 +22,7 @@ import {
EventEmitter,
Input,
NgZone,
+ OnDestroy,
OnInit,
Output,
ViewChild,
@@ -52,7 +53,7 @@ import { Subscription } from 'rxjs';
styleUrls: ['./pipeline-assembly-drawing-area.component.scss'],
standalone: false,
})
-export class PipelineAssemblyDrawingAreaComponent implements OnInit {
+export class PipelineAssemblyDrawingAreaComponent implements OnInit, OnDestroy
{
@Input()
jsplumbBridge: JsplumbBridge;
@@ -109,6 +110,10 @@ export class PipelineAssemblyDrawingAreaComponent
implements OnInit {
}
}
+ ngOnDestroy(): void {
+ this.pipelinePreviewSub?.unsubscribe();
+ }
+
isPipelineAssemblyEmpty() {
return (
this.rawPipelineModel.length === 0 ||
@@ -152,36 +157,33 @@ export class PipelineAssemblyDrawingAreaComponent
implements OnInit {
);
this.editorService
.initiatePipelinePreview(pipeline)
- .subscribe(response => {
- this.pipelinePreview = response;
+ .subscribe(pipelinePreviewModel => {
+ this.pipelinePreview = pipelinePreviewModel;
this.previewModeActive = true;
- this.pipelinePreviewSub = this.editorService
- .getPipelinePreviewResult(response.previewId)
- .subscribe(res => {
- const data = this.livePreviewService.convert(
- res as HttpDownloadProgressEvent,
- );
- if (data) {
- this.livePreviewService.eventSub.next(data);
- }
- });
+ this.subscribeToPreview(pipelinePreviewModel.previewId);
});
} else {
- this.deletePipelineElementPreview(false);
+ this.deletePipelineElementPreview();
}
}
- deletePipelineElementPreview(resume: boolean) {
+ private subscribeToPreview(previewId: string) {
+ this.pipelinePreviewSub = this.editorService
+ .getPipelinePreviewResult(previewId)
+ .subscribe(res => {
+ const data = this.livePreviewService.convert(
+ res as HttpDownloadProgressEvent,
+ );
+ if (data) {
+ this.livePreviewService.eventSub.next(data);
+ }
+ });
+ }
+
+ deletePipelineElementPreview() {
if (this.previewModeActive) {
this.pipelinePreviewSub?.unsubscribe();
- this.editorService
- .deletePipelinePreviewRequest(this.pipelinePreview.previewId)
- .subscribe(() => {
- this.previewModeActive = false;
- if (resume) {
- this.togglePipelineElementLivePreview();
- }
- });
+ this.previewModeActive = false;
}
}
diff --git a/ui/src/app/editor/services/editor.service.ts
b/ui/src/app/editor/services/editor.service.ts
index c5b4445493..b232d8ca08 100644
--- a/ui/src/app/editor/services/editor.service.ts
+++ b/ui/src/app/editor/services/editor.service.ts
@@ -200,10 +200,6 @@ export class EditorService {
);
}
- deletePipelinePreviewRequest(previewId: string): Observable<any> {
- return this.http.delete(this.pipelinePreviewBasePath + '/' +
previewId);
- }
-
getPipelinePreviewResult(previewId: string): Observable<HttpEvent<string>>
{
return this.http.get(`${this.pipelinePreviewBasePath}/${previewId}`, {
responseType: 'text',
@@ -214,6 +210,6 @@ export class EditorService {
}
get pipelinePreviewBasePath() {
- return this.apiBasePath + '/pipeline-element-preview';
+ return this.apiBasePath + '/pipeline-preview';
}
}
