GH-3 - Refactor events registration to prepare easy addition of new r…

…epository types.

- Introduce Repository interface as a customization point
- Get rid of Registry interface by having one generic implementation
- Combine EventPublication and all its subclasses in one domain object
- Make Repository instead of Registry be dependent on EventSerializer

Signed-off-by: Dmitry Belyaev <[email protected]>

.git-authors

@@ -0,0 +1,5 @@
+authors:
+  db: Dmitry Belyaev; dbelyaev
+  bk: Björn Kieling; bkieling
+email:
+  domain: vmware.com

spring-modulith-events/pom.xml

@@ -23,26 +23,6 @@
 
 	<dependencies>
 
-		<dependency>
-			<groupId>org.springframework</groupId>
-			<artifactId>spring-context</artifactId>
-		</dependency>
-
-		<dependency>
-			<groupId>org.springframework</groupId>
-			<artifactId>spring-tx</artifactId>
-		</dependency>
-
-		<dependency>
-			<groupId>org.springframework</groupId>
-			<artifactId>spring-aop</artifactId>
-		</dependency>
-
-		<dependency>
-			<groupId>org.springframework</groupId>
-			<artifactId>spring-jdbc</artifactId>
-		</dependency>
-
 		<dependency>
 			<groupId>org.junit.jupiter</groupId>
 			<artifactId>junit-jupiter-engine</artifactId>
@@ -61,24 +41,6 @@
 			<scope>test</scope>
 		</dependency>
 
-		<dependency>
-			<groupId>org.hsqldb</groupId>
-			<artifactId>hsqldb</artifactId>
-			<scope>test</scope>
-		</dependency>
-
-		<!-- Logging -->
-		<dependency>
-			<groupId>org.slf4j</groupId>
-			<artifactId>slf4j-api</artifactId>
-		</dependency>
-
-		<dependency>
-			<groupId>org.slf4j</groupId>
-			<artifactId>jcl-over-slf4j</artifactId>
-			<scope>runtime</scope>
-		</dependency>
-
 		<dependency>
 			<groupId>ch.qos.logback</groupId>
 			<artifactId>logback-classic</artifactId>

spring-modulith-events/spring-modulith-events-core/pom.xml

@@ -16,5 +16,34 @@
 		<module.name>org.springframework.modulith.events.core</module.name>
 	</properties>
 
+	<dependencies>
 
-</project>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-context</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-tx</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-aop</artifactId>
+		</dependency>
+
+		<!-- Logging -->
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>slf4j-api</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>jcl-over-slf4j</artifactId>
+			<scope>runtime</scope>
+		</dependency>
+
+	</dependencies>
+</project>

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/CompletableEventPublication.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,7 +24,6 @@
  * @author Oliver Drotbohm
  */
 public interface CompletableEventPublication extends EventPublication {
-
 	/**
 	 * Returns the completion date of the publication.
 	 *

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/DefaultEventPublication.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2022 the original author or authors.
  *
  * Licensed 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
+ *      https://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,
@@ -27,7 +27,7 @@
 /**
  * Default {@link CompletableEventPublication} implementation.
  *
- * @author Oliver Gierke
+ * @author Oliver Drotbohm
  */
 @Getter
 @RequiredArgsConstructor(staticName = "of")
@@ -41,10 +41,6 @@ class DefaultEventPublication implements CompletableEventPublication {
 
 	private Optional<Instant> completionDate = Optional.empty();
 
-	/*
-	 * (non-Javadoc)
-	 * @see de.olivergierke.events.CompletableEventPublication#markCompleted()
-	 */
 	@Override
 	public CompletableEventPublication markCompleted() {
 

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/DefaultEventPublicationRegistry.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2017-2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.modulith.events;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.ApplicationListener;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.Assert;
+
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * A registry to capture event publications to {@link ApplicationListener}s. Allows to register those publications, mark
+ * them as completed and lookup incomplete publications.
+ *
+ * @author Oliver Drotbohm, Björn Kieling, Dmitry Belyaev
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class DefaultEventPublicationRegistry implements DisposableBean, EventPublicationRegistry {
+
+	private final @NonNull EventPublicationRepository events;
+
+	/**
+	 * Stores {@link EventPublication}s for the given event and {@link ApplicationListener}s.
+	 *
+	 * @param event must not be {@literal null}.
+	 * @param listeners must not be {@literal null}.
+	 */
+	public void store(Object event, Stream<PublicationTargetIdentifier> listeners) {
+
+		listeners.map(it -> map(event, it))
+				.forEach(events::create);
+	}
+
+	/**
+	 * Returns all {@link EventPublication}s that have not been completed yet.
+	 *
+	 * @return will never be {@literal null}.
+	 */
+	public Iterable<EventPublication> findIncompletePublications() {
+		return events.findByCompletionDateIsNull();
+	}
+
+	/**
+	 * Marks the publication for the given event and {@link PublicationTargetIdentifier} as completed.
+	 *
+	 * @param event must not be {@literal null}.
+	 * @param targetIdentifier must not be {@literal null}.
+	 */
+	@Transactional(propagation = Propagation.REQUIRES_NEW)
+	public void markCompleted(Object event, PublicationTargetIdentifier targetIdentifier) {
+
+		Assert.notNull(event, "Domain event must not be null!");
+		Assert.notNull(targetIdentifier, "Listener identifier must not be null!");
+
+		events.findByEventAndTargetIdentifier(event, targetIdentifier) //
+				.map(DefaultEventPublicationRegistry::LOGCompleted) //
+				.map(e -> CompletableEventPublication.of(e.getEvent(), e.getTargetIdentifier()))
+				.ifPresent(it -> events.updateCompletionDate(it.markCompleted()));
+	}
+
+	@Override
+	public void destroy() {
+
+		List<EventPublication> publications = events.findByCompletionDateIsNull();
+
+		if (publications.isEmpty()) {
+
+			LOG.info("No publications outstanding!");
+			return;
+		}
+
+		LOG.info("Shutting down with the following publications left unfinished:");
+
+		for (int i = 0; i < publications.size(); i++) {
+
+			String prefix = (i + 1) == publications.size() ? "└─" : "├─";
+			EventPublication it = publications.get(i);
+
+			LOG.info("{} - {} - {}", prefix, it.getEvent().getClass().getName(), it.getTargetIdentifier().getValue());
+		}
+	}
+
+	private EventPublication map(Object event, PublicationTargetIdentifier targetIdentifier) {
+
+		EventPublication result = CompletableEventPublication.of(event, targetIdentifier);
+
+		LOG.debug("Registering publication of {} for {}.", //
+				result.getEvent().getClass().getName(), result.getTargetIdentifier().getValue());
+
+		return result;
+	}
+
+	private static EventPublication LOGCompleted(EventPublication publication) {
+
+		LOG.debug("Marking publication of event {} to listener {} completed.", //
+				publication.getEvent().getClass().getName(), publication.getTargetIdentifier().getValue());
+
+		return publication;
+	}
+}

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventPublication.java

@@ -24,11 +24,9 @@
 /**
  * An event publication.
  *
- * @author Oliver Drotbohm
- * @see CompletableEventPublication#of(Object, PublicationTargetIdentifier)
+ * @author Oliver Drotbohm, Björn Kieling, Dmitry Belyaev
  */
 public interface EventPublication extends Comparable<EventPublication> {
-
 	/**
 	 * Returns the event that is published.
 	 *
@@ -78,10 +76,6 @@ default boolean isIdentifiedBy(PublicationTargetIdentifier identifier) {
 		return this.getTargetIdentifier().equals(identifier);
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see java.lang.Comparable#compareTo(java.lang.Object)
-	 */
 	@Override
 	public default int compareTo(EventPublication that) {
 		return this.getPublicationDate().compareTo(that.getPublicationDate());

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventPublicationRegistry.java

@@ -18,13 +18,12 @@
 import java.util.stream.Stream;
 
 import org.springframework.context.ApplicationListener;
-import org.springframework.util.Assert;
 
 /**
  * A registry to capture event publications to {@link ApplicationListener}s. Allows to register those publications, mark
  * them as completed and lookup incomplete publications.
  *
- * @author Oliver Drotbohm
+ * @author Oliver Drotbohm, Björn Kieling, Dmitry Belyaev
  */
 public interface EventPublicationRegistry {
 
@@ -37,29 +36,17 @@ public interface EventPublicationRegistry {
 	void store(Object event, Stream<PublicationTargetIdentifier> listeners);
 
 	/**
-	 * Marks the publication for the given event and {@link PublicationTargetIdentifier} as completed.
-	 *
-	 * @param event must not be {@literal null}.
-	 * @param listener must not be {@literal null}.
-	 */
-	void markCompleted(Object event, PublicationTargetIdentifier listener);
-
-	/**
-	 * Marks the given {@link EventPublication} as completed.
+	 * Returns all {@link EventPublication}s that have not been completed yet.
 	 *
-	 * @param publication must not be {@literal null}.
+	 * @return will never be {@literal null}.
 	 */
-	default void markCompleted(EventPublication publication) {
-
-		Assert.notNull(publication, "Publication must not be null!");
-
-		markCompleted(publication.getEvent(), publication.getTargetIdentifier());
-	}
+	Iterable<EventPublication> findIncompletePublications();
 
 	/**
-	 * Returns all {@link EventPublication}s that have not been completed yet.
+	 * Marks the publication for the given event and {@link PublicationTargetIdentifier} as completed.
 	 *
-	 * @return will never be {@literal null}.
+	 * @param event must not be {@literal null}.
+	 * @param targetIdentifier must not be {@literal null}.
 	 */
-	Iterable<EventPublication> findIncompletePublications();
+	void markCompleted(Object event, PublicationTargetIdentifier targetIdentifier);
 }

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventPublicationRepository.java

@@ -0,0 +1,30 @@
+package org.springframework.modulith.events;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Repository to store {@link EventPublication}s.
+ *
+ * @author Björn Kieling, Dmitry Belyaev
+ */
+public interface EventPublicationRepository {
+
+    EventPublication create(EventPublication publication);
+
+    EventPublication updateCompletionDate(CompletableEventPublication publication);
+
+    /**
+     * Returns all {@link EventPublication} that have not been completed yet.
+     */
+    List<EventPublication> findByCompletionDateIsNull();
+
+    /**
+     * Return the {@link EventPublication} for the given serialized event and listener identifier.
+     *
+     * @param event            must not be {@literal null}.
+     * @param targetIdentifier must not be {@literal null}.
+     * @return
+     */
+    Optional<EventPublication> findByEventAndTargetIdentifier(Object event, PublicationTargetIdentifier targetIdentifier);
+}

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventSerializer.java

@@ -35,5 +35,5 @@ public interface EventSerializer {
 	 * @param type must not be {@literal null}.
 	 * @return
 	 */
-	Object deserialize(Object serialized, Class<?> type);
+	<T> T deserialize(Object serialized, Class<T> type);
 }

spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/PublicationTargetIdentifier.java

@@ -27,7 +27,7 @@
 @RequiredArgsConstructor(staticName = "of")
 public class PublicationTargetIdentifier {
 
-	String value;
+	private String value;
 
 	/*
 	 * (non-Javadoc)