Publish-Subscribe Pattern: Decoupling Producers from Consumers

// VIOLATION: OrderService knows about every consumer directly — Observer-style,
// but now the "subject" is a business service that shouldn't own delivery logic
class OrderService {
    private InventoryService inventoryService;
    private EmailService emailService;
    private AnalyticsService analyticsService;
    // Adding a new consumer (e.g. FraudDetectionService) means OrderService
    // must be changed, redeployed, and re-tested — even though OrderService's
    // actual job (placing orders) hasn't changed at all.
 
    void placeOrder(Order order) {
        // save order...
        inventoryService.reserve(order);
        emailService.sendConfirmation(order);
        analyticsService.record(order);
    }
}

// A message on the wire — payload + minimal metadata
record Message(String topic, Object payload, long timestamp) {
    Message(String topic, Object payload) {
        this(topic, payload, System.currentTimeMillis());
    }
}
 
interface Subscriber {
    void onMessage(Message message);
}
 
interface Broker {
    void publish(String topic, Object payload);
    void subscribe(String topic, Subscriber subscriber);
    void unsubscribe(String topic, Subscriber subscriber);
}
 
// A simple in-process broker. Production brokers (Kafka, RabbitMQ, SNS/SQS)
// implement the same conceptual contract with persistence, delivery guarantees,
// and cross-process transport added.
class InMemoryBroker implements Broker {
    private final Map<String, List<Subscriber>> topics = new ConcurrentHashMap<>();
 
    @Override
    public void subscribe(String topic, Subscriber subscriber) {
        topics.computeIfAbsent(topic, t -> new CopyOnWriteArrayList<>()).add(subscriber);
    }
 
    @Override
    public void unsubscribe(String topic, Subscriber subscriber) {
        List<Subscriber> subs = topics.get(topic);
        if (subs != null) subs.remove(subscriber);
    }
 
    @Override
    public void publish(String topic, Object payload) {
        Message message = new Message(topic, payload);
        List<Subscriber> subs = topics.getOrDefault(topic, List.of());
        for (Subscriber subscriber : subs) {
            // isolate one bad subscriber from breaking delivery to the rest
            try {
                subscriber.onMessage(message);
            } catch (Exception e) {
                System.err.println("Subscriber failed on topic " + topic + ": " + e.getMessage());
            }
        }
    }
}
 
// Publisher: knows only the broker and a topic name — zero knowledge of subscribers
class OrderService {
    private final Broker broker;
    OrderService(Broker broker) { this.broker = broker; }
 
    void placeOrder(Order order) {
        // save order...
        broker.publish("order.placed", order);
    }
}
 
// Subscribers: know only the broker and a topic name — zero knowledge of the publisher
class InventorySubscriber implements Subscriber {
    @Override
    public void onMessage(Message message) {
        Order order = (Order) message.payload();
        System.out.println("[Inventory] Reserving stock for order " + order.id());
    }
}
 
class EmailSubscriber implements Subscriber {
    @Override
    public void onMessage(Message message) {
        Order order = (Order) message.payload();
        System.out.println("[Email] Sending confirmation for order " + order.id());
    }
}
 
record Order(String id) {}
 
class Demo {
    public static void main(String[] args) {
        Broker broker = new InMemoryBroker();
 
        broker.subscribe("order.placed", new InventorySubscriber());
        broker.subscribe("order.placed", new EmailSubscriber());
        // Adding FraudDetectionSubscriber later touches ZERO existing code —
        // just one new subscribe() call, and OrderService is never redeployed.
 
        OrderService orderService = new OrderService(broker);
        orderService.placeOrder(new Order("ORD-1001"));
    }
}

// Wildcard topic matching: "orders.*" matches "orders.placed", "orders.cancelled", etc.
class WildcardBroker implements Broker {
    private final Map<String, List<Subscriber>> patterns = new ConcurrentHashMap<>();
 
    @Override
    public void subscribe(String topicPattern, Subscriber subscriber) {
        patterns.computeIfAbsent(topicPattern, t -> new CopyOnWriteArrayList<>()).add(subscriber);
    }
 
    @Override
    public void unsubscribe(String topicPattern, Subscriber subscriber) {
        List<Subscriber> subs = patterns.get(topicPattern);
        if (subs != null) subs.remove(subscriber);
    }
 
    @Override
    public void publish(String topic, Object payload) {
        Message message = new Message(topic, payload);
        for (var entry : patterns.entrySet()) {
            if (matches(entry.getKey(), topic)) {
                entry.getValue().forEach(s -> s.onMessage(message));
            }
        }
    }
 
    private boolean matches(String pattern, String topic) {
        String regex = pattern.replace(".", "\\.").replace("*", ".*");
        return topic.matches(regex);
    }
}