LocalBus

A message bus (or enterprise service bus) is an architectural pattern that enables different parts of an application to communicate without direct knowledge of each other. Think of it as a smart postal service for your application - components can send messages to topics, and other components can listen for those messages, all without knowing about each other directly.

Even within a single process, this pattern offers powerful benefits:

• Decouple Components: Break complex systems into maintainable parts that can evolve independently
• Single Responsibility: Each component can focus on its core task without handling cross-cutting concerns
• Flexible Architecture: Easily add new features by subscribing to existing events without modifying original code
• Control Flow: Choose immediate or background processing based on your needs
• Testing: Simplified testing as components can be tested in isolation
• Stay Reliable: Built-in error handling and thread safety
• Non-Blocking: Efficient message processing with async I/O

• Key Benefits
  • Performance and Efficiency
  • Ease of Use
  • Decoupling and Modularity
  • Reliability and Safety
• Use Cases
• Key Components
  • Bus
  • Station
  • LocalBus
• Installation
  • Requirements
• Usage
  • LocalBus
  • Bus
  • Station
• Advanced Usage
  • Concurrency Controls
    • Bus
    • Station
      • Message Priority
  • Error Handling
  • Memory Considerations
  • Blocking Operations
  • Shutdown & Cleanup
  • Limitations
  • Demos & Benchmarks
• See Also

LocalBus offers several advantages that make it an attractive choice for Ruby developers looking to implement a pub/sub system within a single process:

• Non-Blocking I/O: Leveraging the power of the Async library, LocalBus ensures efficient message processing without blocking the main thread, leading to improved performance in I/O-bound applications.
• Optimized Resource Usage: By using semaphores and thread pools, LocalBus efficiently manages system resources, allowing for high concurrency without overwhelming the system.

• Simple Setup: With straightforward installation and intuitive API, LocalBus allows developers to quickly integrate pub/sub capabilities into their applications.
• Minimal Configuration: Default settings are optimized for most use cases, reducing the need for complex configurations.

• Component Isolation: LocalBus enables clean separation of concerns by allowing components to communicate through messages without direct dependencies or tight coupling.
• Scalable Architecture: Easily extend your application by adding new subscribers to existing topics, facilitating the addition of new features without modifying existing code.

• Built-in Error Handling: LocalBus includes error boundaries to ensure that failures in one subscriber do not affect others, maintaining system stability.
• Thread Safety: Designed with concurrency in mind, LocalBus provides thread-safe operations to prevent race conditions and ensure data integrity.

LocalBus is versatile and can be applied to various scenarios within a Ruby application. Here are some common use cases and examples:

# Component A subscribes to order creation events
LocalBus.subscribe "order.created" do |message|
  InventoryService.update_stock message.payload[:order_id]
end

# Component B publishes an order creation event
LocalBus.publish "order.created", order_id: 789

# Subscribe to user sign-up events
LocalBus.subscribe "user.signed_up" do |message|
  NotificationService.send_welcome_email message.payload[:user_id]
end

# Publish a user sign-up event
LocalBus.publish "user.signed_up", user_id: 123

# Subscribe to report generation requests
LocalBus.subscribe "report.generate" do |message|
  ReportService.generate message.payload[:report_id]
end

# Publish a report generation request
LocalBus.publish "report.generate", report_id: 456

The Bus acts as a direct transport mechanism for messages, akin to placing a passenger directly onto a bus. When a message is published to the Bus, it is immediately delivered to all subscribers, ensuring prompt execution of tasks. This is achieved through non-blocking I/O operations, which allow the Bus to handle multiple tasks efficiently without blocking the main thread.

The Station serves as a queuing system for messages, similar to a bus station where passengers wait for their bus.

When a message is published to the Station, it is queued and processed at a later time, allowing for deferred execution. This is particularly useful for tasks that can be handled later.

The Station employs a thread pool to manage message processing, enabling high concurrency and efficient resource utilization. Messages can also be prioritized, ensuring that higher-priority tasks are processed first.

The LocalBus class serves as the primary interface to the library, providing a convenient singleton pattern for accessing both Bus and Station functionality. It exposes singleton instances of both Bus and Station providing a simplified API for common pub/sub operations.

By default, LocalBus delegates to the Station singleton for all pub/sub operations, making it ideal for background processing scenarios. This means that when you use LocalBus.publish or LocalBus.subscribe, you're actually working with default Station, benefiting from its queuing and thread pool capabilities.

bundle add local_bus

• Ruby >= 3.0

LocalBus.subscribe "user.created" do |message|
  # business logic (e.g. API calls, database queries, disk operations, etc.)
  "It worked!"
end

message = LocalBus.publish("user.created", user_id: 123)
message.wait        # blocks until all subscribers complete
message.subscribers # blocks and waits until all subscribers complete and returns the subscribers
#=> [#<LocalBus::Subscriber:0x0000000120f75c30 ...>]

message.subscribers.first.value
#=> "It worked!"

# subscribe with any object that responds to `#call`.
worker = ->(message) do
  # business logic (e.g. API calls, database queries, disk operations, etc.)
  "It worked!"
end
LocalBus.subscribe "user.created", callable: worker

bus = LocalBus::Bus.new # ... or LocalBus.instance.bus

# register a subscriber
bus.subscribe "user.created" do |message|
  # business logic (e.g. API calls, database queries, disk operations, etc.)
  "It worked!"
end

message = bus.publish("user.created", user_id: 123)
message.wait        # blocks until all subscribers complete
message.subscribers # waits and returns the subscribers
#=> [#<LocalBus::Subscriber:0x000000012bbb79a8 ...>]

message.subscribers.first.value
#=> "It worked!"

station = LocalBus::Station.new # ... or LocalBus.instance.station

station.subscribe "user.created" do |message|
  # business logic (e.g. API calls, database queries, disk operations, etc.)
  "It worked!"
end

message = station.publish("user.created", user_id: 123)
message.wait        # blocks until all subscribers complete
message.subscribers # blocks and waits until all subscribers complete and returns the subscribers
#=> [#<LocalBus::Subscriber:0x00000001253156e8 ...>]

message.subscribers.first.value
#=> "It worked!"

The Bus leverages Async's Semaphore to limit resource consumption. The configured concurrency limits how many operations can run at once.

# Configure concurrency limits for the Bus (default: Etc.nprocessors)
bus = LocalBus::Bus.new(concurrency: 10)

The Station uses a thread pool for multi-threaded message processing. You can configure the queue size and the number of threads used to process messages.

# Configure the Station
station = LocalBus::Station.new(
  limit: 5_000, # max number of pending messages (default: 10_000)
  threads: 10,  # max number of processing threads (default: Etc.nprocessors)
)

The Station supports assigning a priority to each message. Messages with a higher priority are processed before lower priority messages.

LocalBus.publish("default")                # 3rd to process
LocalBus.publish("important", priority: 5) # 2nd to process
LocalBus.publish("critical", priority: 10) # 1st to process

Error boundaries prevent individual subscriber failures from affecting other subscribers.

LocalBus.subscribe "user.created" do |message|
  raise "Something went wrong!"
  # never reached (business logic...)
end

LocalBus.subscribe "user.created" do |message|
  # This still executes even though the other subscriber has an error
  # business logic (e.g. API calls, database queries, disk operations, etc.)
end

# The publish operation completes with partial success
message = LocalBus.publish("user.created", user_id: 123)
errored_subscribers = message.subscribers.select(&:errored?)
#=> [#<LocalBus::Subscriber:0x000000011ebbcaf0 ...>]

errored_subscribers.first.error
#=> #<LocalBus::Subscriber::Error: Invocation failed! Something went wrong!>

Messages are held in memory until all subscribers have completed. Consider this when publishing large payloads or during high load scenarios.

# memory-efficient publishing of large datasets
large_dataset.each_slice(100) do |batch|
  message = LocalBus.publish("data.process", items: batch)
  message.wait # wait before processing more messages
end

LocalBus facilitates non-blocking I/O but bottlenecks can still be triggered by CPU-intensive operations.

LocalBus.subscribe "cpu.intensive" do |message|
  # CPU bound operation can trigger a bottleneck
end

The Station delays process exit in an attempt to flush the queue and avoid dropped messages. This delay can be configured via the :wait option in the constructor (default: 5).

• The Bus is single-threaded - long-running or CPU-bound subscribers can impact latency
• The Station may drop messages at process exit (messages are not persisted between process restarts)
• No distributed support - limited to single process (intra-process)
• Large message payloads may impact memory usage, especially under high load
• No built-in retry mechanism for failed subscribers (subscribers expose an error property, but you'll need to check and handle such errors)

Consider these limitations when designing your system architecture.

The project includes demo scripts that showcase concurrent processing capabilities:

bin/demo-bus     # demonstrates Bus performance
bin/demo-station # demonstrates Station performance

Both demos simulate I/O-bound operations (1 second latency per subscriber) to show how LocalBus handles concurrent processing.

For example, LocalBus can process 10 messages with 10 I/O-bound subscribers each in ~1 second instead of 100 seconds, on a 10-core system.

• Message Bus - A reliable and robust messaging bus for Ruby and Rack
• Wisper - A micro library providing Ruby objects with Publish-Subscribe capabilities