Class: Flu::EventPublisher

Inherits:
Object
  • Object
show all
Defined in:
lib/flu-rails/event_publisher.rb

Direct Known Subclasses

Dummy::InMemoryEventPublisher

Constant Summary collapse

NOT_CONNECTED_MESSAGE = 
"no connection to RabbitMQ: 'connect' was never called, or " \
"'disconnect' was. The railtie calls it at boot unless " \
"'auto_connect_to_exchange' is false."

Instance Method Summary collapse

Constructor Details

#initialize(configuration) ⇒ EventPublisher

Returns a new instance of EventPublisher.



14
15
16
17
18
 
# File 'lib/flu-rails/event_publisher.rb', line 14

def initialize(configuration)
  @logger        = configuration.logger
  @configuration = configuration
  @exchange_key  = :"flu_exchange_#{object_id}" # Channels are cached per thread
end

Instance Method Details

#connect ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/flu-rails/event_publisher.rb', line 27

def connect
  unless connected?
    connected = false
    while !connected
      begin
        connect_to_exchange
        connected = true
      rescue Bunny::TCPConnectionFailedForAllHosts
        @logger.warn("RabbitMQ connection failed, try again in 1 second.")
        sleep 1
      end
    end
  end
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


42
43
44
 
# File 'lib/flu-rails/event_publisher.rb', line 42

def connected?
  !@connection.nil? && @connection.open?
end

#disconnect ⇒ Object

Closing the connection closes every channel opened on it, and stops the heartbeat and recovery threads Bunny runs alongside it. The guard is on the connection alone: a connection that was opened before the exchange could be declared still has to be closed.



50
51
52
53
54
55
56
 
# File 'lib/flu-rails/event_publisher.rb', line 50

def disconnect
  if !@connection.nil? && @connection.open?
    @connection.close
  end
  @connection = nil
  Thread.current[@exchange_key] = nil
end

#publish(event, persistent = true) ⇒ Object 



20
21
22
23
24
25
 
# File 'lib/flu-rails/event_publisher.rb', line 20

def publish(event, persistent=true)
  routing_key = event.to_routing_key
  @logger.debug { "Publishing event with id '#{event.id}' with routing key: #{routing_key}" }
  exchange.publish(event.to_json, routing_key: routing_key, persistent: persistent)
  @logger.debug { "Event published." }
end