Class: TypedBus::Channel

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

Overview

A named, typed pub/sub channel with explicit acknowledgment tracking.

Subscribers receive Delivery objects and must call ack! or nack!. Messages are considered delivered only when ALL subscribers ACK. NACKed or timed-out deliveries go to the dead letter queue.

Optionally bounded: when max_pending is set, publishers block until pending deliveries drain below the limit.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, type: nil, timeout: 30, max_pending: nil, stats: nil, throttle: 0.0) ⇒ Channel

Returns a new instance of Channel.

Parameters:

  • name (Symbol) —

    channel name

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

    optional type constraint for messages

  • timeout (Numeric) (defaults to: 30) —

    seconds before delivery auto-nacks (default: 30)

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

    backpressure limit (nil = unbounded)

  • stats (Stats, nil) (defaults to: nil) —

    optional stats tracker for counters

  • throttle (Float) (defaults to: 0.0) —

    capacity ratio (0.0..1.0) at which asymptotic backoff begins (0.0 = disabled)



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/typed_bus/channel.rb', line 21

def initialize(name, type: nil, timeout: 30, max_pending: nil, stats: nil, throttle: 0.0)
  @name              = name
  @type              = type
  @timeout           = timeout
  @max_pending       = max_pending
  @stats             = stats
  @throttle          = throttle.to_f
  @subscribers       = {}
  @next_subscriber_id = 0
  @pending_deliveries = {}
  @active_deliveries  = []
  @closed            = false
  @dead_letter_queue = DeadLetterQueue.new
  @backpressure      = Async::Condition.new

  validate_throttle_config!

  log(:info, "channel created (type=#{type || 'any'}, timeout=#{timeout}s, max_pending=#{max_pending || 'unbounded'}, throttle=#{@throttle > 0 ? "#{(@throttle * 100).round}%" : 'off'})")
end

Instance Attribute Details

#dead_letter_queue ⇒ Object (readonly)

Returns the value of attribute dead_letter_queue.



13
14
15
 
# File 'lib/typed_bus/channel.rb', line 13

def dead_letter_queue
  @dead_letter_queue
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



13
14
15
 
# File 'lib/typed_bus/channel.rb', line 13

def name
  @name
end

#type ⇒ Object (readonly)

Returns the value of attribute type.



13
14
15
 
# File 'lib/typed_bus/channel.rb', line 13

def type
  @type
end

Instance Method Details

#clear! ⇒ Object

Hard reset: cancel all timeout tasks, discard all pending state and DLQ.



172
173
174
175
176
177
178
179
 
# File 'lib/typed_bus/channel.rb', line 172

def clear!
  log(:info, "clearing all state (active=#{@active_deliveries.size}, pending=#{@pending_deliveries.size}, dlq=#{@dead_letter_queue.size})")
  @active_deliveries.each(&:cancel_timeout)
  @active_deliveries.clear
  @pending_deliveries.clear
  @dead_letter_queue.clear!
  @backpressure.signal
end

#close ⇒ Object

Close the channel. Stops accepting new publishes/subscribes. Pending deliveries are force-nacked (routed to DLQ with stats).



157
158
159
160
161
162
163
164
165
 
# File 'lib/typed_bus/channel.rb', line 157

def close
  return if @closed

  @closed = true
  pending = @active_deliveries.count(&:pending?)
  log(:info, "closing (force-nacking #{pending} pending deliveries)")
  force_nack_pending
  @backpressure.signal
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


167
168
169
 
# File 'lib/typed_bus/channel.rb', line 167

def closed?
  @closed
end

#pending? ⇒ Boolean

True when there are unresolved deliveries.

Returns:

  • (Boolean) 


151
152
153
 
# File 'lib/typed_bus/channel.rb', line 151

def pending?
  !@pending_deliveries.empty?
end

#pending_count ⇒ Object

Number of unresolved delivery trackers.



146
147
148
 
# File 'lib/typed_bus/channel.rb', line 146

def pending_count
  @pending_deliveries.size
end

#publish(message) ⇒ DeliveryTracker

Publish a message to all current subscribers.

Blocks the calling fiber if the channel is bounded and at capacity.

Parameters:

  • message (Object)

Returns:

Raises:

  • (ArgumentError) —

    if message doesn't match type constraint

  • (RuntimeError) —

    if channel is closed



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/typed_bus/channel.rb', line 49

def publish(message)
  raise "Channel :#{@name} is closed" if @closed

  validate_type!(message)
  apply_throttle! if throttled?
  wait_for_capacity if bounded?

  subscriber_ids = @subscribers.keys

  if subscriber_ids.empty?
    log(:warn, "published with no subscribers, routing to DLQ (message=#{message.class})")
    dead_letter_no_subscribers(message)
    return nil
  end

  log(:info, "published message (type=#{message.class}, subscribers=#{subscriber_ids.size})")

  tracker = DeliveryTracker.new(message, channel_name: @name, subscriber_ids: subscriber_ids)
  @pending_deliveries[tracker.object_id] = tracker

  tracker.on_complete { record_stat(:delivered) }
  tracker.on_resolved { tracker_resolved(tracker) }

  subscriber_ids.each do |sub_id|
    handler = @subscribers[sub_id]
    next unless handler

    delivery = Delivery.new(
      message,
      channel_name: @name,
      subscriber_id: sub_id,
      timeout: @timeout,
      on_ack: ->(_id) {
        @active_deliveries.delete(delivery)
        tracker.ack(sub_id)
      },
      on_nack: ->(_id) {
        @active_deliveries.delete(delivery)
        tracker.nack(sub_id)
        @dead_letter_queue.push(delivery)
        record_stat(:dead_lettered)
        if delivery.timed_out?
          record_stat(:timed_out)
        else
          record_stat(:nacked)
        end
      }
    )

    @active_deliveries << delivery
    log(:debug, "notifying subscriber ##{sub_id} (message=#{message.class})")

    Async do
      handler.call(delivery)
    rescue StandardError => e
      log(:error, "subscriber ##{sub_id} raised #{e.class}: #{e.message}")
      delivery.nack! if delivery.pending?
    end
  end

  tracker
end

#subscribe(&block) ⇒ Integer

Subscribe to messages on this channel. The block receives a Delivery object.

Returns:

  • (Integer) —

    subscriber id (use for unsubscribe)



116
117
118
119
120
121
122
123
 
# File 'lib/typed_bus/channel.rb', line 116

def subscribe(&block)
  raise "Channel :#{@name} is closed" if @closed

  id = next_subscriber_id
  @subscribers[id] = block
  log(:info, "subscriber ##{id} registered (total=#{@subscribers.size})")
  id
end

#subscriber_count ⇒ Object

Number of active subscribers.



141
142
143
 
# File 'lib/typed_bus/channel.rb', line 141

def subscriber_count
  @subscribers.size
end

#unsubscribe(id_or_block) ⇒ Object

Remove a subscriber by id or block reference.

Parameters:

  • id_or_block (Integer, Proc) 


128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/typed_bus/channel.rb', line 128

def unsubscribe(id_or_block)
  case id_or_block
  when Integer
    @subscribers.delete(id_or_block)
    log(:info, "subscriber ##{id_or_block} unsubscribed (remaining=#{@subscribers.size})")
  else
    before = @subscribers.size
    @subscribers.delete_if { |_, v| v == id_or_block }
    log(:info, "subscriber unsubscribed by reference (removed=#{before - @subscribers.size}, remaining=#{@subscribers.size})")
  end
end