Class: BunnyPublisher::Base

Inherits:
Object
  • Object
show all
Includes:
Callbacks
Defined in:
lib/bunny_publisher/base.rb

Overview

Based on Publisher of Sneakers github.com/jondot/sneakers/blob/ed620b642b447701be490666ee284cf7d60ccf22/lib/sneakers/publisher.rb

Constant Summary collapse

RETRIABLE_ERRORS =

A list of errors that can be fixed by a connection recovery

[
  Bunny::ConnectionClosedError,
  Bunny::NetworkFailure,
  Bunny::ConnectionLevelException,
  Timeout::Error # can be raised by Bunny::Channel#with_continuation_timeout
].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(publish_connection: nil, connection: nil, exchange: nil, exchange_options: {}, **options) ⇒ Base

Returns a new instance of Base.



19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/bunny_publisher/base.rb', line 19

def initialize(publish_connection: nil, connection: nil, exchange: nil, exchange_options: {}, **options)
  @mutex = Mutex.new

  @exchange_name = exchange
  @exchange_options = exchange_options
  @options = options

  # Arguments are compatible with Sneakers::CONFIG and if connection given publisher will use it.
  # But using of same connection for publishing & consumers could cause problems.
  # https://www.cloudamqp.com/blog/2017-12-29-part1-rabbitmq-best-practice.html#separate-connections-for-publisher-and-consumer
  # Therefore, publish_connection allows to explicitly make publishers use different connection
  @connection = publish_connection || connection
end

Instance Attribute Details

#channel ⇒ Object (readonly)

Returns the value of attribute channel.



17
18
19
 
# File 'lib/bunny_publisher/base.rb', line 17

def channel
  @channel
end

#connection ⇒ Object (readonly)

Returns the value of attribute connection.



17
18
19
 
# File 'lib/bunny_publisher/base.rb', line 17

def connection
  @connection
end

#exchange ⇒ Object (readonly)

Returns the value of attribute exchange.



17
18
19
 
# File 'lib/bunny_publisher/base.rb', line 17

def exchange
  @exchange
end

Instance Method Details

#close ⇒ Object Also known as: stop 



49
50
51
 
# File 'lib/bunny_publisher/base.rb', line 49

def close
  connection&.close
end

#publish(message, message_options = {}) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/bunny_publisher/base.rb', line 33

def publish(message, message_options = {})
  @mutex.synchronize do
    @message = message
    @message_options = message_options

    run_callbacks(:publish) do
      with_errors_handling do
        ensure_connection!
        exchange.publish(message, message_options.dup) # Bunny modifies message options
      end
    end
  ensure
    @message = @message_options = nil
  end
end