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.
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
176
177
178
|
# File 'lib/hutch/broker.rb', line 176
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.
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
# File 'lib/hutch/broker.rb', line 138
def bind_queue(queue, routing_keys)
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
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.
124
125
126
127
128
129
130
131
132
133
|
# File 'lib/hutch/broker.rb', line 124
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
205
206
207
|
# File 'lib/hutch/broker.rb', line 205
def confirm_select(*args)
@channel.confirm_select(*args)
end
|
#connect(options = {}) ⇒ Object
17
18
19
20
21
22
23
24
25
26
27
28
|
# File 'lib/hutch/broker.rb', line 17
def connect(options = {})
set_up_amqp_connection
set_up_api_connection if options.fetch(:enable_http_api_use, true)
if block_given?
begin
yield
ensure
disconnect
end
end
end
|
#disconnect ⇒ Object
30
31
32
33
34
|
# File 'lib/hutch/broker.rb', line 30
def disconnect
@channel.close if @channel
@connection.close if @connection
@channel, @connection, @exchange, @api_client = nil, nil, nil, nil
end
|
#nack(delivery_tag) ⇒ Object
180
181
182
|
# File 'lib/hutch/broker.rb', line 180
def nack(delivery_tag)
@channel.nack(delivery_tag, false, false)
end
|
#open_channel! ⇒ Object
87
88
89
90
91
92
93
94
95
96
|
# File 'lib/hutch/broker.rb', line 87
def open_channel!
logger.info 'opening rabbitmq channel'
@channel = connection.create_channel.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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
|
# File 'lib/hutch/broker.rb', line 51
def open_connection!
if @config[:uri] && !@config[:uri].empty?
u = URI.parse(@config[:uri])
@config[:mq_host] = u.host
@config[:mq_port] = u.port
@config[:mq_vhost] = u.path.sub(/^\//, "")
@config[:mq_username] = u.user
@config[:mq_password] = u.password
end
host = @config[:mq_host]
port = @config[:mq_port]
vhost = @config[:mq_vhost]
username = @config[:mq_username]
password = @config[:mq_password]
tls = @config[:mq_tls]
tls_key = @config[:mq_tls_key]
tls_cert = @config[:mq_tls_cert]
protocol = tls ? "amqps://" : "amqp://"
sanitized_uri = "#{protocol}#{username}@#{host}:#{port}/#{vhost.sub(/^\//, '')}"
logger.info "connecting to rabbitmq (#{sanitized_uri})"
@connection = Bunny.new(host: host, port: port, vhost: vhost,
tls: tls, tls_key: tls_key, tls_cert: tls_cert,
username: username, password: password,
heartbeat: 30, automatically_recover: true,
network_recovery_interval: 1)
with_bunny_connection_handler(sanitized_uri) do
@connection.start
end
logger.info "connected to RabbitMQ at #{host} as #{username}"
@connection
end
|
#publish(routing_key, message, properties = {}) ⇒ Object
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
# File 'lib/hutch/broker.rb', line 184
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
logger.info("publishing message '#{message.inspect}' to #{routing_key}")
response = @exchange.publish(JSON.dump(message), {persistent: true}.
merge(properties).
merge(global_properties).
merge(non_overridable_properties))
channel.wait_for_confirms if @config[:force_publisher_confirms]
response
end
|
#queue(name) ⇒ Object
Create / get a durable queue and apply namespace if it exists.
115
116
117
118
119
120
121
|
# File 'lib/hutch/broker.rb', line 115
def queue(name)
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)
end
end
|
#reject(delivery_tag, requeue = false) ⇒ Object
172
173
174
|
# File 'lib/hutch/broker.rb', line 172
def reject(delivery_tag, requeue=false)
@channel.reject(delivery_tag, requeue)
end
|
#requeue(delivery_tag) ⇒ Object
168
169
170
|
# File 'lib/hutch/broker.rb', line 168
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.
39
40
41
42
43
44
45
46
47
48
49
|
# File 'lib/hutch/broker.rb', line 39
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.
101
102
103
104
105
106
107
108
109
110
111
112
|
# File 'lib/hutch/broker.rb', line 101
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
164
165
166
|
# File 'lib/hutch/broker.rb', line 164
def stop
@channel.work_pool.kill
end
|
#using_publisher_confirmations? ⇒ Boolean
213
214
215
|
# File 'lib/hutch/broker.rb', line 213
def using_publisher_confirmations?
@channel.using_publisher_confirmations?
end
|
#wait_for_confirms ⇒ Object
209
210
211
|
# File 'lib/hutch/broker.rb', line 209
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.
157
158
159
160
161
162
|
# File 'lib/hutch/broker.rb', line 157
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
|