Add a simple implementation of an event bus using reactor

.gitignore

@@ -1,6 +1,7 @@
 # Gradle
 .gradle
 build/
+out/
 
 # Ignore Gradle GUI config
 gradle-app.setting

README.md

@@ -21,6 +21,8 @@ Take a look, play and have fun with it!
     `./gradlew assemble --warning-mode all`
     2. Run the tests and plugins verification tasks:
     `./gradlew check --warning-mode all`
+    3. Execute the main application entrypoint:
+    `./gradlew run --warning-mode all`
 4. Start developing!
 
 ## 🤔 How to update dependencies

build.gradle

@@ -1,15 +1,22 @@
 apply plugin: 'java'
+apply plugin: 'application'
 
 sourceCompatibility = 1.11
 targetCompatibility = 1.11
 
 repositories {
     mavenCentral()
+    jcenter()
 }
 
 dependencies {
-    testCompile('org.junit.jupiter:junit-jupiter-api:5.3.2')
-    testRuntime('org.junit.jupiter:junit-jupiter-engine:5.3.2')
+    // Prod
+    implementation 'io.projectreactor:reactor-bus:2.0.8.RELEASE'
+
+    // Test
+    testCompile "org.mockito:mockito-core:2.+"
+    testCompile 'org.junit.jupiter:junit-jupiter-api:5.+'
+    testRuntime 'org.junit.jupiter:junit-jupiter-engine:5.+'
 }
 
 test {
@@ -22,4 +29,8 @@ test {
     reports {
         html.enabled = true
     }
-}
+}
+
+application {
+    mainClassName = "tv.codely.context.video.module.video.infrastructure.VideoPublisherCliController"
+}

src/main/java/tv/codely/context/notification/module/push/application/create/SendPushToSubscribersOnVideoPublished.java

@@ -0,0 +1,22 @@
+package tv.codely.context.notification.module.push.application.create;
+
+import tv.codely.context.video.module.video.domain.VideoPublished;
+import tv.codely.shared.application.DomainEventSubscriber;
+
+public class SendPushToSubscribersOnVideoPublished implements DomainEventSubscriber<VideoPublished> {
+    @Override
+    public Class<VideoPublished> subscribedTo() {
+        return VideoPublished.class;
+    }
+
+    @Override
+    public void consume(VideoPublished event) {
+        System.out.println(
+            String.format(
+                "Hey! There is a new video with title <%s> and description <%s>",
+                event.title(),
+                event.description()
+            )
+        );
+    }
+}

src/main/java/tv/codely/context/video/module/video/application/publish/VideoPublisher.java

@@ -0,0 +1,23 @@
+package tv.codely.context.video.module.video.application.publish;
+
+import tv.codely.context.video.module.video.domain.Video;
+import tv.codely.context.video.module.video.domain.VideoDescription;
+import tv.codely.context.video.module.video.domain.VideoTitle;
+import tv.codely.shared.domain.EventBus;
+
+public final class VideoPublisher {
+    private final EventBus eventBus;
+
+    public VideoPublisher(EventBus eventBus) {
+        this.eventBus = eventBus;
+    }
+
+    public void publish(String rawTitle, String rawDescription) {
+        final var title = new VideoTitle(rawTitle);
+        final var description = new VideoDescription(rawDescription);
+
+        final var video = Video.publish(title, description);
+
+        eventBus.publish(video.pullDomainEvents());
+    }
+}

src/main/java/tv/codely/context/video/module/video/domain/Video.java

@@ -0,0 +1,23 @@
+package tv.codely.context.video.module.video.domain;
+
+import tv.codely.shared.domain.AggregateRoot;
+
+public final class Video extends AggregateRoot {
+    private final VideoTitle title;
+    private final VideoDescription description;
+
+    private Video(VideoTitle title, VideoDescription description) {
+        this.title = title;
+        this.description = description;
+    }
+
+    public static Video publish(VideoTitle title, VideoDescription description) {
+        var video = new Video(title, description);
+
+        var videoCreated = new VideoPublished(title.value(), description.value());
+
+        video.record(videoCreated);
+
+        return video;
+    }
+}

src/main/java/tv/codely/context/video/module/video/domain/VideoDescription.java

@@ -0,0 +1,28 @@
+package tv.codely.context.video.module.video.domain;
+
+public final class VideoDescription {
+    private final String value;
+
+    public VideoDescription(final String value) {
+        this.value = value;
+    }
+
+    public String value() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        VideoDescription that = (VideoDescription) o;
+
+        return value.equals(that.value);
+    }
+
+    @Override
+    public int hashCode() {
+        return value.hashCode();
+    }
+}

src/main/java/tv/codely/context/video/module/video/domain/VideoPublished.java

@@ -0,0 +1,45 @@
+package tv.codely.context.video.module.video.domain;
+
+import tv.codely.shared.domain.DomainEvent;
+
+public final class VideoPublished implements DomainEvent {
+    private static final String FULL_QUALIFIED_EVENT_NAME = "codelytv.video.video.event.1.video.published";
+
+    private final String title;
+    private final String description;
+
+    public VideoPublished(String title, String description) {
+        this.title = title;
+        this.description = description;
+    }
+
+    public String fullQualifiedEventName() {
+        return FULL_QUALIFIED_EVENT_NAME;
+    }
+
+    public String title() {
+        return title;
+    }
+
+    public String description() {
+        return description;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        VideoPublished that = (VideoPublished) o;
+
+        if (!title.equals(that.title)) return false;
+        return description.equals(that.description);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = title.hashCode();
+        result = 31 * result + description.hashCode();
+        return result;
+    }
+}

src/main/java/tv/codely/context/video/module/video/domain/VideoTitle.java

@@ -0,0 +1,28 @@
+package tv.codely.context.video.module.video.domain;
+
+public final class VideoTitle {
+    private final String value;
+
+    public VideoTitle(String value) {
+        this.value = value;
+    }
+
+    public String value() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        VideoTitle that = (VideoTitle) o;
+
+        return value.equals(that.value);
+    }
+
+    @Override
+    public int hashCode() {
+        return value.hashCode();
+    }
+}

src/main/java/tv/codely/context/video/module/video/infrastructure/VideoPublisherCliController.java

@@ -0,0 +1,24 @@
+package tv.codely.context.video.module.video.infrastructure;
+
+import tv.codely.context.notification.module.push.application.create.SendPushToSubscribersOnVideoPublished;
+import tv.codely.context.video.module.video.application.publish.VideoPublisher;
+import tv.codely.shared.application.DomainEventSubscriber;
+import tv.codely.shared.domain.EventBus;
+import tv.codely.shared.infrastructure.bus.ReactorEventBus;
+
+import java.util.Set;
+
+public class VideoPublisherCliController {
+    public static void main(String[] args) {
+        final Set<DomainEventSubscriber> subscribers = Set.of(
+            new SendPushToSubscribersOnVideoPublished()
+        );
+        final EventBus eventBus = new ReactorEventBus(subscribers);
+        final var videoPublisher = new VideoPublisher(eventBus);
+
+        final var videoTitle = "\uD83C\uDF89 New YouTube.com/CodelyTV video title";
+        final var videoDescription = "This should be the video description \uD83D\uDE42";
+
+        videoPublisher.publish(videoTitle, videoDescription);
+    }
+}

src/main/java/tv/codely/shared/application/DomainEventSubscriber.java

@@ -0,0 +1,9 @@
+package tv.codely.shared.application;
+
+import tv.codely.shared.domain.DomainEvent;
+
+public interface DomainEventSubscriber<EventType extends DomainEvent> {
+    Class<EventType> subscribedTo();
+
+    void consume(EventType event);
+}

src/main/java/tv/codely/shared/domain/AggregateRoot.java

@@ -0,0 +1,19 @@
+package tv.codely.shared.domain;
+
+import java.util.LinkedList;
+import java.util.List;
+
+public abstract class AggregateRoot {
+    private List<DomainEvent> recordedDomainEvents = new LinkedList<>();
+
+    final public List<DomainEvent> pullDomainEvents() {
+        final var recordedDomainEvents = this.recordedDomainEvents;
+        this.recordedDomainEvents = new LinkedList<>();
+
+        return recordedDomainEvents;
+    }
+
+    final protected void record(DomainEvent event) {
+        recordedDomainEvents.add(event);
+    }
+}

src/main/java/tv/codely/shared/domain/DomainEvent.java

@@ -0,0 +1,5 @@
+package tv.codely.shared.domain;
+
+public interface DomainEvent {
+    String fullQualifiedEventName();
+}

src/main/java/tv/codely/shared/domain/EventBus.java

@@ -0,0 +1,7 @@
+package tv.codely.shared.domain;
+
+import java.util.List;
+
+public interface EventBus {
+    void publish(final List<DomainEvent> events);
+}

src/main/java/tv/codely/shared/infrastructure/bus/ReactorEventBus.java

@@ -0,0 +1,49 @@
+package tv.codely.shared.infrastructure.bus;
+
+import reactor.bus.Event;
+import reactor.bus.EventBus;
+import reactor.bus.selector.Selector;
+import reactor.fn.Consumer;
+import tv.codely.shared.application.DomainEventSubscriber;
+import tv.codely.shared.domain.DomainEvent;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+import static reactor.bus.selector.Selectors.$;
+
+public class ReactorEventBus implements tv.codely.shared.domain.EventBus {
+    private final EventBus bus;
+
+    public ReactorEventBus(final Set<DomainEventSubscriber> subscribers) {
+        bus = EventBus.create();
+
+        subscribers.forEach(this::registerOnEventBus);
+    }
+
+    @Override
+    public void publish(final List<DomainEvent> events) {
+        events.forEach(this::publish);
+    }
+
+    private void publish(final DomainEvent event) {
+        Class<? extends DomainEvent> eventIdentifier = event.getClass();
+        Event<DomainEvent> wrappedEvent = Event.wrap(event);
+
+        bus.notify(eventIdentifier, wrappedEvent);
+    }
+
+    private void registerOnEventBus(final DomainEventSubscriber subscriber) {
+        final Selector eventIdentifier = $(subscriber.subscribedTo());
+
+        bus.on(eventIdentifier, eventConsumer(subscriber));
+    }
+
+    private Consumer<Event> eventConsumer(final DomainEventSubscriber subscriber) {
+        return (Event reactorEvent) -> {
+            DomainEvent unwrappedEvent = (DomainEvent) reactorEvent.getData();
+            subscriber.consume(unwrappedEvent);
+        };
+    }
+}

src/test/java/tv/codely/context/video/module/video/application/publish/VideoPublisherShould.java

@@ -0,0 +1,28 @@
+package tv.codely.context.video.module.video.application.publish;
+
+import org.junit.jupiter.api.Test;
+import tv.codely.context.video.module.video.domain.VideoPublished;
+import tv.codely.shared.domain.EventBus;
+
+import java.util.List;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+final class VideoPublisherShould {
+    @Test
+    void publish_the_video_published_domain_event() {
+        final EventBus eventBus = mock(EventBus.class);
+        final var videoPublisher = new VideoPublisher(eventBus);
+
+        final var videoTitle = "\uD83C\uDF89 New YouTube.com/CodelyTV video title";
+        final var videoDescription = "This should be the video description \uD83D\uDE42";
+
+        videoPublisher.publish(videoTitle, videoDescription);
+
+        final var expectedVideoCreated = new VideoPublished(videoTitle, videoDescription);
+
+        verify(eventBus).publish(List.of(expectedVideoCreated));
+    }
+
+}