Add Apache Kafka Event Interface For Repositories

basyx.aasenvironment/basyx.aasenvironment.component/pom.xml

@@ -83,6 +83,32 @@
 			<groupId>org.eclipse.digitaltwin.basyx</groupId>
 			<artifactId>basyx.aasenvironment-feature-authorization</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.eclipse.digitaltwin.basyx</groupId>
+			<artifactId>basyx.aasrepository-feature-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.eclipse.digitaltwin.basyx</groupId>
+			<artifactId>basyx.submodelrepository-feature-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.eclipse.digitaltwin.basyx</groupId>
+			<artifactId>basyx.aasrepository-feature-kafka</artifactId>
+			<classifier>tests</classifier>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.eclipse.digitaltwin.basyx</groupId>
+			<artifactId>basyx.submodelservice-feature-kafka</artifactId>
+			<classifier>tests</classifier>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.eclipse.digitaltwin.basyx</groupId>
+			<artifactId>basyx.submodelrepository-feature-kafka</artifactId>
+			<classifier>tests</classifier>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.eclipse.digitaltwin.basyx</groupId>
 			<artifactId>basyx.http</artifactId>
@@ -100,8 +126,14 @@
 			<scope>test</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.eclipse.digitaltwin.basyx</groupId>
-			<artifactId>basyx.mongodbcore</artifactId>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.vintage</groupId>
+			<artifactId>junit-vintage-engine</artifactId>
+			<scope>test</scope>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

basyx.aasenvironment/basyx.aasenvironment.component/src/test/java/org/eclipse/digitaltwin/basyx/aasenvironment/component/KafkaEventsInMemoryStorageIntegrationTest.java

@@ -0,0 +1,155 @@
+/*******************************************************************************
+ * Copyright (C) 2024 DFKI GmbH (https://www.dfki.de/en/web)
+ * 
+ * Permission is hereby granted, free of charge, to any person obtaining
+ * a copy of this software and associated documentation files (the
+ * "Software"), to deal in the Software without restriction, including
+ * without limitation the rights to use, copy, modify, merge, publish,
+ * distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to
+ * the following conditions:
+ * 
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ * 
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+ * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+ * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ * 
+ * SPDX-License-Identifier: MIT
+ * 
+ ******************************************************************************/
+package org.eclipse.digitaltwin.basyx.aasenvironment.component;
+
+import java.util.concurrent.TimeUnit;
+
+import org.eclipse.digitaltwin.aas4j.v3.dataformat.json.JsonSerializer;
+import org.eclipse.digitaltwin.aas4j.v3.model.AssetAdministrationShell;
+import org.eclipse.digitaltwin.aas4j.v3.model.Submodel;
+import org.eclipse.digitaltwin.basyx.aasrepository.AasRepository;
+import org.eclipse.digitaltwin.basyx.aasrepository.feature.kafka.AasEventKafkaListener;
+import org.eclipse.digitaltwin.basyx.aasrepository.feature.kafka.KafkaAasRepositoryFeature;
+import org.eclipse.digitaltwin.basyx.aasrepository.feature.kafka.TestShells;
+import org.eclipse.digitaltwin.basyx.aasrepository.feature.kafka.events.model.AasEvent;
+import org.eclipse.digitaltwin.basyx.core.pagination.PaginationInfo;
+import org.eclipse.digitaltwin.basyx.submodelrepository.SubmodelRepository;
+import org.eclipse.digitaltwin.basyx.submodelrepository.feature.kafka.KafkaSubmodelRepositoryFeature;
+import org.eclipse.digitaltwin.basyx.submodelservice.feature.kafka.SubmodelEventKafkaListener;
+import org.eclipse.digitaltwin.basyx.submodelservice.feature.kafka.TestSubmodels;
+import org.eclipse.digitaltwin.basyx.submodelservice.feature.kafka.events.model.SubmodelEvent;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Import;
+import org.springframework.http.MediaType;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.DirtiesContext.ClassMode;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.test.web.servlet.result.MockMvcResultMatchers;
+
+/**
+ * @author sonnenberg (DFKI GmbH)
+ */
+@DirtiesContext(classMode = ClassMode.AFTER_CLASS)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.MOCK)
+@ComponentScan(basePackages = { "org.eclipse.digitaltwin.basyx"})
+@RunWith(SpringRunner.class)
+@TestPropertySource(properties = {
+		"basyx.environment=",
+		"basyx.feature.kafka.enabled=true",
+		"spring.kafka.bootstrap-servers=PLAINTEXT_HOST://localhost:9092"
+})
+@AutoConfigureMockMvc
+@Import({ SubmodelEventKafkaListener.class, AasEventKafkaListener.class})
+public class KafkaEventsInMemoryStorageIntegrationTest {
+
+	@Autowired
+	private AasEventKafkaListener aasEventListener;
+
+	@Autowired
+	private SubmodelEventKafkaListener submodelEventListener;
+
+	@Autowired
+	private KafkaAasRepositoryFeature aasFeature;
+
+	@Autowired
+	private KafkaSubmodelRepositoryFeature submodelFeature;
+
+	@Autowired
+	private MockMvc mvc;
+
+	@Autowired
+	private JsonSerializer serializer;
+
+	@Autowired
+	private SubmodelRepository smRepo;
+
+	@Autowired
+	private AasRepository aasRepo;
+
+	@Before
+	public void awaitAssignment() throws InterruptedException {
+		aasEventListener.awaitTopicAssignment();
+		submodelEventListener.awaitTopicAssignment();
+
+		cleanup();
+	}
+
+	@Test
+	public void testCreateAas() throws Exception {
+		AssetAdministrationShell shell = TestShells.shell();
+		String body = serializer.write(shell);
+
+		mvc.perform(MockMvcRequestBuilders.post("/shells").contentType(MediaType.APPLICATION_JSON).content(body).accept(MediaType.APPLICATION_JSON))
+				.andExpect(MockMvcResultMatchers.status().isCreated())
+				.andExpect(MockMvcResultMatchers.content().json(body));
+		AasEvent aasEvt = aasEventListener.next();
+		Assert.assertEquals(shell, aasEvt.getAas());
+		Assert.assertEquals(shell.getId(), aasEvt.getId());
+		Assert.assertNull(aasEvt.getSubmodelId());
+		Assert.assertNull(aasEvt.getAssetInformation());
+		Assert.assertNull(aasEvt.getReference());
+
+		Submodel sm = TestSubmodels.createSubmodel("http://submodels/123", "123", "hello"); 
+		body = serializer.write(sm);
+		mvc.perform(MockMvcRequestBuilders.post("/submodels").contentType(MediaType.APPLICATION_JSON).content(body).accept(MediaType.APPLICATION_JSON))
+				.andExpect(MockMvcResultMatchers.status().isCreated());
+		SubmodelEvent smEvt = submodelEventListener.next();
+		Assert.assertEquals(sm, smEvt.getSubmodel());
+		Assert.assertEquals(sm.getId(), smEvt.getId());
+		Assert.assertNull(smEvt.getSmElement());
+		Assert.assertNull(smEvt.getSmElementPath());
+	}
+
+
+	@Test
+	public void testFeatureIsEnabled() {
+		Assert.assertTrue(aasFeature.isEnabled());
+		Assert.assertTrue(submodelFeature.isEnabled());
+	}
+
+	@After
+	public void cleanup() throws InterruptedException {	
+		for (AssetAdministrationShell aas : aasRepo.getAllAas(new PaginationInfo(null, null)).getResult()) {
+			aasRepo.deleteAas(aas.getId());
+		}
+
+		for (Submodel sm : smRepo.getAllSubmodels(new PaginationInfo(null, null)).getResult()) {
+			smRepo.deleteSubmodel(sm.getId());
+		}
+		while(submodelEventListener.next(300, TimeUnit.MICROSECONDS) != null);
+	}
+}

basyx.aasregistry/basyx.aasregistry-service-basetests/src/main/java/org/eclipse/digitaltwin/basyx/aasregistry/service/tests/integration/BaseIntegrationTest.java

@@ -139,20 +139,23 @@ public abstract class BaseIntegrationTest {
 	protected RegistryAndDiscoveryInterfaceApi api;
 
 	@Before
-	public void initClient() throws ApiException {
+	public void setUp() throws Exception {
+		initClient();
+		cleanup();
+	}
+
+	protected void initClient() throws Exception {
 		api = new RegistryAndDiscoveryInterfaceApi("http", "127.0.0.1", port);
-		api.deleteAllShellDescriptors();
-		queue().assertNoAdditionalMessage();
 	}
 
-	@After
-	public void cleanup() throws ApiException {
-		queue().assertNoAdditionalMessage();
+	protected void cleanup() throws ApiException, InterruptedException {
+		queue().pullAdditionalMessages();
 		GetAssetAdministrationShellDescriptorsResult result = api.getAllAssetAdministrationShellDescriptors(null, null, null, null);
 		for (AssetAdministrationShellDescriptor eachDescriptor : result.getResult()) {
 			api.deleteAssetAdministrationShellDescriptorById(eachDescriptor.getId());
 			assertThatEventWasSend(RegistryEvent.builder().id(eachDescriptor.getId()).type(EventType.AAS_UNREGISTERED).build());
 		}
+		queue().pullAdditionalMessages();
 	}
 
 	@Test
@@ -230,7 +233,6 @@ public void whenDeleteAll_thenAllDescriptorsAreRemoved() throws ApiException {
 			assertThat(events.remove(RegistryEvent.builder().id("id_" + i).type(EventType.AAS_UNREGISTERED).build())).isTrue();
 		}
 		assertThat(events.isEmpty());
-		queue().assertNoAdditionalMessage();
 	}
 
 	@Test
@@ -246,7 +248,7 @@ public void whenCreateAndDeleteDescriptors_thenAllDescriptorsAreRemoved() throws
 		all = api.getAllAssetAdministrationShellDescriptors(null, null, null, null).getResult();
 		assertThat(all).isEmpty();
 
-		queue().assertNoAdditionalMessage();
+		queue().pullAdditionalMessages();
 	}
 
 	@Test
@@ -279,7 +281,7 @@ public void whenRegisterAndUnregisterSubmodel_thenSubmodelIsCreatedAndDeleted()
 		aasDescriptor = api.getAssetAdministrationShellDescriptorById(aasId);
 		assertThat(aasDescriptor.getSubmodelDescriptors()).doesNotContain(toRegister);
 
-		queue().assertNoAdditionalMessage();
+		queue().pullAdditionalMessages();
 	}
 
 	@Test

basyx.aasregistry/basyx.aasregistry-service-basetests/src/main/java/org/eclipse/digitaltwin/basyx/aasregistry/service/tests/integration/EventQueue.java

@@ -55,16 +55,8 @@ public void reset() {
 		}
 	}
 
-	public void assertNoAdditionalMessage() {
-		try {
-			String message = messageQueue.poll(1, TimeUnit.SECONDS);
-			if (message != null) {
-				throw new EventListenerException("Got additional message: " + message);
-			}
-		} catch (InterruptedException e) {
-			Thread.currentThread().interrupt();
-			throw new EventListenerException(e);
-		}
+	public void pullAdditionalMessages() throws InterruptedException {
+		while(messageQueue.poll(100, TimeUnit.MILLISECONDS) != null);
 	}
 
 	public RegistryEvent poll() {
@@ -81,11 +73,11 @@ public RegistryEvent poll() {
 			throw new EventListenerException(e);
 		}
 	}
-	
+
 	public static final class EventListenerException extends RuntimeException {
 
 		private static final long serialVersionUID = 1L;
-		
+
 		public EventListenerException(Throwable e) {
 			super(e);
 		}

basyx.aasregistry/basyx.aasregistry-service-release-kafka-mem/src/test/java/org/eclipse/digitaltwin/basyx/aasregistry/service/storage/memory/KafkaEventsInMemoryStorageIntegrationTest.java

@@ -24,17 +24,14 @@
  ******************************************************************************/
 package org.eclipse.digitaltwin.basyx.aasregistry.service.storage.memory;
 
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.common.TopicPartition;
 import org.eclipse.digitaltwin.basyx.aasregistry.service.tests.integration.BaseIntegrationTest;
 import org.eclipse.digitaltwin.basyx.aasregistry.service.tests.integration.EventQueue;
-import org.junit.Before;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaHandler;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.listener.ConsumerSeekAware;
@@ -51,10 +48,12 @@ public class KafkaEventsInMemoryStorageIntegrationTest extends BaseIntegrationTe
 	@Autowired
 	private RegistrationEventKafkaListener listener;
 
-	@Before
-	public void awaitAssignment() throws InterruptedException {
+	@Override
+	public void setUp() throws Exception {
 		listener.awaitTopicAssignment();
+		super.setUp();
 	}
+
 
 	@Override
 	public EventQueue queue() {
@@ -68,9 +67,6 @@ private static class RegistrationEventKafkaListener implements ConsumerSeekAware
 		private final EventQueue queue;
 		private final CountDownLatch latch = new CountDownLatch(1);
 
-		@Value("${spring.kafka.template.default-topic}")
-		private String topicName;	
-
 		@SuppressWarnings("unused")
 		public RegistrationEventKafkaListener(ObjectMapper mapper) {
 			this.queue = new EventQueue(mapper);
@@ -85,8 +81,7 @@ public void receiveMessage(String content) {
 		public void onPartitionsAssigned(Map<TopicPartition, Long> assignments,
 					ConsumerSeekCallback callback) {
 			for (TopicPartition eachPartition : assignments.keySet()) {
-				if (topicName.equals(eachPartition.topic())) {
-					callback.seekToEnd(List.of(eachPartition));
+				if ("aas-registry".equals(eachPartition.topic())) {
 					latch.countDown();
 				}
 			}		

basyx.aasregistry/basyx.aasregistry-service-release-kafka-mongodb/src/test/java/org/eclipse/digitaltwin/basyx/aasregistry/service/storage/mongodb/AuthorizedClientTest.java

@@ -61,23 +61,23 @@ public class AuthorizedClientTest extends BaseIntegrationTest {
 	@Value("${local.server.port}")
 	private int port;
 
-	@Before
-	public void awaitAssignment() throws InterruptedException {
+	@Override
+	public void setUp() throws Exception {
 		listener.awaitTopicAssignment();
+		super.setUp();
 	}
 
 	@Override
 	public EventQueue queue() {
 		return listener.getQueue();
 	}
 
-	@Before
 	@Override
-	public void initClient() throws ApiException {
+	public void initClient() throws Exception {
 		api = new AuthorizedConnectedAasRegistry("http://127.0.0.1:" + port, new TokenManager("http://localhost:9096/realms/BaSyx/protocol/openid-connect/token", new ClientCredentialAccessTokenProvider(new ClientCredential("workstation-1", "nY0mjyECF60DGzNmQUjL81XurSl8etom"))));
 
 		api.deleteAllShellDescriptors();
-		queue().assertNoAdditionalMessage();
+		queue().pullAdditionalMessages();
 	}
 
 	@Test
@@ -110,4 +110,4 @@ public void whenPostShellDescriptor_LocationIsReturned() throws ApiException, IO
 		// TODO: It uses normal GET unauthorized request, need to override and refactor
 	}
 
-}
+}