Class: Ears::PublisherChannelPool

Inherits:
Object
  • Object
show all
Defined in:
lib/ears/publisher_channel_pool.rb

Overview

Channel pool management for publishers. Provides thread-safe channel pooling separate from consumer channels. Maintains two separate pools: one for standard publishing and one for confirmed publishing.

Class Method Summary collapse

Class Method Details

.reset! ⇒ void

This method returns an undefined value.

Resets both channel pools, forcing new channels to be created. This is useful for connection recovery scenarios.



30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/ears/publisher_channel_pool.rb', line 30

def reset!
  std_pool = @standard_pool
  cnf_pool = @confirms_pool
  @standard_pool = nil
  @confirms_pool = nil
  @creator_pid = nil

  std_pool&.shutdown(&:close)
  cnf_pool&.shutdown(&:close)
  nil
end

.reset_confirms_pool! ⇒ void

This method returns an undefined value.

Resets only the confirms channel pool, forcing new confirmed channels to be created. This is useful when confirmation failures occur and channels may have contaminated state.



46
47
48
49
50
51
52
 
# File 'lib/ears/publisher_channel_pool.rb', line 46

def reset_confirms_pool!
  cnf_pool = @confirms_pool
  @confirms_pool = nil

  cnf_pool&.shutdown(&:close)
  nil
end

.with_channel(confirms: false) {|channel| ... } ⇒ Object

Executes the given block with a channel from the appropriate pool.

Parameters:

  • confirms (Boolean) (defaults to: false) —

    Whether to use a channel with publisher confirms enabled

Yield Parameters:

  • channel (Bunny::Channel) —

    The channel to use for publishing

Returns:

  • (Object) —

    The result of the block



15
16
17
18
19
20
21
22
23
24
 
# File 'lib/ears/publisher_channel_pool.rb', line 15

def with_channel(confirms: false, &block)
  # CRITICAL: Preserve fork-safety at the entry point
  reset! if @creator_pid && @creator_pid != Process.pid

  pool = confirms ? confirms_pool : standard_pool
  pool.with do |channel|
    validate_channel_health!(channel, confirms)
    block.call(channel)
  end
end