feat: add publish-subscribe design pattern (#2898) (#3223)

* feat: add publisher-subscriber design pattern (#2898) * fix: check styles on publisher-subscriber design pattern (#2898) * fix: check styles on publisher-subscriber design pattern (#2898) * fix: check styles on publisher-subscriber design pattern (#2898) * fix: improve coverage on publisher-subscriber design pattern (#2898) * fix: fix dependancy on publisher-subscriber design pattern (#2898) * fix: fix documentation on publisher-subscriber design pattern (#2898) * fix: apply spotless on publisher-subscriber design pattern (#2898) * fix: changes after review on publisher-subscriber design pattern (#2898) * fix: changes after review on publisher-subscriber design pattern (#2898) * fix: fix sonar on publisher-subscriber design pattern (#2898) * fix: fix sonar on publisher-subscriber design pattern (#2898) --------- Co-authored-by: Sanura Hettiarachchi <sanura.hettiarachchi@tipico.com>
2026-05-14 08:58:26 +00:00 · 2025-03-30 19:24:15 +02:00
parent 0ca162a55c
commit 418557129b
19 changed files with 1003 additions and 0 deletions
@@ -188,6 +188,7 @@
    <module>property</module>
    <module>prototype</module>
    <module>proxy</module>
+    <module>publish-subscribe</module>
    <module>queue-based-load-leveling</module>
    <module>reactor</module>
    <module>registry</module>
@@ -0,0 +1,302 @@
+---
+title: "Publish-Subscribe Pattern in Java: Decoupling the solution with asynchronous communication"
+shortTitle: Publish-Subscribe
+description: "Explore the Publish-Subscribe design pattern in Java with detailed examples. Learn how it helps to create loosely coupled, scalable, and flexible systems by allowing components to communicate asynchronously without knowing each other directly."
+category: Behavioral
+language: en
+tag:
+    - Decoupling
+    - Event-driven
+    - Gang Of Four
+    - Publish/subscribe
+---
+
+## Intent of the Publish-Subscribe Design Pattern
+
+The Publish-Subscribe design pattern is widely used in software architecture to transmit data between various components in a system.
+It is a behavioral design pattern aimed at achieving loosely coupled communication between objects.
+The primary intent is to allow a one-to-many dependency relationship where one object (the Publisher) notifies multiple other objects (the Subscribers)
+about changes or events, without needing to know who or what the subscribers are.
+
+## Detailed Explanation of Publish-Subscribe Pattern with Real-World Examples
+
+### Real-world example
+
+- Messaging systems like Kafka, RabbitMQ, AWS SNS, JMS
+    - **Kafka** : publishes messages to topics and subscribers consumes them in real time for analytics, logs or other purposes.
+    - **RabbitMQ** : Uses exchanges as publisher and queues as subscribers to route messages
+    - **AWS SNS** : Simple Notification Service (SNS) received the messages from publishers with topic and the subscribers on that topic will receive the messages. (SQS, Lambda functions, emails, SMS)
+
+
+- Event driven microservices
+    - **Publisher** : Point of Sale(PoS) system records the sale of an item and publish the event
+    - **Subscribers** : Inventory management service updates stock, Billing service sends e-bill to customer
+
+
+- Newsletter subscriptions
+    - **Publisher** : Writes a new blog post and publish to subscribers
+    - **Subscribers** : All the subscribers to the newsletter receive the email
+
+### In plain words
+
+The Publish-Subscribe design pattern allows senders (publishers) to broadcast messages to multiple receivers (subscribers) without knowing who they are,
+enabling loose coupling and asynchronous communication in a system
+
+### Wikipedia says
+
+In software architecture, publish–subscribe or pub/sub is a messaging pattern where publishers categorize messages into classes that are received by subscribers.
+This is contrasted to the typical messaging pattern model where publishers send messages directly to subscribers.
+
+Similarly, subscribers express interest in one or more classes and only receive messages that are of interest, without knowledge of which publishers, if any, there are.
+
+Publish–subscribe is a sibling of the message queue paradigm, and is typically one part of a larger message-oriented middleware system.
+Most messaging systems support both the pub/sub and message queue models in their API; e.g., Java Message Service (JMS).
+
+### Architectural Diagram
+![pub-sub](./etc/pub-sub.png)
+
+## Programmatic Example of Publish-Subscribe Pattern in Java
+
+First we need to identify the Event on which we need the pub-sub methods to trigger.
+For example:
+
+- Sending alerts based on the weather events such as earthquakes, floods and tornadoes
+- Sending alerts based on the temperature
+- Sending an email to different customer support emails when a support ticket is created.
+
+The Message class below will hold the content of the message we need to pass between the publisher and the subscribers.
+
+```java
+public record Message(Object content) {
+}
+
+```
+
+The Topic class will have the topic **name** based on the event
+
+- Weather events TopicName WEATHER
+- Weather events TopicName TEMPERATURE
+- Support ticket created TopicName CUSTOMER_SUPPORT
+- Any other custom topic depending on use case
+- Also, the Topic contains a list of subscribers that will listen to that topic
+
+We can add or remove subscribers from the subscription to the topic
+
+```java
+public class Topic {
+
+    private final TopicName name;
+    private final Set<Subscriber> subscribers = new CopyOnWriteArraySet<>();
+    //...//
+}
+```
+
+Then we can create the publisher. The publisher class has a set of topics.
+
+- Each new topic has to be registered in the publisher.
+- Publish method will publish the _Message_ to the corresponding _Topic_.
+
+```java
+public class PublisherImpl implements Publisher {
+
+    private static final Logger logger = LoggerFactory.getLogger(PublisherImpl.class);
+    private final Set<Topic> topics = new HashSet<>();
+
+    @Override
+    public void registerTopic(Topic topic) {
+        topics.add(topic);
+    }
+
+    @Override
+    public void publish(Topic topic, Message message) {
+        if (!topics.contains(topic)) {
+            logger.error("This topic is not registered: {}", topic.getName());
+            return;
+        }
+        topic.publish(message);
+    }
+}
+```
+
+Finally, we can Subscribers to the Topics we want to listen to.
+
+- For WEATHER topic we will create _WeatherSubscriber_
+- _WeatherSubscriber_ can also subscribe to TEMPERATURE topic
+- For CUSTOMER_SUPPORT topic we will create _CustomerSupportSubscribe_
+- Also to demonstrate the async behavior we will create a _DelayedWeatherSubscriber_ who has a 0.2 sec processing deplay
+
+All classes will have a _onMessage_ method which will take a Message input.
+
+- On message method will verify the content of the message is as expected
+- After content is verified it will perform the operation based on the message
+    - _WeatherSubscriber_ will send a weather or temperature alert based on the _Message_
+    - _CustomerSupportSubscribe_will send an email based on the _Message_
+    - _DelayedWeatherSubscriber_ will send a weather alert based on the _Message_ after a delay
+
+```java
+public interface Subscriber {
+    void onMessage(Message message);
+}
+```
+
+And here is the invocation of the publisher and subscribers.
+
+```java
+public static void main(String[] args) throws InterruptedException {
+
+    final String topicWeather = "WEATHER";
+    final String topicTemperature = "TEMPERATURE";
+    final String topicCustomerSupport = "CUSTOMER_SUPPORT";
+
+    // 1. create the publisher.
+    Publisher publisher = new PublisherImpl();
+
+    // 2. define the topics and register on publisher
+    Topic weatherTopic = new Topic(topicWeather);
+    publisher.registerTopic(weatherTopic);
+
+    Topic temperatureTopic = new Topic(topicTemperature);
+    publisher.registerTopic(temperatureTopic);
+
+    Topic supportTopic = new Topic(topicCustomerSupport);
+    publisher.registerTopic(supportTopic);
+
+    // 3. Create the subscribers and subscribe to the relevant topics
+    // weatherSub1 will subscribe to two topics WEATHER and TEMPERATURE.
+    Subscriber weatherSub1 = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSub1);
+    temperatureTopic.addSubscriber(weatherSub1);
+
+    // weatherSub2 will subscribe to WEATHER topic
+    Subscriber weatherSub2 = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSub2);
+
+    // delayedWeatherSub will subscribe to WEATHER topic
+    // NOTE :: DelayedWeatherSubscriber has a 0.2 sec delay of processing message.
+    Subscriber delayedWeatherSub = new DelayedWeatherSubscriber();
+    weatherTopic.addSubscriber(delayedWeatherSub);
+
+    // subscribe the customer support subscribers to the CUSTOMER_SUPPORT topic.
+    Subscriber supportSub1 = new CustomerSupportSubscriber();
+    supportTopic.addSubscriber(supportSub1);
+    Subscriber supportSub2 = new CustomerSupportSubscriber();
+    supportTopic.addSubscriber(supportSub2);
+
+    // 4. publish message from each topic
+    publisher.publish(weatherTopic, new Message("earthquake"));
+    publisher.publish(temperatureTopic, new Message("23C"));
+    publisher.publish(supportTopic, new Message("support@test.de"));
+
+    // 5. unregister subscriber from TEMPERATURE topic
+    temperatureTopic.removeSubscriber(weatherSub1);
+
+    // 6. publish message under TEMPERATURE topic
+    publisher.publish(temperatureTopic, new Message("0C"));
+
+    /*
+     * Finally, we wait for the subscribers to consume messages to check the output.
+     * The output can change on each run, depending on how long the execution on each
+     * subscriber would take
+     * Expected behavior:
+     * - weatherSub1 will consume earthquake and 23C
+     * - weatherSub2 will consume earthquake
+     * - delayedWeatherSub will take longer and consume earthquake
+     * - supportSub1, supportSub2 will consume support@test.de
+     * - the message 0C will not be consumed because weatherSub1 unsubscribed from TEMPERATURE topic
+     */
+    TimeUnit.SECONDS.sleep(2);
+}
+```
+
+Program output:
+
+Note that the order of output could change everytime you run the program.
+The subscribers could take different time to consume the message.
+
+```
+14:01:45.599 [ForkJoinPool.commonPool-worker-6] INFO com.iluwatar.publish.subscribe.subscriber.CustomerSupportSubscriber -- Customer Support Subscriber: 1416331388 sent the email to: support@test.de
+14:01:45.599 [ForkJoinPool.commonPool-worker-4] INFO com.iluwatar.publish.subscribe.subscriber.WeatherSubscriber -- Weather Subscriber: 1949521124 issued message: 23C
+14:01:45.599 [ForkJoinPool.commonPool-worker-2] INFO com.iluwatar.publish.subscribe.subscriber.WeatherSubscriber -- Weather Subscriber: 60629172 issued message: earthquake
+14:01:45.599 [ForkJoinPool.commonPool-worker-5] INFO com.iluwatar.publish.subscribe.subscriber.CustomerSupportSubscriber -- Customer Support Subscriber: 1807508804 sent the email to: support@test.de
+14:01:45.599 [ForkJoinPool.commonPool-worker-1] INFO com.iluwatar.publish.subscribe.subscriber.WeatherSubscriber -- Weather Subscriber: 1949521124 issued message: earthquake
+14:01:47.600 [ForkJoinPool.commonPool-worker-3] INFO com.iluwatar.publish.subscribe.subscriber.DelayedWeatherSubscriber -- Delayed Weather Subscriber: 2085808749 issued message: earthquake
+```
+
+## When to Use the Publish-Subscribe Pattern
+
+- Event-Driven Systems
+    - Use Pub/Sub when your system relies on events (e.g., user registration, payment completion).
+    - Example: After a user registers, send a welcome email and log the action simultaneously.
+
+- Asynchronous Communication
+    - When tasks can be performed without waiting for immediate responses.
+    - Example: In an e-commerce app, notify the warehouse and the user after a successful order.
+
+- Decoupling Components
+    - Ideal for systems where producers and consumers should not depend on each other.
+    - Example: A logging service listens for logs from multiple microservices.
+
+- Scaling Systems
+    - Useful when you need to scale services without changing the core application logic.
+    - Example: Broadcasting messages to thousands of clients (chat applications, IoT).
+
+- Broadcasting Notifications
+    - When a message should be delivered to multiple receivers.
+    - Example: Sending promotional offers to multiple user devices.
+
+- Microservices Communication
+    - Allow independent services to communicate without direct coupling.
+    - Example: An order service publishes an event, and both the billing and shipping services process it.
+
+## When to avoid the Publish-Subscribe Pattern
+
+- Simple applications where direct calls suffice.
+- Strong consistency requirements (e.g., banking transactions).
+- Low-latency synchronous communication needed.
+
+## Benefits and Trade-offs of Publish-Subscribe Pattern
+
+### Benefits:
+
+- Decoupling
+    - Publishers and subscribers are independent of each other.
+    - Publishers don’t need to know who the subscribers are, and vice versa.
+    - Changes in one component don’t affect the other.
+- Scalability
+    - New subscribers can be added without modifying publishers.
+    - Supports distributed systems where multiple services consume the same events.
+- Dynamic Subscription
+    - Subscribers can subscribe/unsubscribe at runtime.
+    - Enables flexible event-driven architectures.
+- Asynchronous Communication
+    - Publishers and subscribers operate independently, improving performance.
+    - Useful for background processing (e.g., notifications, logging).
+- Broadcast Communication
+    - A single event can be consumed by multiple subscribers.
+    - Useful for fan-out scenarios (e.g., notifications, analytics).
+- Resilience & Fault Tolerance
+    - If a subscriber fails, others can still process messages.
+    - Message brokers (e.g., Kafka, RabbitMQ) can retry or persist undelivered messages.
+
+### Trade-offs:
+
+- Complexity in Debugging
+    - Since publishers and subscribers are decoupled, tracing event flow can be difficult.
+    - Requires proper logging and monitoring tools.
+- Message Ordering & Consistency
+    - Ensuring message order across subscribers can be challenging (e.g., Kafka vs. RabbitMQ).
+    - Some systems may process events out of order.
+- Potential Latency
+    - Asynchronous processing introduces delays compared to direct calls.
+    - Not ideal for real-time synchronous requirements.
+
+## Related Java Design Patterns
+
+* [Observer Pattern](https://github.com/sanurah/java-design-patterns/blob/master/observer/): Both involve a producer (subject/publisher) notifying consumers (observers/subscribers). Observer is synchronous & tightly coupled (observers know the subject). Pub-Sub is asynchronous & decoupled (via a message broker).
+* [Mediator Pattern](https://github.com/sanurah/java-design-patterns/blob/master/mediator/): A mediator centralizes communication between components (like a message broker in Pub-Sub). Mediator focuses on reducing direct dependencies between objects. Pub-Sub focuses on broadcasting events to unknown subscribers.
+
+## References and Credits
+
+* [Apache Kafka – Pub-Sub Model](https://kafka.apache.org/documentation/#design_pubsub)
+* [Microsoft – Publish-Subscribe Pattern](https://learn.microsoft.com/en-us/azure/architecture/patterns/publisher-subscriber)
+* [Martin Fowler – Event-Driven Architecture](https://martinfowler.com/articles/201701-event-driven.html)
@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.iluwatar</groupId>
+        <artifactId>java-design-patterns</artifactId>
+        <version>1.26.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>publish-subscribe</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>
@@ -0,0 +1,115 @@
+package com.iluwatar.publish.subscribe;
+
+import com.iluwatar.publish.subscribe.model.Message;
+import com.iluwatar.publish.subscribe.model.Topic;
+import com.iluwatar.publish.subscribe.publisher.Publisher;
+import com.iluwatar.publish.subscribe.publisher.PublisherImpl;
+import com.iluwatar.publish.subscribe.subscriber.CustomerSupportSubscriber;
+import com.iluwatar.publish.subscribe.subscriber.DelayedWeatherSubscriber;
+import com.iluwatar.publish.subscribe.subscriber.Subscriber;
+import com.iluwatar.publish.subscribe.subscriber.WeatherSubscriber;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * The Publish and Subscribe pattern is a messaging paradigm used in software architecture with
+ * several key points:
+ * <li>Decoupling of publishers and subscribers: Publishers and subscribers operate independently,
+ *     and there's no direct link between them. This enhances the scalability and * modularity of
+ *     applications.
+ * <li>Event-driven communication: The pattern facilitates event-driven architectures by allowing
+ *     publishers to broadcast events without concerning themselves with who receives the events.
+ * <li>Dynamic subscription: Subscribers can dynamically choose to listen for specific events or
+ *     messages they are interested in, often by subscribing to a particular topic or channel.
+ * <li>Asynchronous processing: The pattern inherently supports asynchronous message processing,
+ *     enabling efficient handling of events and improving application responsiveness.
+ * <li>Scalability: By decoupling senders and receivers, the pattern can support a large number of
+ *     publishers and subscribers, making it suitable for scalable systems.
+ * <li>Flexibility and adaptability: New subscribers or publishers can be added to the system
+ *     without significant changes to the existing components, making the system highly adaptable to
+ *     evolving requirements.
+ *
+ *     <p>In this example we will create three topics WEATHER, TEMPERATURE and CUSTOMER_SUPPORT.
+ *     Then we will register those topics in the {@link Publisher}. After that we will create two
+ *     {@link WeatherSubscriber}s, one {@link DelayedWeatherSubscriber} and two {@link
+ *     CustomerSupportSubscriber}.The subscribers will subscribe to the relevant topics. One {@link
+ *     WeatherSubscriber} will subscribe to two topics (WEATHER, TEMPERATURE). {@link
+ *     DelayedWeatherSubscriber} has a delay in message processing. Now we can publish the three
+ *     {@link Topic}s with different content in the {@link Message}s. And we can observe the output
+ *     in the log where, one {@link WeatherSubscriber} will output the message with weather and the
+ *     other {@link WeatherSubscriber} will output weather and temperature. {@link
+ *     CustomerSupportSubscriber}s will output the message with customer support email. {@link
+ *     DelayedWeatherSubscriber} has a delay in processing and will output the message at last. Each
+ *     subscriber is only listening to the subscribed topics.
+ */
+public class App {
+
+  /**
+   * Program entry point.
+   *
+   * @param args command line args
+   */
+  public static void main(String[] args) throws InterruptedException {
+
+    final String topicWeather = "WEATHER";
+    final String topicTemperature = "TEMPERATURE";
+    final String topicCustomerSupport = "CUSTOMER_SUPPORT";
+
+    // 1. create the publisher.
+    Publisher publisher = new PublisherImpl();
+
+    // 2. define the topics and register on publisher
+    Topic weatherTopic = new Topic(topicWeather);
+    publisher.registerTopic(weatherTopic);
+
+    Topic temperatureTopic = new Topic(topicTemperature);
+    publisher.registerTopic(temperatureTopic);
+
+    Topic supportTopic = new Topic(topicCustomerSupport);
+    publisher.registerTopic(supportTopic);
+
+    // 3. Create the subscribers and subscribe to the relevant topics
+    // weatherSub1 will subscribe to two topics WEATHER and TEMPERATURE.
+    Subscriber weatherSub1 = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSub1);
+    temperatureTopic.addSubscriber(weatherSub1);
+
+    // weatherSub2 will subscribe to WEATHER topic
+    Subscriber weatherSub2 = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSub2);
+
+    // delayedWeatherSub will subscribe to WEATHER topic
+    // NOTE :: DelayedWeatherSubscriber has a 0.2 sec delay of processing message.
+    Subscriber delayedWeatherSub = new DelayedWeatherSubscriber();
+    weatherTopic.addSubscriber(delayedWeatherSub);
+
+    // subscribe the customer support subscribers to the CUSTOMER_SUPPORT topic.
+    Subscriber supportSub1 = new CustomerSupportSubscriber();
+    supportTopic.addSubscriber(supportSub1);
+    Subscriber supportSub2 = new CustomerSupportSubscriber();
+    supportTopic.addSubscriber(supportSub2);
+
+    // 4. publish message from each topic
+    publisher.publish(weatherTopic, new Message("earthquake"));
+    publisher.publish(temperatureTopic, new Message("23C"));
+    publisher.publish(supportTopic, new Message("support@test.de"));
+
+    // 5. unregister subscriber from TEMPERATURE topic
+    temperatureTopic.removeSubscriber(weatherSub1);
+
+    // 6. publish message under TEMPERATURE topic
+    publisher.publish(temperatureTopic, new Message("0C"));
+
+    /*
+     * Finally, we wait for the subscribers to consume messages to check the output.
+     * The output can change on each run, depending on how long the execution on each
+     * subscriber would take
+     * Expected behavior:
+     * - weatherSub1 will consume earthquake and 23C
+     * - weatherSub2 will consume earthquake
+     * - delayedWeatherSub will take longer and consume earthquake
+     * - supportSub1, supportSub2 will consume support@test.de
+     * - the message 0C will not be consumed because weatherSub1 unsubscribed from TEMPERATURE topic
+     */
+    TimeUnit.SECONDS.sleep(2);
+  }
+}
@@ -0,0 +1,4 @@
+package com.iluwatar.publish.subscribe.model;
+
+/** This class represents a Message that holds the published content. */
+public record Message(Object content) {}
@@ -0,0 +1,48 @@
+package com.iluwatar.publish.subscribe.model;
+
+import com.iluwatar.publish.subscribe.subscriber.Subscriber;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArraySet;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+
+/** This class represents a Topic that topic name and subscribers. */
+@Getter
+@Setter
+@RequiredArgsConstructor
+public class Topic {
+
+  private final String topicName;
+  private final Set<Subscriber> subscribers = new CopyOnWriteArraySet<>();
+
+  /**
+   * Add a subscriber to the list of subscribers.
+   *
+   * @param subscriber subscriber to add
+   */
+  public void addSubscriber(Subscriber subscriber) {
+    subscribers.add(subscriber);
+  }
+
+  /**
+   * Remove a subscriber to the list of subscribers.
+   *
+   * @param subscriber subscriber to remove
+   */
+  public void removeSubscriber(Subscriber subscriber) {
+    subscribers.remove(subscriber);
+  }
+
+  /**
+   * Publish a message to subscribers.
+   *
+   * @param message message with content to publish
+   */
+  public void publish(Message message) {
+    for (Subscriber subscriber : subscribers) {
+      CompletableFuture.runAsync(() -> subscriber.onMessage(message));
+    }
+  }
+}
@@ -0,0 +1,23 @@
+package com.iluwatar.publish.subscribe.publisher;
+
+import com.iluwatar.publish.subscribe.model.Message;
+import com.iluwatar.publish.subscribe.model.Topic;
+
+/** This class represents a Publisher. */
+public interface Publisher {
+
+  /**
+   * Register a topic in the publisher.
+   *
+   * @param topic the topic to be registered
+   */
+  void registerTopic(Topic topic);
+
+  /**
+   * Register a topic in the publisher.
+   *
+   * @param topic the topic to publish the message under
+   * @param message message with content to be published
+   */
+  void publish(Topic topic, Message message);
+}
@@ -0,0 +1,29 @@
+package com.iluwatar.publish.subscribe.publisher;
+
+import com.iluwatar.publish.subscribe.model.Message;
+import com.iluwatar.publish.subscribe.model.Topic;
+import java.util.HashSet;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** This class is an implementation of the Publisher. */
+public class PublisherImpl implements Publisher {
+
+  private static final Logger logger = LoggerFactory.getLogger(PublisherImpl.class);
+  private final Set<Topic> topics = new HashSet<>();
+
+  @Override
+  public void registerTopic(Topic topic) {
+    topics.add(topic);
+  }
+
+  @Override
+  public void publish(Topic topic, Message message) {
+    if (!topics.contains(topic)) {
+      logger.error("This topic is not registered: {}", topic.getTopicName());
+      return;
+    }
+    topic.publish(message);
+  }
+}
@@ -0,0 +1,26 @@
+package com.iluwatar.publish.subscribe.subscriber;
+
+import com.iluwatar.publish.subscribe.model.Message;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** This class subscribes to CUSTOMER_SUPPORT topic. */
+@Slf4j
+public class CustomerSupportSubscriber implements Subscriber {
+
+  private static final Logger logger = LoggerFactory.getLogger(CustomerSupportSubscriber.class);
+
+  @Override
+  public void onMessage(Message message) {
+    if (message.content() instanceof String content) {
+      logger.info(
+          "Customer Support Subscriber: {} sent the email to: {}", this.hashCode(), content);
+    } else {
+      logger.error(
+          "Unknown content type: {} expected: {}",
+          message.content().getClass().getSimpleName(),
+          String.class.getSimpleName());
+    }
+  }
+}
@@ -0,0 +1,37 @@
+package com.iluwatar.publish.subscribe.subscriber;
+
+import com.iluwatar.publish.subscribe.model.Message;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** This class subscribes to WEATHER topic. */
+@Slf4j
+public class DelayedWeatherSubscriber implements Subscriber {
+
+  private static final Logger logger = LoggerFactory.getLogger(DelayedWeatherSubscriber.class);
+
+  @Override
+  public void onMessage(Message message) {
+    if (message.content() instanceof String content) {
+      processData();
+      logger.info("Delayed Weather Subscriber: {} issued message: {}", this.hashCode(), content);
+    } else {
+      logger.error(
+          "Unknown content type: {} expected: {}",
+          message.content().getClass().getSimpleName(),
+          String.class.getSimpleName());
+    }
+  }
+
+  /** create an artificial delay to mimic the persistence and timeouts in real world. */
+  private void processData() {
+    try {
+      TimeUnit.MILLISECONDS.sleep(2000);
+    } catch (InterruptedException e) {
+      logger.error("Interrupted!", e);
+      Thread.currentThread().interrupt();
+    }
+  }
+}
@@ -0,0 +1,14 @@
+package com.iluwatar.publish.subscribe.subscriber;
+
+import com.iluwatar.publish.subscribe.model.Message;
+
+/** This class represents a Subscriber. */
+public interface Subscriber {
+
+  /**
+   * On message method will trigger when the subscribed event is published.
+   *
+   * @param message the message contains the content of the published event
+   */
+  void onMessage(Message message);
+}
@@ -0,0 +1,25 @@
+package com.iluwatar.publish.subscribe.subscriber;
+
+import com.iluwatar.publish.subscribe.model.Message;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** This class subscribes to WEATHER or TEMPERATURE topic. */
+@Slf4j
+public class WeatherSubscriber implements Subscriber {
+
+  private static final Logger logger = LoggerFactory.getLogger(WeatherSubscriber.class);
+
+  @Override
+  public void onMessage(Message message) {
+    if (message.content() instanceof String content) {
+      logger.info("Weather Subscriber: {} issued message: {}", this.hashCode(), content);
+    } else {
+      logger.error(
+          "Unknown content type: {} expected: {}",
+          message.content().getClass().getSimpleName(),
+          String.class.getSimpleName());
+    }
+  }
+}
@@ -0,0 +1,13 @@
+package com.iluwatar.publish.subscribe;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import org.junit.jupiter.api.Test;
+
+public class AppTest {
+
+  @Test
+  void shouldExecuteApplicationWithoutException() {
+    assertDoesNotThrow(() -> App.main(new String[] {}));
+  }
+}
@@ -0,0 +1,40 @@
+package com.iluwatar.publish.subscribe;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.slf4j.LoggerFactory;
+
+public class LoggerExtension implements BeforeEachCallback, AfterEachCallback {
+
+  private final ListAppender<ILoggingEvent> listAppender = new ListAppender<>();
+  private final Logger logger = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
+
+  @Override
+  public void afterEach(ExtensionContext extensionContext) throws Exception {
+    listAppender.stop();
+    listAppender.list.clear();
+    logger.detachAppender(listAppender);
+  }
+
+  @Override
+  public void beforeEach(ExtensionContext extensionContext) throws Exception {
+    logger.addAppender(listAppender);
+    listAppender.start();
+  }
+
+  public List<String> getMessages() {
+    return listAppender.list.stream().map(e -> e.getMessage()).collect(Collectors.toList());
+  }
+
+  public List<String> getFormattedMessages() {
+    return listAppender.list.stream()
+        .map(e -> e.getFormattedMessage())
+        .collect(Collectors.toList());
+  }
+}
@@ -0,0 +1,17 @@
+package com.iluwatar.publish.subscribe.model;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+import org.junit.jupiter.api.Test;
+
+public class MessageTest {
+
+  @Test
+  public void testMessage() {
+    final String content = "some content";
+    Message message = new Message(content);
+    assertInstanceOf(String.class, message.content());
+    assertEquals(content, String.valueOf(message.content()));
+  }
+}
@@ -0,0 +1,40 @@
+package com.iluwatar.publish.subscribe.model;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+import com.iluwatar.publish.subscribe.subscriber.Subscriber;
+import com.iluwatar.publish.subscribe.subscriber.WeatherSubscriber;
+import java.lang.reflect.Field;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+
+public class TopicTest {
+
+  private static final String TOPIC_WEATHER = "WEATHER";
+
+  @Test
+  void testTopic() {
+    Topic topic = new Topic(TOPIC_WEATHER);
+    assertEquals(TOPIC_WEATHER, topic.getTopicName());
+  }
+
+  @Test
+  void testSubscribing() throws NoSuchFieldException, IllegalAccessException {
+
+    Topic topic = new Topic(TOPIC_WEATHER);
+    Subscriber sub = new WeatherSubscriber();
+    topic.addSubscriber(sub);
+
+    Field field = topic.getClass().getDeclaredField("subscribers");
+    field.setAccessible(true);
+    Object value = field.get(topic);
+    assertInstanceOf(Set.class, value);
+
+    Set<?> subscribers = (Set<?>) field.get(topic);
+    assertEquals(1, subscribers.size());
+
+    topic.removeSubscriber(sub);
+    assertEquals(0, subscribers.size());
+  }
+}
@@ -0,0 +1,60 @@
+package com.iluwatar.publish.subscribe.publisher;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+import com.iluwatar.publish.subscribe.LoggerExtension;
+import com.iluwatar.publish.subscribe.model.Message;
+import com.iluwatar.publish.subscribe.model.Topic;
+import java.lang.reflect.Field;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class PublisherTest {
+
+  @RegisterExtension public LoggerExtension loggerExtension = new LoggerExtension();
+
+  private static final String TOPIC_WEATHER = "WEATHER";
+  private static final String TOPIC_CUSTOMER_SUPPORT = "CUSTOMER_SUPPORT";
+
+  @Test
+  void testRegisterTopic() throws NoSuchFieldException, IllegalAccessException {
+    Topic topic = new Topic(TOPIC_CUSTOMER_SUPPORT);
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(topic);
+
+    Field field = publisher.getClass().getDeclaredField("topics");
+    field.setAccessible(true);
+    Object value = field.get(publisher);
+    assertInstanceOf(Set.class, value);
+
+    Set<?> topics = (Set<?>) field.get(publisher);
+    assertEquals(1, topics.size());
+  }
+
+  @Test
+  void testPublish() {
+    Topic topic = new Topic(TOPIC_WEATHER);
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(topic);
+
+    Message message = new Message("weather");
+    assertDoesNotThrow(() -> publisher.publish(topic, message));
+  }
+
+  @Test
+  void testPublishUnregisteredTopic() {
+    Topic topic = new Topic(TOPIC_WEATHER);
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(topic);
+
+    Topic topicUnregistered = new Topic(TOPIC_CUSTOMER_SUPPORT);
+    Message message = new Message("support");
+    publisher.publish(topicUnregistered, message);
+    assertEquals(
+        "This topic is not registered: CUSTOMER_SUPPORT",
+        loggerExtension.getFormattedMessages().getFirst());
+  }
+}
@@ -0,0 +1,183 @@
+package com.iluwatar.publish.subscribe.subscriber;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.iluwatar.publish.subscribe.LoggerExtension;
+import com.iluwatar.publish.subscribe.model.Message;
+import com.iluwatar.publish.subscribe.model.Topic;
+import com.iluwatar.publish.subscribe.publisher.Publisher;
+import com.iluwatar.publish.subscribe.publisher.PublisherImpl;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SubscriberTest {
+
+  private static final Logger logger = LoggerFactory.getLogger(SubscriberTest.class);
+  @RegisterExtension public LoggerExtension loggerExtension = new LoggerExtension();
+
+  private static final String TOPIC_WEATHER = "WEATHER";
+  private static final String TOPIC_TEMPERATURE = "TEMPERATURE";
+  private static final String TOPIC_CUSTOMER_SUPPORT = "CUSTOMER_SUPPORT";
+
+  @Test
+  void testSubscribeToMultipleTopics() {
+
+    Topic topicWeather = new Topic(TOPIC_WEATHER);
+    Topic topicTemperature = new Topic(TOPIC_TEMPERATURE);
+    Subscriber weatherSubscriber = new WeatherSubscriber();
+
+    topicWeather.addSubscriber(weatherSubscriber);
+    topicTemperature.addSubscriber(weatherSubscriber);
+
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(topicWeather);
+    publisher.registerTopic(topicTemperature);
+
+    publisher.publish(topicWeather, new Message("earthquake"));
+    publisher.publish(topicTemperature, new Message("-2C"));
+
+    waitForOutput();
+    assertEquals(2, loggerExtension.getFormattedMessages().size());
+  }
+
+  @Test
+  void testOnlyReceiveSubscribedTopic() {
+
+    Topic weatherTopic = new Topic(TOPIC_WEATHER);
+    Subscriber weatherSubscriber = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSubscriber);
+
+    Topic customerSupportTopic = new Topic(TOPIC_CUSTOMER_SUPPORT);
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(weatherTopic);
+    publisher.registerTopic(customerSupportTopic);
+
+    publisher.publish(customerSupportTopic, new Message("support@test.de"));
+
+    waitForOutput();
+    assertEquals(0, loggerExtension.getFormattedMessages().size());
+  }
+
+  @Test
+  void testMultipleSubscribersOnSameTopic() {
+
+    Topic weatherTopic = new Topic(TOPIC_WEATHER);
+    Subscriber weatherSubscriber1 = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSubscriber1);
+
+    Subscriber weatherSubscriber2 = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSubscriber2);
+
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(weatherTopic);
+
+    publisher.publish(weatherTopic, new Message("tornado"));
+
+    waitForOutput();
+    assertEquals(2, loggerExtension.getFormattedMessages().size());
+    assertEquals(
+        "Weather Subscriber: " + weatherSubscriber1.hashCode() + " issued message: tornado",
+        getMessage(weatherSubscriber1.hashCode()));
+    assertEquals(
+        "Weather Subscriber: " + weatherSubscriber2.hashCode() + " issued message: tornado",
+        getMessage(weatherSubscriber2.hashCode()));
+  }
+
+  @Test
+  void testMultipleSubscribersOnDifferentTopics() {
+
+    Topic weatherTopic = new Topic(TOPIC_WEATHER);
+    Subscriber weatherSubscriber = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSubscriber);
+
+    Topic customerSupportTopic = new Topic(TOPIC_CUSTOMER_SUPPORT);
+    Subscriber customerSupportSubscriber = new CustomerSupportSubscriber();
+    customerSupportTopic.addSubscriber(customerSupportSubscriber);
+
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(weatherTopic);
+    publisher.registerTopic(customerSupportTopic);
+
+    publisher.publish(weatherTopic, new Message("flood"));
+    publisher.publish(customerSupportTopic, new Message("support@test.at"));
+
+    waitForOutput();
+    assertEquals(2, loggerExtension.getFormattedMessages().size());
+    assertEquals(
+        "Weather Subscriber: " + weatherSubscriber.hashCode() + " issued message: flood",
+        getMessage(weatherSubscriber.hashCode()));
+    assertEquals(
+        "Customer Support Subscriber: "
+            + customerSupportSubscriber.hashCode()
+            + " sent the email to: support@test.at",
+        getMessage(customerSupportSubscriber.hashCode()));
+  }
+
+  @Test
+  void testInvalidContentOnTopics() {
+
+    Topic weatherTopic = new Topic(TOPIC_WEATHER);
+    Subscriber weatherSubscriber = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSubscriber);
+
+    Topic customerSupportTopic = new Topic(TOPIC_CUSTOMER_SUPPORT);
+    Subscriber customerSupportSubscriber = new CustomerSupportSubscriber();
+    customerSupportTopic.addSubscriber(customerSupportSubscriber);
+
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(weatherTopic);
+    publisher.registerTopic(customerSupportTopic);
+
+    publisher.publish(weatherTopic, new Message(123));
+    publisher.publish(customerSupportTopic, new Message(34.56));
+
+    waitForOutput();
+    assertTrue(loggerExtension.getFormattedMessages().getFirst().contains("Unknown content type"));
+    assertTrue(loggerExtension.getFormattedMessages().get(1).contains("Unknown content type"));
+  }
+
+  @Test
+  void testUnsubscribe() {
+
+    Topic weatherTopic = new Topic(TOPIC_WEATHER);
+    Subscriber weatherSubscriber = new WeatherSubscriber();
+    weatherTopic.addSubscriber(weatherSubscriber);
+
+    Publisher publisher = new PublisherImpl();
+    publisher.registerTopic(weatherTopic);
+
+    publisher.publish(weatherTopic, new Message("earthquake"));
+
+    weatherTopic.removeSubscriber(weatherSubscriber);
+    publisher.publish(weatherTopic, new Message("tornado"));
+
+    waitForOutput();
+    assertEquals(1, loggerExtension.getFormattedMessages().size());
+    assertTrue(loggerExtension.getFormattedMessages().getFirst().contains("earthquake"));
+    assertFalse(loggerExtension.getFormattedMessages().getFirst().contains("tornado"));
+  }
+
+  private String getMessage(int subscriberHash) {
+    Optional<String> message =
+        loggerExtension.getFormattedMessages().stream()
+            .filter(str -> str.contains(String.valueOf(subscriberHash)))
+            .findFirst();
+    assertTrue(message.isPresent());
+    return message.get();
+  }
+
+  private void waitForOutput() {
+    try {
+      TimeUnit.SECONDS.sleep(1);
+    } catch (InterruptedException e) {
+      logger.error("Interrupted!", e);
+      Thread.currentThread().interrupt();
+    }
+  }
+}