behavioral integration distributed systems

Message Broker

Reference Wikipedia ↗

The Message Broker pattern facilitates communication and data exchange between different applications, systems, and services. It acts as an intermediary, receiving messages from producers and routing them to interested consumers. This decoupling allows components to operate independently, improving scalability, resilience, and maintainability. Instead of direct point-to-point connections, components interact through the broker, enabling asynchronous communication and flexible integration.

Usage

The Message Broker pattern is widely used in scenarios requiring loose coupling and asynchronous communication. Common use cases include:

Event-Driven Architectures: Systems react to events published by other components. For example, a user registration event might trigger welcome email sending and profile creation in separate services.
Microservices Communication: Enabling independent microservices to exchange data without direct dependencies.
Background Task Processing: Offloading time-consuming tasks from the main application thread to be processed asynchronously by worker services.
Data Streaming: Handling high-volume, real-time data streams from various sources.
Integration with Legacy Systems: Providing a standardized interface for integrating newer applications with older, less flexible systems.

Examples

RabbitMQ: A popular open-source message broker that implements the Advanced Message Queuing Protocol (AMQP). It’s used extensively in enterprise applications for reliable message delivery, routing, and queuing. Many applications use RabbitMQ for task queues, asynchronous processing, and integrating disparate systems.
Apache Kafka: A distributed streaming platform often used as a message broker. Kafka is designed for high-throughput, fault-tolerant data pipelines and streaming applications. It’s commonly used in real-time data analytics, log aggregation, and event sourcing architectures, such as those found in LinkedIn and Netflix.
Amazon SQS (Simple Queue Service): A fully managed message queuing service offered by Amazon Web Services. It allows developers to decouple application components by using message queues to coordinate workflows. SQS is often used in serverless architectures and for building scalable, distributed systems.

Specimens

15 implementations

Specimen 111.01 Dart View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate through intermediary message channels (brokers). Components, termed publishers and subscribers, don’t need to know about each other directly. Publishers send messages to the broker, and subscribers receive messages they’ve registered for. This promotes scalability, flexibility and resilience. In Dart, this is elegantly implemented using Streams and StreamControllers, which naturally handle asynchronous message passing. The StreamController acts as our broker, allowing components to broadcast and listen for events without tight coupling. This approach leverages Dart’s asynchronous-first design and stream manipulation capabilities.

// message_broker.dart
import 'dart:async';

class MessageBroker {
  final _controller = StreamController<String>();

  Stream<String> get stream => _controller.stream;

  void publish(String message) {
    _controller.add(message);
  }

  void close() {
    _controller.close();
  }
}

// publisher.dart
import 'message_broker.dart';

class Publisher {
  final MessageBroker broker;
  final String name;

  Publisher(this.broker, this.name);

  void sendMessage(String message) {
    broker.publish('$name: $message');
  }
}

// subscriber.dart
import 'dart:async';
import 'message_broker.dart';

class Subscriber {
  final String name;
  final MessageBroker broker;

  Subscriber(this.name, this.broker) {
    broker.stream.listen((message) {
      print('$name received: $message');
    });
  }
}

// main.dart
import 'message_broker.dart';
import 'publisher.dart';
import 'subscriber.dart';

void main() {
  final broker = MessageBroker();
  final publisher1 = Publisher(broker, 'News Source 1');
  final publisher2 = Publisher(broker, 'News Source 2');
  final subscriber1 = Subscriber('User A', broker);
  final subscriber2 = Subscriber('User B', broker);

  publisher1.sendMessage('Breaking news: Dart is awesome!');
  publisher2.sendMessage('Another update: Flutter is growing fast!');

  // Allow time for messages to be processed before closing
  Future.delayed(Duration(seconds: 2), () => broker.close());
}

Specimen 111.02 Scala View specimen ↗

The Message Broker pattern decouples system components by enabling asynchronous communication through a central message bus. Components don’t directly interact; instead, they publish messages to topics/queues, and other components subscribe to those topics/queues to receive and process messages. This promotes loose coupling, scalability, and resilience. Here, we simulate a simple broker using actors in Scala, providing topic-based publishing and subscription. Actors naturally handle asynchronous messaging, making them a good fit. Using case classes for messages ensures type safety and a concise message format, typical of Scala.

import akka.actor._
import scala.collection.mutable

object MessageBroker {
  // Messages
  sealed trait BrokerMessage
  case class Subscribe(topic: String, subscriber: ActorRef) extends BrokerMessage
  case class Publish(topic: String, message: String) extends BrokerMessage

  // Actor definition
  class Broker extends Actor {
    private val subscribers = mutable.Map[String, Set[ActorRef]]()

    override def receive: Receive = {
      case Subscribe(topic, subscriber) =>
        subscribers.getOrElseUpdate(topic, Set.empty) += subscriber
        println(s"Subscriber $subscriber registered for topic: $topic")

      case Publish(topic, message) =>
        subscribers.get(topic) match {
          case Some(subs) =>
            subs.foreach(_ ! message)
            println(s"Published message '$message' to topic: $topic")
          case None =>
            println(s"No subscribers for topic: $topic")
        }
    }
  }

  // Example Subscribers
  class Subscriber(topic: String, broker: ActorRef) extends Actor {
    broker ! Subscribe(topic, self)

    override def receive: Receive = {
      case message: String =>
        println(s"$self received message: $message from topic: $topic")
    }
  }

  def main(args: Array[String]): Unit = {
    val system = ActorSystem("MessageBrokerSystem")
    val broker = system.actorOf(Props[Broker], "messageBroker")

    val subscriber1 = system.actorOf(Props(new Subscriber("news", broker)), "subscriber1")
    val subscriber2 = system.actorOf(Props(new Subscriber("sports", broker)), "subscriber2")
    val subscriber3 = system.actorOf(Props(new Subscriber("news", broker)), "subscriber3")

    broker ! Publish("news", "Breaking news: Scala is amazing!")
    broker ! Publish("sports", "Local team wins championship!")
    broker ! Publish("politics", "No subscribers here!")

    system.terminate()
  }
}

Specimen 111.03 PHP View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate through an intermediary message store. Components don’t directly call each other; instead, they publish messages to the broker, and other components subscribe to specific message types to receive and process them. This enhances scalability, fault tolerance, and flexibility as components can be added or removed without impacting others.

This PHP implementation uses a simple array-based message store and callback functions for subscriptions. MessageBroker::publish() adds a message to the store, triggering any registered subscribers for that message type. MessageBroker::subscribe() registers a callback associated with a message type. It’s a basic, in-memory broker suited for illustrating the concept; real-world scenarios would utilize a more robust queueing system like RabbitMQ or Redis. Using static methods for broker management and callable/closures for subscribers are common PHP practices.

<?php

class MessageBroker
{
    private static array $subscriptions = [];
    private static array $messages = [];

    public static function subscribe(string $type, callable $callback): void
    {
        if (!isset(self::$subscriptions[$type])) {
            self::$subscriptions[$type] = [];
        }
        self::$subscriptions[$type][] = $callback;
    }

    public static function publish(string $type, mixed $data): void
    {
        self::$messages[$type][] = $data;

        if (isset(self::$subscriptions[$type])) {
            foreach (self::$subscriptions[$type] as $callback) {
                $callback($data);
            }
        }
    }

    public static function getMessages(string $type): array
    {
        return self::$messages[$type] ?? [];
    }
}

// Example Usage
$logHandler = function (mixed $message) {
    echo "Log: " . json_encode($message) . PHP_EOL;
};

$emailHandler = function (mixed $message) {
    echo "Email: " . $message['email'] . PHP_EOL;
};

MessageBroker::subscribe('logEvent', $logHandler);
MessageBroker::subscribe('userCreated', $emailHandler);

MessageBroker::publish('logEvent', ['timestamp' => time(), 'message' => 'Example log message']);
MessageBroker::publish('userCreated', ['name' => 'Alice', 'email' => 'alice@example.com']);

// Retrieve messages for a specific type
$logMessages = MessageBroker::getMessages('logEvent');
print_r($logMessages);

?>

Specimen 111.04 Ruby View specimen ↗

The Message Broker pattern decouples application components by allowing them to communicate via messages, without direct dependencies. A central message broker receives messages from publishers and routes them to appropriate subscribers. This promotes scalability, flexibility, and fault tolerance. My Ruby implementation uses a simple hash-based “broker” to store subscribers and their associated topics. Publishers send messages to the broker, which then iterates through subscribers, delivering messages to those interested in the published topic. This utilizes Ruby’s flexible hash structure and each method for efficient subscriber notification, fitting its functional style for message distribution.

# message_broker.rb

class MessageBroker
  def initialize
    @subscribers = {} # topic => [subscriber_procs]
  end

  def subscribe(topic, &subscriber)
    (@subscribers[topic] ||= []) << subscriber
  end

  def publish(topic, message)
    @subscribers[topic]&.each { |subscriber| subscriber.call(message) }
  end
end

# Example Usage:

# Create a broker instance
broker = MessageBroker.new

# Define some subscribers
subscriber1 = ->(message) { puts "Subscriber 1 received: #{message}" }
subscriber2 = ->(message) { puts "Subscriber 2 received: #{message}" }

# Subscribe them to topics
broker.subscribe("news", &subscriber1)
broker.subscribe("news", &subscriber2)
broker.subscribe("weather", &subscriber1)

# Publish some messages
broker.publish("news", "Breaking news: Ruby is awesome!")
broker.publish("weather", "It's sunny today.")
broker.publish("sports", "No subscribers for this topic.")

Specimen 111.05 Swift View specimen ↗

The Message Broker pattern decouples application components by allowing them to communicate through intermediary message channels. Components don’t need to know about each other directly; they simply publish messages to the broker, and other components subscribe to the messages they’re interested in. This promotes scalability, maintainability, and flexibility.

This Swift implementation uses a simple NotificationCenter as the message broker. Publisher structs define message types. Components publish messages using the post function of NotificationCenter, and subscribe by observing notifications for specific message types. This approach leverages Swift’s built-in event handling mechanism, making it concise and idiomatic. The use of structs for messages aligns with Swift’s value-type philosophy, and the NotificationCenter handles the decoupling efficiently.

// Define message types as structs
struct UserCreated {
    let userId: Int
    let username: String
}

struct ProductPurchased {
    let productId: Int
    let userId: Int
}

// Message Broker (using NotificationCenter)
import Foundation

class MessageBroker {
    static let shared = MessageBroker()

    private init() {}

    func publish<T: Hashable>(message: T, userInfo: [AnyHashable : Any]? = nil) {
        NotificationCenter.default.post(name: NSNotification.Name(rawValue: String(describing: message)), object: nil, userInfo: userInfo)
    }

    func subscribe<T: Hashable>(observer: Any, selector: Selector, messageType: T.Type) {
        NotificationCenter.default.addObserver(observer, selector: selector, name: NSNotification.Name(rawValue: String(describing: messageType)), object: nil)
    }
}

// Example Component 1: User Service
class UserService {
    func createUser(userId: Int, username: String) {
        print("Creating user \(username) with ID \(userId)")
        let message = UserCreated(userId: userId, username: username)
        MessageBroker.shared.publish(message: message)
    }
}

// Example Component 2: Email Service
class EmailService {
    @objc func handleUserCreated(notification: Notification) {
        guard let userInfo = notification.userInfo, let userId = userInfo["userId"] as? Int, let username = userInfo["username"] as? String else {
            return
        }
        print("Sending welcome email to \(username) (ID: \(userId))")
    }

    func setupSubscription() {
        MessageBroker.shared.subscribe(observer: self, selector: #selector(handleUserCreated), messageType: UserCreated.self)
    }
}

// Example Usage
let userService = UserService()
let emailService = EmailService()
emailService.setupSubscription()

userService.createUser(userId: 123, username: "Alice")

Specimen 111.06 Kotlin View specimen ↗

The Message Broker pattern decouples components of an application by allowing them to communicate through a central message bus. This promotes scalability and maintainability as components aren’t directly dependent on each other. Here, we use Kotlin Coroutines and Channels to implement a simple in-memory message broker. MessageBroker acts as the bus, with publish and subscribe functions. publish sends messages to the channel, and subscribe creates a coroutine that receives messages. Kotlin’s coroutines provide a concise way to handle asynchronous message consumption, and Channels are a type-safe conduit for communication. This approach leverages Kotlin’s concurrency features for a clean and efficient implementation.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

class MessageBroker {
    private val channel = Channel<String>()

    suspend fun publish(message: String) {
        channel.send(message)
    }

    fun subscribe(): ReceiveChannel<String> {
        return channel
    }
}

fun main() = runBlocking {
    val broker = MessageBroker()

    // Subscriber 1
    launch {
        broker.subscribe().collect { message ->
            println("Subscriber 1 received: $message")
        }
    }

    // Subscriber 2
    launch {
        broker.subscribe().collect { message ->
            println("Subscriber 2 received: $message")
        }
    }

    // Publisher
    coroutineScope {
        launch {
            broker.publish("Hello, Kotlin!")
            delay(500)
            broker.publish("Another message")
            delay(500)
            broker.publish("Last message")
        }
    }
}

Specimen 111.07 Rust View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate asynchronously through a central message intermediary, the broker. This promotes scalability and resilience. The Rust implementation uses channels (specifically mpsc - multiple producer, single consumer) to simulate the broker. Producers send messages to the broker’s transmitting end, and the consumer receives them from the receiving end. This example demonstrates a simple broker with a string message type. It’s idiomatic Rust because it leverages ownership and borrowing through the channel, ensuring memory safety, and avoids shared mutable state, aligning with Rust’s concurrency model. The use of move ensures ownership is transferred to the thread.

use std::sync::mpsc;
use std::thread;

// Define a type for a message handler (function pointer)
type MessageHandler = fn(&str);

struct MessageBroker {
    sender: mpsc::Sender<String>,
}

impl MessageBroker {
    fn new() -> Self {
        let (sender, receiver) = mpsc::channel();

        // Spawn a thread to handle incoming messages
        thread::spawn(move || {
            let mut handlers: Vec<MessageHandler> = Vec::new();

            // Example handler registration (could be more dynamic)
            handlers.push(|msg| println!("Handler 1 received: {}", msg));
            handlers.push(|msg| println!("Handler 2 received: {}", msg));
            
            for msg in receiver {
                for handler in &handlers {
                    handler(&msg);
                }
            }
        });

        MessageBroker { sender }
    }

    fn publish(&self, message: String) {
        self.sender.send(message).unwrap();
    }
}

fn main() {
    let broker = MessageBroker::new();

    broker.publish("Hello, world!".to_string());
    broker.publish("Another message!".to_string());
    broker.publish("Rust message broker example".to_string());
    
    // Keep the main thread alive long enough for the broker thread to process messages.
    thread::sleep(std::time::Duration::from_millis(100));
}

Specimen 111.08 Go View specimen ↗

The Message Broker pattern decouples application components by allowing them to communicate via intermediary message queues. Publishers send messages to a topic (or exchange) without knowing about subscribers. Subscribers express interest in specific topics and receive messages as they arrive. This promotes scalability and resilience, as components can be added or removed without impacting others.

This Go implementation uses channels to act as message queues. A Broker struct manages a map of topics (strings) to channels. Publish sends a message to all subscribed channels. Subscribe returns a channel for a given topic, allowing consumers to listen for messages. This approach is idiomatic Go due to its emphasis on concurrency via channels and lightweight, composable structures. Goroutines would typically handle publishing and subscribing in a real application.

package main

import "fmt"

// Broker manages message channels for different topics.
type Broker struct {
	topics map[string]chan interface{}
	mutex  chan struct{} // Simplified mutex for channel safety
}

// NewBroker creates a new Message Broker instance.
func NewBroker() *Broker {
	return &Broker{
		topics: make(map[string]chan interface{}),
		mutex:  make(chan struct{}, 1), // Buffered channel acts as a mutex
	}
}

// Publish sends a message to all subscribers of a topic.
func (b *Broker) Publish(topic string, message interface{}) {
	b.mutex <- struct{}{} // Acquire lock
	defer func() {
		<-b.mutex // Release lock
	}()

	subscribers, ok := b.topics[topic]
	if !ok {
		return // No subscribers for this topic
	}
	for _, sub := range b.topics[topic] {
		go func(ch chan interface{}) {
			ch <- message
		}(sub)
	}
}

// Subscribe returns a channel for a given topic.  If the topic doesn't exist, it's created.
func (b *Broker) Subscribe(topic string) <-chan interface{} {
	b.mutex <- struct{}{} // Acquire lock
	defer func() {
		<-b.mutex // Release lock
	}()

	if _, ok := b.topics[topic]; !ok {
		b.topics[topic] = make(chan interface{})
	}
	return b.topics[topic]
}

func main() {
	broker := NewBroker()

	topic1 := "news"
	sub1 := broker.Subscribe(topic1)
	sub2 := broker.Subscribe(topic1)

	go func() {
		broker.Publish(topic1, "Breaking news: Go is awesome!")
	}()

	go func() {
		msg1 := <-sub1
		fmt.Println("Subscriber 1 received:", msg1)
	}()

	go func() {
		msg2 := <-sub2
		fmt.Println("Subscriber 2 received:", msg2)
	}()

	// Keep the program running to receive messages
	var input string
	fmt.Scanln(&input)
}

Specimen 111.09 C View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate via asynchronous messages. A central message broker receives messages from publishers and routes them to appropriate subscribers. This promotes scalability, fault tolerance, and flexibility.

The C implementation uses a simple circular buffer as the message queue and function pointers to represent publishers and subscribers. A message_broker_t struct manages the queue and registration of callbacks. publish() adds a message to the queue, and a separate thread (or polling loop) calls registered subscriber functions when messages are available. This approach avoids busy-waiting and allows components to operate independently. Using function pointers is a common C technique for achieving callback-based event handling, fitting the language’s procedural nature.

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

#define QUEUE_SIZE 10

typedef void (*subscriber_callback_t)(const char *message);

typedef struct {
    subscriber_callback_t subscribers[10];
    int subscriber_count;
    char queue[QUEUE_SIZE][256];
    int head;
    int tail;
    pthread_mutex_t mutex;
} message_broker_t;

message_broker_t *message_broker_create() {
    message_broker_t *broker = (message_broker_t *)malloc(sizeof(message_broker_t));
    if (broker) {
        broker->subscriber_count = 0;
        broker->head = 0;
        broker->tail = 0;
        pthread_mutex_init(&broker->mutex, NULL);
    }
    return broker;
}

void message_broker_subscribe(message_broker_t *broker, subscriber_callback_t callback) {
    if (broker && callback && broker->subscriber_count < 10) {
        pthread_mutex_lock(&broker->mutex);
        broker->subscribers[broker->subscriber_count++] = callback;
        pthread_mutex_unlock(&broker->mutex);
    }
}

void message_broker_publish(message_broker_t *broker, const char *message) {
    if (broker && message) {
        pthread_mutex_lock(&broker->mutex);
        strcpy(broker->queue[broker->tail], message);
        broker->tail = (broker->tail + 1) % QUEUE_SIZE;
        pthread_mutex_unlock(&broker->mutex);
    }
}

void *message_broker_worker(void *arg) {
    message_broker_t *broker = (message_broker_t *)arg;
    while (1) {
        pthread_mutex_lock(&broker->mutex);
        if (broker->head != broker->tail) {
            char message[256];
            strcpy(message, broker->queue[broker->head]);
            broker->head = (broker->head + 1) % QUEUE_SIZE;
            pthread_mutex_unlock(&broker->mutex);

            for (int i = 0; i < broker->subscriber_count; i++) {
                broker->subscribers[i](message);
            }
        } else {
            pthread_mutex_unlock(&broker->mutex);
            usleep(100000); // Sleep for 100ms to avoid busy-waiting
        }
    }
    return NULL;
}

void message_broker_destroy(message_broker_t *broker) {
    if (broker) {
        pthread_mutex_destroy(&broker->mutex);
        free(broker);
    }
}

// Example Subscribers
void subscriber1(const char *message) {
    printf("Subscriber 1 received: %s\n", message);
}

void subscriber2(const char *message) {
    printf("Subscriber 2 received: %s\n", message);
}

int main() {
    message_broker_t *broker = message_broker_create();

    message_broker_subscribe(broker, subscriber1);
    message_broker_subscribe(broker, subscriber2);

    pthread_t worker_thread;
    pthread_create(&worker_thread, NULL, message_broker_worker, broker);

    // Publish some messages
    message_broker_publish(broker, "Hello, world!");
    message_broker_publish(broker, "This is a test message.");
    message_broker_publish(broker, "Another message for subscribers.");

    sleep(2); // Let the messages be processed

    message_broker_destroy(broker);
    pthread_join(worker_thread, NULL);

    return 0;
}

Specimen 111.10 C++ View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate through intermediary message queues. Components, or “clients,” don’t directly call each other; instead, they publish messages to topics and subscribe to topics of interest. A central broker manages message delivery. This promotes scalability, flexibility, and resilience.

The C++ implementation uses a MessageBroker class to hold topic-to-subscriber mappings. Clients register with the broker to receive messages for specific topics. The publish method iterates through subscribers of a topic and delivers the message. Using std::function allows for flexible subscriber callback types. This approach leverages C++’s standard library for efficient data structures and function objects, fitting its object-oriented nature and promoting loose coupling.

#include <iostream>
#include <vector>
#include <functional>
#include <map>
#include <string>

class MessageBroker {
public:
    using Callback = std::function<void(const std::string&)>;

    void subscribe(const std::string& topic, Callback callback) {
        subscribers[topic].push_back(callback);
    }

    void publish(const std::string& topic, const std::string& message) {
        for (auto& callback : subscribers[topic]) {
            callback(message);
        }
    }

private:
    std::map<std::string, std::vector<Callback>> subscribers;
};

// Example Clients
class ClientA {
public:
    void onMessage(const std::string& message) {
        std::cout << "Client A received: " << message << std::endl;
    }

    void registerWithBroker(MessageBroker& broker, const std::string& topic) {
        broker.subscribe(topic, std::bind(&ClientA::onMessage, this, std::placeholders::_1));
    }
};

class ClientB {
public:
    void onMessage(const std::string& message) {
        std::cout << "Client B received: " << message << std::endl;
    }

    void registerWithBroker(MessageBroker& broker, const std::string& topic) {
        broker.subscribe(topic, [this](const std::string& msg){ onMessage(msg); });
    }
};

int main() {
    MessageBroker broker;
    ClientA clientA;
    ClientB clientB;

    clientA.registerWithBroker(broker, "news");
    clientB.registerWithBroker(broker, "news");
    clientB.registerWithBroker(broker, "sports");

    broker.publish("news", "Breaking news: C++ is awesome!");
    broker.publish("sports", "Local team wins championship!");
    broker.publish("weather", "Sunny today!"); // No subscribers

    return 0;
}

Specimen 111.11 C_sharp View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate through intermediary message queues. Components don’t need to know about each other; they simply send and receive messages. This promotes scalability, resilience, and flexibility.

This C# example uses a simple MessageBroker class that holds a list of message subscriptions (handlers). SendMessage iterates through relevant subscriptions and invokes their HandleMessage methods. We define a concrete OrderService and EmailService which subscribe to messages of type OrderCreatedMessage. This implementation leverages C# delegates for flexible message handling and is a common approach for event-driven architectures in C#. The use of interfaces (IMessageHandler) further enhances decoupling.

// IMessageHandler.cs
public interface IMessageHandler
{
    Type GetMessageType();
    void HandleMessage(object message);
}

// MessageBroker.cs
public class MessageBroker
{
    private readonly List<IMessageHandler> _handlers = new();

    public void Subscribe<T>(IMessageHandler handler)
    {
        _handlers.Add(handler);
    }

    public void Unsubscribe<T>(IMessageHandler handler)
    {
        _handlers.Remove(handler);
    }

    public void SendMessage(object message)
    {
        var messageType = message.GetType();
        foreach (var handler in _handlers)
        {
            if (handler.GetMessageType() == messageType)
            {
                handler.HandleMessage(message);
            }
        }
    }
}

// OrderCreatedMessage.cs
public class OrderCreatedMessage
{
    public int OrderId { get; }
    public decimal TotalAmount { get; }

    public OrderCreatedMessage(int orderId, decimal totalAmount)
    {
        OrderId = orderId;
        TotalAmount = totalAmount;
    }
}

// OrderService.cs
public class OrderService : IMessageHandler
{
    public Type GetMessageType() => typeof(OrderCreatedMessage);

    public void HandleMessage(object message)
    {
        var orderMessage = (OrderCreatedMessage)message;
        Console.WriteLine($"Order Service: Order created with ID {orderMessage.OrderId}, Amount: {orderMessage.TotalAmount}");
    }
}

// EmailService.cs
public class EmailService : IMessageHandler
{
    public Type GetMessageType() => typeof(OrderCreatedMessage);

    public void HandleMessage(object message)
    {
        var orderMessage = (OrderCreatedMessage)message;
        Console.WriteLine($"Email Service: Sending confirmation email for Order ID {orderMessage.OrderId}");
    }
}

// Example Usage
public class Program
{
    public static void Main(string[] args)
    {
        var broker = new MessageBroker();
        var orderService = new OrderService();
        var emailService = new EmailService();

        broker.Subscribe(orderService);
        broker.Subscribe(emailService);

        broker.SendMessage(new OrderCreatedMessage(123, 50.00m));
    }
}

Specimen 111.12 TypeScript View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate via messages. Components don’t need to know about each other directly; they interact through a central message broker. This promotes scalability, flexibility, and resilience.

This TypeScript implementation uses a simple MessageBroker class to manage message channels (topics). Components subscribe to channels and publish messages to them. Subscribers receive messages via a callback function. The use of a Map to store subscribers per channel is efficient for lookup. TypeScript’s type safety is leveraged to ensure messages are handled correctly, and the asynchronous nature of callbacks aligns with TypeScript’s modern asynchronous programming style. This avoids tight coupling and allows for dynamic addition/removal of subscribers.

// message-broker.ts
type SubscriberCallback = (message: any) => void;

class MessageBroker {
  private subscribers: Map<string, SubscriberCallback[]> = new Map();

  subscribe(channel: string, callback: SubscriberCallback): void {
    if (!this.subscribers.has(channel)) {
      this.subscribers.set(channel, []);
    }
    this.subscribers.get(channel)!.push(callback);
  }

  publish(channel: string, message: any): void {
    if (this.subscribers.has(channel)) {
      const callbacks = this.subscribers.get(channel)!;
      callbacks.forEach(callback => callback(message));
    }
  }
}

// Example Usage
const broker = new MessageBroker();

// Subscriber 1
broker.subscribe('news', (message) => {
  console.log('Subscriber 1 received news:', message);
});

// Subscriber 2
broker.subscribe('news', (message) => {
  console.log('Subscriber 2 received news:', message);
});

// Subscriber to a different channel
broker.subscribe('alerts', (message) => {
  console.log('Alerts Subscriber received:', message);
});

// Publishing messages
broker.publish('news', { headline: 'TypeScript is awesome!', content: 'This is a great language.' });
broker.publish('alerts', { type: 'warning', message: 'Low disk space!' });

Specimen 111.13 JavaScript View specimen ↗

The Message Broker pattern decouples application components by enabling them to communicate through intermediary message queues. Components (producers) send messages to the broker without knowing who the consumers are, and consumers subscribe to specific message types to receive relevant updates. This promotes scalability, flexibility, and resilience.

This JavaScript implementation uses a simple object to represent the message broker. Producers publish messages to topics, and consumers subscribe to topics via a callback function. When a message is published, the broker iterates through the subscribers for that topic and invokes their callbacks. This approach leverages JavaScript’s first-class function capabilities and object-oriented nature for a clean and flexible design, avoiding tight coupling between components. It’s a common pattern for event-driven architectures in JavaScript, particularly in front-end frameworks and Node.js applications.

class MessageBroker {
  constructor() {
    this.topics = {};
  }

  subscribe(topic, callback) {
    if (!this.topics[topic]) {
      this.topics[topic] = [];
    }
    this.topics[topic].push(callback);
  }

  publish(topic, message) {
    if (this.topics[topic]) {
      this.topics[topic].forEach(callback => {
        callback(message);
      });
    }
  }
}

// Example Usage:
const broker = new MessageBroker();

// Subscriber 1
broker.subscribe('user.created', (user) => {
  console.log('User created:', user.name);
});

// Subscriber 2
broker.subscribe('user.created', (user) => {
  console.log('Sending welcome email to:', user.email);
});

// Subscriber 3 - different topic
broker.subscribe('order.placed', (order) => {
  console.log('Order placed:', order.id);
});

// Publishing messages
broker.publish('user.created', { name: 'Alice', email: 'alice@example.com' });
broker.publish('user.created', { name: 'Bob', email: 'bob@example.com' });
broker.publish('order.placed', { id: '12345' });

Specimen 111.14 Python View specimen ↗

The Message Broker pattern decouples application components by enabling asynchronous communication. Instead of components directly calling each other, they send and receive messages via a central message broker. This improves scalability, resilience, and flexibility. This example uses Python’s pubsub library for a simple implementation. Publishers “publish” messages to a topic without knowing who the subscribers are. Subscribers register their interest in specific topics and receive messages published to those topics. This approach is idiomatic Python due to its emphasis on readability and leveraging existing libraries for common tasks, avoiding unnecessary complexity.

# message_broker.py
import pubsub

class MessageBroker:
    def __init__(self):
        self.pub = pubsub.Publisher()
        self.sub = pubsub.Subscriber()

    def publish(self, topic, message):
        self.pub.publish(topic, message)

    def subscribe(self, topic, callback):
        self.sub.subscribe(topic, callback)
        return self.sub.unsubscribe  # Return unsubscribe function

def subscriber_function(message, topic):
    print(f"Subscriber received: {message} on topic: {topic}")

if __name__ == '__main__':
    broker = MessageBroker()

    # Subscribe to a topic
    unsubscribe = broker.subscribe("my_topic", subscriber_function)

    # Publish messages
    broker.publish("my_topic", "Hello, world!")
    broker.publish("my_topic", "Another message")
    broker.publish("another_topic", "This is a different topic")

    # Unsubscribe (optional)
    # unsubscribe()

Specimen 111.15 Java View specimen ↗

The Message Broker pattern decouples application components by enabling asynchronous communication through a central message store. Components don’t directly interact; instead, they publish messages to the broker, and other components subscribe to receive messages of interest. This promotes scalability, resilience, and flexibility.

This Java implementation uses a simple in-memory MessageBroker class to store and deliver messages. Publisher components publish messages with a topic, and Subscriber components register for specific topics. When a message is published, the broker iterates through its subscribers and delivers the message to those interested in the topic. Using interfaces (Publisher, Subscriber) allows for loose coupling and easy extension. The use of java.util.HashMap and java.util.ArrayList are standard Java collections for this purpose, making the code idiomatic.

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

interface Publisher {
    void publish(String topic, String message);
}

interface Subscriber {
    void receive(String topic, String message);
}

class MessageBroker {
    private final Map<String, List<Subscriber>> subscribers = new HashMap<>();

    public void subscribe(String topic, Subscriber subscriber) {
        subscribers.computeIfAbsent(topic, k -> new ArrayList<>()).add(subscriber);
    }

    public void unsubscribe(String topic, Subscriber subscriber) {
        subscribers.getOrDefault(topic, new ArrayList<>()).remove(subscriber);
    }

    public void publish(String topic, String message) {
        subscribers.getOrDefault(topic, new ArrayList<>())
                .forEach(subscriber -> subscriber.receive(topic, message));
    }
}

class ConcreteSubscriber implements Subscriber {
    private final String name;

    public ConcreteSubscriber(String name) {
        this.name = name;
    }

    @Override
    public void receive(String topic, String message) {
        System.out.println(name + " received on topic " + topic + ": " + message);
    }
}

public class MessageBrokerExample {
    public static void main(String[] args) {
        MessageBroker broker = new MessageBroker();

        ConcreteSubscriber sub1 = new ConcreteSubscriber("Subscriber 1");
        ConcreteSubscriber sub2 = new ConcreteSubscriber("Subscriber 2");

        broker.subscribe("news", sub1);
        broker.subscribe("news", sub2);
        broker.subscribe("sports", sub1);

        Publisher publisher = broker::publish;
        publisher.publish("news", "Breaking news: Java is awesome!");
        publisher.publish("sports", "Local team wins championship!");
    }
}