Class: Hutch::Broker

Inherits:
Object
  • Object
show all
Includes:
Logging
Defined in:
lib/hutch/broker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Logging

#logger, logger, logger=, setup_logger

Constructor Details

#initialize(config = nil) ⇒ Broker

Returns a new instance of Broker.



13
14
15
 
# File 'lib/hutch/broker.rb', line 13

def initialize(config = nil)
  @config = config || Hutch::Config
end

Instance Attribute Details

#api_client ⇒ Object

Returns the value of attribute api_client.



11
12
13
 
# File 'lib/hutch/broker.rb', line 11

def api_client
  @api_client
end

#channel ⇒ Object

Returns the value of attribute channel.



11
12
13
 
# File 'lib/hutch/broker.rb', line 11

def channel
  @channel
end

#connection ⇒ Object

Returns the value of attribute connection.



11
12
13
 
# File 'lib/hutch/broker.rb', line 11

def connection
  @connection
end

#exchange ⇒ Object

Returns the value of attribute exchange.



11
12
13
 
# File 'lib/hutch/broker.rb', line 11

def exchange
  @exchange
end

Instance Method Details

#ack(delivery_tag) ⇒ Object 



187
188
189
 
# File 'lib/hutch/broker.rb', line 187

def ack(delivery_tag)
  @channel.ack(delivery_tag, false)
end

#bind_queue(queue, routing_keys) ⇒ Object

Bind a queue to the broker’s exchange on the routing keys provided. Any existing bindings on the queue that aren’t present in the array of routing keys will be unbound.



142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/hutch/broker.rb', line 142

def bind_queue(queue, routing_keys)
  if http_api_use_enabled?
    # Find the existing bindings, and unbind any redundant bindings
    queue_bindings = bindings.select { |dest, keys| dest == queue.name }
    queue_bindings.each do |dest, keys|
      keys.reject { |key| routing_keys.include?(key) }.each do |key|
        logger.debug "removing redundant binding #{queue.name} <--> #{key}"
        queue.unbind(@exchange, routing_key: key)
      end
    end
  end

  # Ensure all the desired bindings are present
  routing_keys.each do |routing_key|
    logger.debug "creating binding #{queue.name} <--> #{routing_key}"
    queue.bind(@exchange, routing_key: routing_key)
  end
end

#bindings ⇒ Object

Return a mapping of queue names to the routing keys they’re bound to.



128
129
130
131
132
133
134
135
136
137
 
# File 'lib/hutch/broker.rb', line 128

def bindings
  results = Hash.new { |hash, key| hash[key] = [] }
  @api_client.bindings.each do |binding|
    next if binding['destination'] == binding['routing_key']
    next unless binding['source'] == @config[:mq_exchange]
    next unless binding['vhost'] == @config[:mq_vhost]
    results[binding['destination']] << binding['routing_key']
  end
  results
end

#confirm_select(*args) ⇒ Object 



216
217
218
 
# File 'lib/hutch/broker.rb', line 216

def confirm_select(*args)
  @channel.confirm_select(*args)
end

#connect(options = {}) ⇒ Object 



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/hutch/broker.rb', line 17

def connect(options = {})
  @options = options
  set_up_amqp_connection
  if http_api_use_enabled?
    logger.info "HTTP API use is enabled"
    set_up_api_connection
  else
    logger.info "HTTP API use is disabled"
  end

  if tracing_enabled?
    logger.info "tracing is enabled using #{@config[:tracer]}"
  else
    logger.info "tracing is disabled"
  end

  if block_given?
    begin
      yield
    ensure
      disconnect
    end
  end
end

#disconnect ⇒ Object 



42
43
44
45
46
 
# File 'lib/hutch/broker.rb', line 42

def disconnect
  @channel.close    if @channel
  @connection.close if @connection
  @channel, @connection, @exchange, @api_client = nil, nil, nil, nil
end

#http_api_use_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


103
104
105
106
107
108
109
110
111
112
 
# File 'lib/hutch/broker.rb', line 103

def http_api_use_enabled?
  op = @options.fetch(:enable_http_api_use, true)
  cf = if @config[:enable_http_api_use].nil?
         true
       else
         @config[:enable_http_api_use]
       end

  op && cf
end

#nack(delivery_tag) ⇒ Object 



191
192
193
 
# File 'lib/hutch/broker.rb', line 191

def nack(delivery_tag)
  @channel.nack(delivery_tag, false, false)
end

#open_channel! ⇒ Object 



76
77
78
79
80
81
82
83
84
85
 
# File 'lib/hutch/broker.rb', line 76

def open_channel!
  logger.info "opening rabbitmq channel with pool size #{consumer_pool_size}"
  @channel = connection.create_channel(nil, consumer_pool_size).tap do |ch|
    ch.prefetch(@config[:channel_prefetch]) if @config[:channel_prefetch]
    if @config[:publisher_confirms] || @config[:force_publisher_confirms]
      logger.info 'enabling publisher confirms'
      ch.confirm_select
    end
  end
end

#open_connection! ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/hutch/broker.rb', line 63

def open_connection!
  logger.info "connecting to rabbitmq (#{sanitized_uri})"

  @connection = Bunny.new(connection_params)

  with_bunny_connection_handler(sanitized_uri) do
    @connection.start
  end

  logger.info "connected to RabbitMQ at #{connection_params[:host]} as #{connection_params[:username]}"
  @connection
end

#publish(routing_key, message, properties = {}) ⇒ Object 



195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/hutch/broker.rb', line 195

def publish(routing_key, message, properties = {})
  ensure_connection!(routing_key, message)

  non_overridable_properties = {
    routing_key: routing_key,
    timestamp: Time.now.to_i,
    content_type: 'application/json'
  }
  properties[:message_id] ||= generate_id

  json = JSON.dump(message)
  logger.info("publishing message '#{json}' to #{routing_key}")
  response = @exchange.publish(json, {persistent: true}.
    merge(properties).
    merge(global_properties).
    merge(non_overridable_properties))

  channel.wait_for_confirms if @config[:force_publisher_confirms]
  response
end

#queue(name, arguments = {}) ⇒ Object

Create / get a durable queue and apply namespace if it exists.



119
120
121
122
123
124
125
 
# File 'lib/hutch/broker.rb', line 119

def queue(name, arguments = {})
  with_bunny_precondition_handler('queue') do
    namespace = @config[:namespace].to_s.downcase.gsub(/[^-_:\.\w]/, "")
    name = name.prepend(namespace + ":") unless namespace.empty?
    channel.queue(name, durable: true, arguments: arguments)
  end
end

#reject(delivery_tag, requeue = false) ⇒ Object 



183
184
185
 
# File 'lib/hutch/broker.rb', line 183

def reject(delivery_tag, requeue=false)
  @channel.reject(delivery_tag, requeue)
end

#requeue(delivery_tag) ⇒ Object 



179
180
181
 
# File 'lib/hutch/broker.rb', line 179

def requeue(delivery_tag)
  @channel.reject(delivery_tag, true)
end

#set_up_amqp_connection ⇒ Object

Connect to RabbitMQ via AMQP. This sets up the main connection and channel we use for talking to RabbitMQ. It also ensures the existance of the exchange we’ll be using.



51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/hutch/broker.rb', line 51

def set_up_amqp_connection
  open_connection!
  open_channel!

  exchange_name = @config[:mq_exchange]
  logger.info "using topic exchange '#{exchange_name}'"

  with_bunny_precondition_handler('exchange') do
    @exchange = @channel.topic(exchange_name, durable: true)
  end
end

#set_up_api_connection ⇒ Object

Set up the connection to the RabbitMQ management API. Unfortunately, this is necessary to do a few things that are impossible over AMQP. E.g. listing queues and bindings.



90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/hutch/broker.rb', line 90

def set_up_api_connection
  logger.info "connecting to rabbitmq HTTP API (#{api_config.sanitized_uri})"

  with_authentication_error_handler do
    with_connection_error_handler do
      @api_client = CarrotTop.new(host: api_config.host, port: api_config.port,
                                  user: api_config.username, password: api_config.password,
                                  ssl: api_config.ssl)
      @api_client.exchanges
    end
  end
end

#stop ⇒ Object 



170
171
172
173
174
175
176
177
 
# File 'lib/hutch/broker.rb', line 170

def stop
  # Enqueue a failing job that kills the consumer loop
  channel_work_pool.shutdown
  # Give `timeout` seconds to jobs that are still being processed
  channel_work_pool.join(@config[:graceful_exit_timeout])
  # If after `timeout` they are still running, they are killed
  channel_work_pool.kill
end

#tracing_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


114
115
116
 
# File 'lib/hutch/broker.rb', line 114

def tracing_enabled?
  @config[:tracer] && @config[:tracer] != Hutch::Tracers::NullTracer
end

#using_publisher_confirmations? ⇒ Boolean

Returns:

  • (Boolean) 


224
225
226
 
# File 'lib/hutch/broker.rb', line 224

def using_publisher_confirmations?
  @channel.using_publisher_confirmations?
end

#wait_for_confirms ⇒ Object 



220
221
222
 
# File 'lib/hutch/broker.rb', line 220

def wait_for_confirms
  @channel.wait_for_confirms
end

#wait_on_threads(timeout) ⇒ Object

Each subscriber is run in a thread. This calls Thread#join on each of the subscriber threads.



163
164
165
166
167
168
 
# File 'lib/hutch/broker.rb', line 163

def wait_on_threads(timeout)
  # Thread#join returns nil when the timeout is hit. If any return nil,
  # the threads didn't all join so we return false.
  per_thread_timeout = timeout.to_f / work_pool_threads.length
  work_pool_threads.none? { |thread| thread.join(per_thread_timeout).nil? }
end