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."
CONNECTION_LOST_MESSAGE = 
"the connection to RabbitMQ is down. Bunny reopens it in the " \
"background when 'automatically_recover' is on, and publishing " \
"works again once it has."
CONNECTION_FAILED_MESSAGE = 
"could not reach RabbitMQ within %s seconds. Publishing reopens " \
"the connection itself once the broker answers again."
RECONNECTION_INTERVAL = 
5

Instance Method Summary collapse

Constructor Details

#initialize(configuration) ⇒ EventPublisher

Returns a new instance of EventPublisher.



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

def initialize(configuration)
  @logger          = configuration.logger
  @configuration   = configuration
  @mutex           = Mutex.new
  @next_attempt_at = 0
end

Instance Method Details

#connect ⇒ Object

Retries a broker that is not there yet, for at most 'max_connect_wait' seconds. Waiting on it forever would hold whatever called it -- the railtie calls it from 'to_prepare', which runs on every code reload, holding the reload interlock and the request that triggered it.



39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/flu-rails/event_publisher.rb', line 39

def connect
  @mutex.synchronize do
    next if connected?
    give_up_at = deadline
    begin
      connect_to_exchange
    rescue Bunny::TCPConnectionFailedForAllHosts
      raise ConnectionLostError, format(CONNECTION_FAILED_MESSAGE, @configuration.max_connect_wait) if expired?(give_up_at)
      @logger.warn("RabbitMQ connection failed, try again in 1 second.")
      sleep 1
      retry
    end
  end
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


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

def connected?
  !forked? && !@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. An inherited connection is dropped rather than closed: its socket is the parent's.



63
64
65
66
67
68
69
70
 
# File 'lib/flu-rails/event_publisher.rb', line 63

def disconnect
  @mutex.synchronize do
    @connection.close if connected?
    @connection = nil
    @pid        = nil
    Thread.current[exchange_key] = nil
  end
end

#publish(event, persistent = true) ⇒ Object 



27
28
29
30
31
32
33
34
 
# File 'lib/flu-rails/event_publisher.rb', line 27

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." }
rescue Bunny::ConnectionClosedError
  raise ConnectionLostError, CONNECTION_LOST_MESSAGE
end