Class: TypedBus::MessageBus

Inherits:
Object
  • Object
show all
Defined in:
lib/typed_bus/message_bus.rb

Overview

Registry facade for named, typed pub/sub channels.

Examples:

bus = TypedBus::MessageBus.new
bus.add_channel(:events, type: MyEvent)

bus.subscribe(:events) do |delivery|
  puts delivery.message
  delivery.ack!
end

Async { bus.publish(:events, MyEvent.new(...)) }

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(timeout: :use_default, max_pending: :use_default, throttle: :use_default) ⇒ MessageBus

Returns a new instance of MessageBus.

Parameters:

  • timeout (Numeric, Symbol) (defaults to: :use_default) —

    delivery ACK deadline (overrides global default)

  • max_pending (Integer, nil, Symbol) (defaults to: :use_default) —

    backpressure limit (overrides global default)

  • throttle (Float, Symbol) (defaults to: :use_default) —

    default throttle threshold for all channels (overrides global default)



23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/typed_bus/message_bus.rb', line 23

def initialize(timeout: :use_default, max_pending: :use_default, throttle: :use_default)
  global  = TypedBus.configuration
  @config = global.dup
  resolved = global.resolve(timeout: timeout, max_pending: max_pending, throttle: throttle)
  @config.timeout     = resolved[:timeout]
  @config.max_pending = resolved[:max_pending]
  @config.throttle    = resolved[:throttle]

  @channels = {}
  @stats = Stats.new
  effective_throttle = @config.throttle.to_f
  log(:info, "message bus initialized (default_throttle=#{effective_throttle > 0 ? "#{(effective_throttle * 100).round}%" : 'off'})")
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



18
19
20
 
# File 'lib/typed_bus/message_bus.rb', line 18

def config
  @config
end

#stats ⇒ Object (readonly)

Returns the value of attribute stats.



18
19
20
 
# File 'lib/typed_bus/message_bus.rb', line 18

def stats
  @stats
end

Instance Method Details

#add_channel(name, type: nil, timeout: :use_default, max_pending: :use_default, throttle: :use_default) ⇒ Channel

Register a named channel.

When a parameter is not provided (left as :use_default), the bus-level config value is used. Pass an explicit value to override for this channel.

Parameters:

  • name (Symbol)
  • type (Class, nil) (defaults to: nil) —

    optional type constraint

  • timeout (Numeric, Symbol) (defaults to: :use_default) —

    delivery ACK deadline in seconds

  • max_pending (Integer, nil, Symbol) (defaults to: :use_default) —

    backpressure limit

  • throttle (Float, Symbol) (defaults to: :use_default) —

    capacity ratio where backoff begins (0.0 = disabled)

Returns:



48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/typed_bus/message_bus.rb', line 48

def add_channel(name, type: nil, timeout: :use_default, max_pending: :use_default, throttle: :use_default)
  resolved = @config.resolve(timeout: timeout, max_pending: max_pending, throttle: throttle)
  effective_timeout     = resolved[:timeout]
  effective_max_pending = resolved[:max_pending]
  effective_throttle    = resolved[:throttle]

  log(:info, "adding channel :#{name} (type=#{type || 'any'}, timeout=#{effective_timeout}s, max_pending=#{effective_max_pending || 'unbounded'})")
  channel = Channel.new(name, type: type, timeout: effective_timeout, max_pending: effective_max_pending, stats: @stats, throttle: effective_throttle)
  @channels[name] = channel
  channel
end

#channel?(name) ⇒ Boolean

Check if a channel exists.

Parameters:

  • name (Symbol)

Returns:

  • (Boolean) 


140
141
142
 
# File 'lib/typed_bus/message_bus.rb', line 140

def channel?(name)
  @channels.key?(name)
end

#channel_names ⇒ Array<Symbol>

List registered channel names.

Returns:

  • (Array<Symbol>) 


133
134
135
 
# File 'lib/typed_bus/message_bus.rb', line 133

def channel_names
  @channels.keys
end

#clear! ⇒ Object

Close all channels, clear DLQs, and reset stats.



145
146
147
148
149
 
# File 'lib/typed_bus/message_bus.rb', line 145

def clear!
  log(:info, "clearing all channels and resetting stats")
  @channels.each_value(&:clear!)
  @stats.reset!
end

#close(channel_name) ⇒ Object

Close a specific channel.

Parameters:

  • channel_name (Symbol) 


120
121
122
123
 
# File 'lib/typed_bus/message_bus.rb', line 120

def close(channel_name)
  channel = fetch_channel!(channel_name)
  channel.close
end

#close_all ⇒ Object

Close all channels (graceful shutdown).



126
127
128
129
 
# File 'lib/typed_bus/message_bus.rb', line 126

def close_all
  log(:info, "closing all channels (#{@channels.size} channels)")
  @channels.each_value(&:close)
end

#dead_letters(channel_name) ⇒ DeadLetterQueue

Access a channel's dead letter queue.

Parameters:

  • channel_name (Symbol)

Returns:



113
114
115
116
 
# File 'lib/typed_bus/message_bus.rb', line 113

def dead_letters(channel_name)
  channel = fetch_channel!(channel_name)
  channel.dead_letter_queue
end

#pending?(channel_name) ⇒ Boolean

Check if a channel has pending (unresolved) deliveries.

Parameters:

  • channel_name (Symbol)

Returns:

  • (Boolean) 


97
98
99
100
 
# File 'lib/typed_bus/message_bus.rb', line 97

def pending?(channel_name)
  channel = fetch_channel!(channel_name)
  channel.pending?
end

#pending_count(channel_name) ⇒ Integer

Number of unresolved deliveries on a channel.

Parameters:

  • channel_name (Symbol)

Returns:

  • (Integer) 


105
106
107
108
 
# File 'lib/typed_bus/message_bus.rb', line 105

def pending_count(channel_name)
  channel = fetch_channel!(channel_name)
  channel.pending_count
end

#publish(channel_name, message) ⇒ Object

Publish a message to a named channel.

Parameters:

  • channel_name (Symbol)
  • message (Object) 


71
72
73
74
75
 
# File 'lib/typed_bus/message_bus.rb', line 71

def publish(channel_name, message)
  channel = fetch_channel!(channel_name)
  @stats.increment(:"#{channel_name}_published")
  channel.publish(message)
end

#remove_channel(name) ⇒ Object

Close and remove a channel.

Parameters:

  • name (Symbol) 


62
63
64
65
66
 
# File 'lib/typed_bus/message_bus.rb', line 62

def remove_channel(name)
  log(:info, "removing channel :#{name}")
  channel = @channels.delete(name)
  channel&.close
end

#subscribe(channel_name, &block) ⇒ Integer

Subscribe to a named channel. The block receives a Delivery object.

Parameters:

  • channel_name (Symbol)

Returns:

  • (Integer) —

    subscriber id



81
82
83
84
 
# File 'lib/typed_bus/message_bus.rb', line 81

def subscribe(channel_name, &block)
  channel = fetch_channel!(channel_name)
  channel.subscribe(&block)
end

#unsubscribe(channel_name, id_or_block) ⇒ Object

Unsubscribe from a named channel.

Parameters:

  • channel_name (Symbol)
  • id_or_block (Integer, Proc) 


89
90
91
92
 
# File 'lib/typed_bus/message_bus.rb', line 89

def unsubscribe(channel_name, id_or_block)
  channel = fetch_channel!(channel_name)
  channel.unsubscribe(id_or_block)
end