Class: Hutch::Broker
- Inherits:
-
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.
12
13
14
|
# File 'lib/hutch/broker.rb', line 12
def initialize(config = nil)
@config = config || Hutch::Config
end
|
Instance Attribute Details
#api_client ⇒ Object
Returns the value of attribute api_client.
10
11
12
|
# File 'lib/hutch/broker.rb', line 10
def api_client
@api_client
end
|
#channel ⇒ Object
Returns the value of attribute channel.
10
11
12
|
# File 'lib/hutch/broker.rb', line 10
def channel
@channel
end
|
#connection ⇒ Object
Returns the value of attribute connection.
10
11
12
|
# File 'lib/hutch/broker.rb', line 10
def connection
@connection
end
|
#exchange ⇒ Object
Returns the value of attribute exchange.
10
11
12
|
# File 'lib/hutch/broker.rb', line 10
def exchange
@exchange
end
|
Instance Method Details
#ack(delivery_tag) ⇒ Object
190
191
192
|
# File 'lib/hutch/broker.rb', line 190
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.
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
# File 'lib/hutch/broker.rb', line 141
def bind_queue(queue, routing_keys)
if http_api_use_enabled?
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
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.
127
128
129
130
131
132
133
134
135
136
|
# File 'lib/hutch/broker.rb', line 127
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
230
231
232
|
# File 'lib/hutch/broker.rb', line 230
def confirm_select(*args)
@channel.confirm_select(*args)
end
|
#connect(options = {}) ⇒ Object
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
# File 'lib/hutch/broker.rb', line 16
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
41
42
43
44
45
|
# File 'lib/hutch/broker.rb', line 41
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
102
103
104
105
106
107
108
109
110
111
|
# File 'lib/hutch/broker.rb', line 102
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
194
195
196
|
# File 'lib/hutch/broker.rb', line 194
def nack(delivery_tag)
@channel.nack(delivery_tag, false, false)
end
|
#open_channel! ⇒ Object
75
76
77
78
79
80
81
82
83
84
|
# File 'lib/hutch/broker.rb', line 75
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|
@connection.prefetch_channel(ch, @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
62
63
64
65
66
67
68
69
70
71
72
73
|
# File 'lib/hutch/broker.rb', line 62
def open_connection!
logger.info "connecting to rabbitmq (#{sanitized_uri})"
@connection = Hutch::Adapter.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 = {}, options = {}) ⇒ Object
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
|
# File 'lib/hutch/broker.rb', line 198
def publish(routing_key, message, properties = {}, options = {})
ensure_connection!(routing_key, message)
serializer = options[:serializer] || @config[:serializer]
non_overridable_properties = {
routing_key: routing_key,
timestamp: @connection.current_timestamp,
content_type: serializer.content_type,
}
properties[:message_id] ||= generate_id
payload = serializer.encode(message)
logger.info {
spec =
if serializer.binary?
"#{payload.bytesize} bytes message"
else
"message '#{payload}'"
end
"publishing #{spec} to #{routing_key}"
}
response = @exchange.publish(payload, {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.
118
119
120
121
122
123
124
|
# File 'lib/hutch/broker.rb', line 118
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
186
187
188
|
# File 'lib/hutch/broker.rb', line 186
def reject(delivery_tag, requeue=false)
@channel.reject(delivery_tag, requeue)
end
|
#requeue(delivery_tag) ⇒ Object
182
183
184
|
# File 'lib/hutch/broker.rb', line 182
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.
50
51
52
53
54
55
56
57
58
59
60
|
# File 'lib/hutch/broker.rb', line 50
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.
89
90
91
92
93
94
95
96
97
98
99
100
|
# File 'lib/hutch/broker.rb', line 89
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
169
170
171
172
173
174
175
176
177
178
179
180
|
# File 'lib/hutch/broker.rb', line 169
def stop
if defined?(JRUBY_VERSION)
channel.close
else
channel_work_pool.shutdown
channel_work_pool.join(@config[:graceful_exit_timeout])
channel_work_pool.kill
end
end
|
#tracing_enabled? ⇒ Boolean
113
114
115
|
# File 'lib/hutch/broker.rb', line 113
def tracing_enabled?
@config[:tracer] && @config[:tracer] != Hutch::Tracers::NullTracer
end
|
#using_publisher_confirmations? ⇒ Boolean
238
239
240
|
# File 'lib/hutch/broker.rb', line 238
def using_publisher_confirmations?
@channel.using_publisher_confirmations?
end
|
#wait_for_confirms ⇒ Object
234
235
236
|
# File 'lib/hutch/broker.rb', line 234
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.
162
163
164
165
166
167
|
# File 'lib/hutch/broker.rb', line 162
def wait_on_threads(timeout)
per_thread_timeout = timeout.to_f / work_pool_threads.length
work_pool_threads.none? { |thread| thread.join(per_thread_timeout).nil? }
end
|