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
26
27
 
# File 'lib/flu-rails/event_publisher.rb', line 20

def initialize(configuration)
  @logger          = configuration.logger
  @configuration   = configuration
  @mutex           = Mutex.new
  @next_attempt_at = 0
  @exchanges       = {}
  @exchanges_mutex = Mutex.new
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.



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

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

Not connected while Bunny is reopening the channels either, although the connection is 'open?' from its handshake on: 'exchange' refuses to publish in that window, and 'PendingPublications' asks here before it retries. A publisher that said it was connected and then refused cost the event one of its attempts on every drain -- the one right after the failing commit, then the next commit or the end of the job -- and a thread committing twice within the few milliseconds Bunny takes to reopen the channels lost the event for a committed transaction. Saying it is not connected makes the event wait for Bunny to be done, as it does while the connection is down.

Returns:

  • (Boolean) 


63
64
65
 
# File 'lib/flu-rails/event_publisher.rb', line 63

def connected?
  !forked? && !@connection.nil? && @connection.open? && !being_reopened?
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.



72
73
74
75
76
77
78
79
 
# File 'lib/flu-rails/event_publisher.rb', line 72

def disconnect
  @mutex.synchronize do
    @connection.close if connected?
    @connection = nil
    @pid        = nil
    forget_exchanges
  end
end

#publish(event, persistent = true) ⇒ Object 



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

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