Class: Flu::EventPublisher

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

Direct Known Subclasses

Dummy::InMemoryEventPublisher

Instance Method Summary collapse

Constructor Details

#initialize(configuration) ⇒ EventPublisher

Returns a new instance of EventPublisher.



7
8
9
10
 
# File 'lib/flu-rails/event_publisher.rb', line 7

def initialize(configuration)
  @logger        = configuration.logger
  @configuration = configuration
end

Instance Method Details

#connect ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/flu-rails/event_publisher.rb', line 19

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) 


34
35
36
37
38
39
40
 
# File 'lib/flu-rails/event_publisher.rb', line 34

def connected?
  if @connection.nil? || @exchange.nil?
    false
  else
    @connection.open?
  end
end

#disconnect ⇒ Object

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



46
47
48
49
50
51
52
53
 
# File 'lib/flu-rails/event_publisher.rb', line 46

def disconnect
  if !@connection.nil? && @connection.open?
    @connection.close 
  end
  @connection = nil
  @channel    = nil
  @exchange   = nil
end

#publish(event, persistent = true) ⇒ Object 



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

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