You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by oe...@apache.org on 2026/06/11 12:00:12 UTC
(streampipes) 01/02: Add data stream change events
This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch 4544-create-a-simple-event-publisher-that-broadcasts-changes-of-a-data-stream
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 33e053f92a4ed19a26a5e90f95d78393f33d3b8f
Author: Sven Oehler <oe...@web.de>
AuthorDate: Thu Jun 11 13:59:32 2026 +0200
Add data stream change events
---
.../rest/event/DataStreamDeletedEvent.java | 22 ++++++++++++++++++++
.../rest/event/DataStreamUpdatedEvent.java | 24 ++++++++++++++++++++++
.../rest/impl/pe/DataStreamResource.java | 16 ++++++++++++++-
3 files changed, 61 insertions(+), 1 deletion(-)
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/DataStreamDeletedEvent.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/DataStreamDeletedEvent.java
new file mode 100644
index 0000000000..bcbbd36c8e
--- /dev/null
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/DataStreamDeletedEvent.java
@@ -0,0 +1,22 @@
+/*
+ * 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.rest.event;
+
+public record DataStreamDeletedEvent(String elementId) {
+}
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/DataStreamUpdatedEvent.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/DataStreamUpdatedEvent.java
new file mode 100644
index 0000000000..ac20a0ed11
--- /dev/null
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/DataStreamUpdatedEvent.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.rest.event;
+
+import org.apache.streampipes.model.SpDataStream;
+
+public record DataStreamUpdatedEvent(SpDataStream dataStream) {
+}
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
index 53ad620aff..3ed8bd5bcd 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
@@ -27,9 +27,12 @@ import org.apache.streampipes.model.message.NotificationType;
import org.apache.streampipes.model.monitoring.SpLogMessage;
import org.apache.streampipes.resource.management.DataStreamResourceManager;
import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
+import org.apache.streampipes.rest.event.DataStreamDeletedEvent;
+import org.apache.streampipes.rest.event.DataStreamUpdatedEvent;
import org.apache.streampipes.rest.security.AuthConstants;
import org.apache.http.client.HttpResponseException;
+import org.springframework.context.ApplicationEventPublisher;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PostFilter;
@@ -50,9 +53,12 @@ import java.util.List;
public class DataStreamResource extends AbstractAuthGuardedRestResource {
private final DataStreamUpdateManagement dataStreamUpdateManagement;
+ private final ApplicationEventPublisher eventPublisher;
- public DataStreamResource(ExtensionServiceRequestManager requestManager) {
+ public DataStreamResource(ExtensionServiceRequestManager requestManager,
+ ApplicationEventPublisher eventPublisher) {
this.dataStreamUpdateManagement = new DataStreamUpdateManagement(requestManager);
+ this.eventPublisher = eventPublisher;
}
@GetMapping(path = "/available", produces = MediaType.APPLICATION_JSON_VALUE)
@@ -72,6 +78,7 @@ public class DataStreamResource extends AbstractAuthGuardedRestResource {
@DeleteMapping(path = "/{elementId}", produces = MediaType.APPLICATION_JSON_VALUE)
@PreAuthorize(AuthConstants.HAS_WRITE_PIPELINE_ELEMENT_PRIVILEGE)
public ResponseEntity<Message> delete(@PathVariable("elementId") String elementId) {
+ publishEvent(new DataStreamDeletedEvent(elementId));
getDataStreamResourceManager().delete(elementId);
return constructSuccessMessage(NotificationType.STORAGE_SUCCESS.uiNotification());
}
@@ -95,6 +102,7 @@ public class DataStreamResource extends AbstractAuthGuardedRestResource {
"Element ID in path variable does not match element ID in request body");
} else {
dataStreamUpdateManagement.updateDataStream(updatedElement);
+ publishEvent(new DataStreamUpdatedEvent(updatedElement));
return constructSuccessMessage(NotificationType.STORAGE_SUCCESS.uiNotification());
}
}
@@ -128,4 +136,10 @@ public class DataStreamResource extends AbstractAuthGuardedRestResource {
return getSpResourceManager().manageDataStreams();
}
+ private void publishEvent(Object event) {
+ if (eventPublisher != null) {
+ eventPublisher.publishEvent(event);
+ }
+ }
+
}