This is an automated email from the ASF dual-hosted git repository.
SvenO3 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 2d1d3d8958 feat(#4544): Create a simple event publisher that
broadcasts changes of a data stream (#4546)
2d1d3d8958 is described below
commit 2d1d3d895846b430e27aec8960d7616a5729559b
Author: Sven Oehler <[email protected]>
AuthorDate: Thu Jun 11 14:33:09 2026 +0200
feat(#4544): Create a simple event publisher that broadcasts changes of a
data stream (#4546)
---
.../rest/event/AdapterDeletedEvent.java | 24 ++++++++++++++++++++++
.../rest/event/AdapterUpdatedEvent.java | 24 ++++++++++++++++++++++
.../rest/event/DataStreamDeletedEvent.java | 22 ++++++++++++++++++++
.../rest/event/DataStreamUpdatedEvent.java | 24 ++++++++++++++++++++++
.../rest/impl/connect/AdapterResource.java | 22 ++++++++++++++++++++
.../rest/impl/pe/DataStreamResource.java | 16 ++++++++++++++-
6 files changed, 131 insertions(+), 1 deletion(-)
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/AdapterDeletedEvent.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/AdapterDeletedEvent.java
new file mode 100644
index 0000000000..d6bb23574e
--- /dev/null
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/AdapterDeletedEvent.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.connect.adapter.AdapterDescription;
+
+public record AdapterDeletedEvent(AdapterDescription adapterDescription) {
+}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/AdapterUpdatedEvent.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/AdapterUpdatedEvent.java
new file mode 100644
index 0000000000..0e67fdbe67
--- /dev/null
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/event/AdapterUpdatedEvent.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.connect.adapter.AdapterDescription;
+
+public record AdapterUpdatedEvent(AdapterDescription adapterDescription) {
+}
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/connect/AdapterResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
index f408b9c1f4..235135bcdf 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
@@ -37,6 +37,8 @@ import org.apache.streampipes.model.monitoring.SpLogMessage;
import org.apache.streampipes.model.util.ElementIdGenerator;
import org.apache.streampipes.resource.management.PermissionResourceManager;
import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.rest.event.AdapterDeletedEvent;
+import org.apache.streampipes.rest.event.AdapterUpdatedEvent;
import org.apache.streampipes.rest.security.AuthConstants;
import org.apache.streampipes.rest.security.SpPermissionEvaluator;
import org.apache.streampipes.rest.shared.constants.SpMediaType;
@@ -47,6 +49,8 @@ import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.http.HttpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEventPublisher;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PostFilter;
@@ -71,9 +75,17 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
private static final Logger LOG =
LoggerFactory.getLogger(AdapterResource.class);
private final ExtensionServiceRequestManager requestManager;
+ private final ApplicationEventPublisher eventPublisher;
public AdapterResource(WorkerRestClient workerRestClient,
ExtensionServiceRequestManager requestManager) {
+ this(workerRestClient, requestManager, null);
+ }
+
+ @Autowired
+ public AdapterResource(WorkerRestClient workerRestClient,
+ ExtensionServiceRequestManager requestManager,
+ ApplicationEventPublisher eventPublisher) {
super(() -> new AdapterMasterManagement(
StorageDispatcher.INSTANCE.getNoSqlStore()
.getAdapterInstanceStorage(),
@@ -84,6 +96,7 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
StorageDispatcher.INSTANCE.getNoSqlStore().getExtensionsServiceStorage(),
requestManager));
this.requestManager = requestManager;
+ this.eventPublisher = eventPublisher;
}
@PostMapping(consumes = MediaType.APPLICATION_JSON_VALUE)
@@ -123,6 +136,7 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
var updateManager = new AdapterUpdateManagement(managementService,
requestManager);
try {
updateManager.updateAdapter(adapterDescription);
+ publishEvent(new AdapterUpdatedEvent(adapterDescription));
} catch (AdapterException e) {
LOG.error("Error while updating adapter with id {}",
adapterDescription.getElementId(), e);
return ok(Notifications.error(e.getMessage(),
ExceptionUtils.getStackTrace(e)));
@@ -237,6 +251,7 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
if (pipelinesUsingAdapter.isEmpty()) {
try {
managementService.deleteAdapter(elementId);
+ publishEvent(new AdapterDeletedEvent(adapter));
return ok(Notifications.success("Adapter with id: " + elementId +
" is deleted."));
} catch (AdapterException e) {
@@ -290,6 +305,7 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
PipelineManager.deletePipeline(pipelineId);
}
managementService.deleteAdapter(elementId);
+ publishEvent(new AdapterDeletedEvent(adapter));
return ok(Notifications.success("Adapter with id: " + elementId
+ " and all pipelines using the adapter are deleted."));
@@ -337,4 +353,10 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
.getPipelinesUsingAdapter(adapterId);
}
+ private void publishEvent(Object event) {
+ if (eventPublisher != null) {
+ eventPublisher.publishEvent(event);
+ }
+ }
+
}
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);
+ }
+ }
+
}
