Class: Bunny::TopologyRegistry

Inherits:
Object
  • Object
show all
Defined in:
lib/bunny/topology_registry.rb

Overview

As queues, exchanges, bindings are created and deleted, connections keep track of the topology using this registry.

Then, when the connection and its channels are recovered, this registry is used as the source of truth during topology recovery.

This registry takes care of dropping auto-delete exchanges or queues when their respective conditions for removal hold.

See Also:

Queues collapse

Consumers collapse

Exchanges collapse

Instance Attribute Summary collapse

Queues collapse

Consumers collapse

Exchanges collapse

Instance Method Summary collapse

Constructor Details

#initialize(opts = {}) ⇒ TopologyRegistry

Returns a new instance of TopologyRegistry.



34
35
36
37
38
 
# File 'lib/bunny/topology_registry.rb', line 34

def initialize(opts = {})
  @filter = opts.fetch(:topology_recovery_filter, DefaultTopologyRecoveryFilter.new)

  self.reset!
end

Instance Attribute Details

#consumers ⇒ Hash<String, Bunny::RecordedConsumer> (readonly)

Returns:



124
125
126
 
# File 'lib/bunny/topology_registry.rb', line 124

def consumers
  @consumers
end

#exchange_bindings ⇒ Set<Bunny::RecordedExchangeBinding> (readonly)

Returns:



250
251
252
 
# File 'lib/bunny/topology_registry.rb', line 250

def exchange_bindings
  @exchange_bindings
end

#exchanges ⇒ Hash<String, Bunny::RecordedExchange> (readonly)

Returns:



181
182
183
 
# File 'lib/bunny/topology_registry.rb', line 181

def exchanges
  @exchanges
end

#queue_bindings ⇒ Set<Bunny::RecordedQueueBinding> (readonly)

Returns:



241
242
243
 
# File 'lib/bunny/topology_registry.rb', line 241

def queue_bindings
  @queue_bindings
end

#queues ⇒ Hash<String, Bunny::RecordedQueue> (readonly)

Returns:



60
61
62
 
# File 'lib/bunny/topology_registry.rb', line 60

def queues
  @queues
end

Instance Method Details

#delete_recorded_consumer(consumer_tag) ⇒ Object

Parameters:

  • consumer_tag (String) 


153
154
155
156
157
158
159
 
# File 'lib/bunny/topology_registry.rb', line 153

def delete_recorded_consumer(consumer_tag)
  @consumer_mutex.synchronize do
    if (val = @consumers.delete(consumer_tag))
      self.maybe_delete_recorded_auto_delete_queue(val.queue_name)
    end
  end
end

#delete_recorded_exchange(exchange) ⇒ Object

Parameters:



211
212
213
 
# File 'lib/bunny/topology_registry.rb', line 211

def delete_recorded_exchange(exchange)
  self.delete_recorded_exchange_named(exchange.name)
end

#delete_recorded_exchange_binding(ch, source_name, destination_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • source_name (String)
  • destination_name (String)
  • routing_key (String)
  • arguments (Hash) 


308
309
310
311
312
313
314
315
316
317
318
319
320
 
# File 'lib/bunny/topology_registry.rb', line 308

def delete_recorded_exchange_binding(ch, source_name, destination_name, routing_key, arguments)
  b = RecordedExchangeBinding.new(ch)
    .with_source(source_name)
    .with_destination(destination_name)
    .with_routing_key(routing_key)
    .with_arguments(arguments)

  @binding_mutex.synchronize do
    if @exchange_bindings.delete?(b)
      self.maybe_delete_recorded_auto_delete_exchange(source_name)
    end
  end
end

#delete_recorded_exchange_named(name) ⇒ Object

Parameters:

  • name (String) 


216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/bunny/topology_registry.rb', line 216

def delete_recorded_exchange_named(name)
  @exchange_mutex.synchronize do
    @exchanges.delete(name)
    bs1 = self.remove_recorded_bindings_with_source(name)
    bs2 = self.remove_recorded_bindings_with_exchange_destination(name)

    bs1.each do |b|
      self.maybe_delete_recorded_auto_delete_exchange(b.source)
    end

    bs2.each do |b|
      self.maybe_delete_recorded_auto_delete_exchange(b.source)
    end
  end
end

#delete_recorded_queue(queue) ⇒ Object

Parameters:



92
93
94
 
# File 'lib/bunny/topology_registry.rb', line 92

def delete_recorded_queue(queue)
  self.delete_recorded_queue_named(queue.name)
end

#delete_recorded_queue_binding(ch, exchange_name, queue_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • exchange_name (String)
  • queue_name (String)
  • routing_key (String)
  • arguments (Hash) 


278
279
280
281
282
283
284
285
286
 
# File 'lib/bunny/topology_registry.rb', line 278

def delete_recorded_queue_binding(ch, exchange_name, queue_name, routing_key, arguments)
  b = RecordedQueueBinding.new(ch)
  .with_source(exchange_name)
  .with_destination(queue_name)
  .with_routing_key(routing_key)
  .with_arguments(arguments)

  @binding_mutex.synchronize { @queue_bindings.delete(b) }
end

#delete_recorded_queue_named(name) ⇒ Object

Parameters:

  • name (String) 


104
105
106
107
108
109
110
111
112
113
 
# File 'lib/bunny/topology_registry.rb', line 104

def delete_recorded_queue_named(name)
  @queue_mutex.synchronize do
    @queues.delete(name)

    bs = self.remove_recorded_bindings_with_queue_destination(name)
    bs.each do |b|
      self.maybe_delete_recorded_auto_delete_exchange(b.source)
    end
  end
end

#delete_recorded_queue_named_without_cascading(name) ⇒ Object

Parameters:

  • name (String) 


97
98
99
100
101
 
# File 'lib/bunny/topology_registry.rb', line 97

def delete_recorded_queue_named_without_cascading(name)
  @queue_mutex.synchronize do
    @queues.delete(name)
  end
end

#filtered_consumers ⇒ Array<Bunny::RecordedConsumer>

Returns:

See Also:



128
129
130
 
# File 'lib/bunny/topology_registry.rb', line 128

def filtered_consumers
  @filter.filter_consumers(@consumers.values)
end

#filtered_exchange_bindings ⇒ Array<Bunny::RecordedExchangeBinding>

Returns:

See Also:



254
255
256
 
# File 'lib/bunny/topology_registry.rb', line 254

def filtered_exchange_bindings
  @filter.filter_exchange_bindings(@exchange_bindings)
end

#filtered_exchanges ⇒ Array<Bunny::RecordedExchange>

Returns:

See Also:



185
186
187
 
# File 'lib/bunny/topology_registry.rb', line 185

def filtered_exchanges
  @filter.filter_exchanges(@exchanges.values)
end

#filtered_queue_bindings ⇒ Array<Bunny::RecordedQueueBinding>

Returns:

See Also:



245
246
247
 
# File 'lib/bunny/topology_registry.rb', line 245

def filtered_queue_bindings
  @filter.filter_queue_bindings(@queue_bindings)
end

#filtered_queues ⇒ Array<Bunny::RecordedQueue>

Returns:

See Also:



64
65
66
 
# File 'lib/bunny/topology_registry.rb', line 64

def filtered_queues
  @filter.filter_queues(@queues.values)
end

#has_more_consumers_on_queue?(consumers, name) ⇒ Boolean

Parameters:

Returns:

  • (Boolean) 


421
422
423
424
425
 
# File 'lib/bunny/topology_registry.rb', line 421

def has_more_consumers_on_queue?(consumers, name)
  return false if consumers.empty?

  consumers.any? { |val| val.queue_name == name }
end

#has_more_destinations_bound_to_exchange?(queue_bindings, exchange_bindings, name) ⇒ Boolean

Parameters:

Returns:

  • (Boolean) 


430
431
432
433
434
435
436
437
 
# File 'lib/bunny/topology_registry.rb', line 430

def has_more_destinations_bound_to_exchange?(queue_bindings, exchange_bindings, name)
  return false if queue_bindings.empty? && exchange_bindings.empty?

  condition_one = queue_bindings.any? { |val| val.source == name }
  condition_two = exchange_bindings.any? { |val| val.source == name }

  condition_one || condition_two
end

#maybe_delete_recorded_auto_delete_exchange(name) ⇒ Object

Parameters:

  • name (String) —

    Auto-delete exchange name



372
373
374
375
376
377
378
379
380
 
# File 'lib/bunny/topology_registry.rb', line 372

def maybe_delete_recorded_auto_delete_exchange(name)
  @exchange_mutex.synchronize do
    if (x = @exchanges[name]) && x.auto_delete
      unless self.has_more_destinations_bound_to_exchange?(@queue_bindings.dup, @exchange_bindings.dup, name)
        self.delete_recorded_exchange_named(name)
      end
    end
  end
end

#maybe_delete_recorded_auto_delete_queue(name) ⇒ Object

Parameters:

  • name (String) —

    Auto-delete queue name



361
362
363
364
365
366
367
368
369
 
# File 'lib/bunny/topology_registry.rb', line 361

def maybe_delete_recorded_auto_delete_queue(name)
  @queue_mutex.synchronize do
    if (q = @queues[name]) && q.auto_delete?
      unless self.has_more_consumers_on_queue?(@consumers.values, name)
        self.delete_recorded_queue(q)
      end
    end
  end
end

#propagate_queue_name_change_to_bindings(old_name, new_name) ⇒ Object

Parameters:

  • old_name (String)
  • new_name (String) 


346
347
348
349
350
351
352
 
# File 'lib/bunny/topology_registry.rb', line 346

def propagate_queue_name_change_to_bindings(old_name, new_name)
  @binding_mutex.synchronize do
    @queue_bindings.each do |rb|
      rb.update_destination_to(new_name) if rb.destination == old_name
    end
  end
end

#propagate_queue_name_change_to_consumers(old_name, new_name) ⇒ Object

Parameters:

  • old_name (String)
  • new_name (String) 


164
165
166
167
168
169
170
 
# File 'lib/bunny/topology_registry.rb', line 164

def propagate_queue_name_change_to_consumers(old_name, new_name)
  @consumer_mutex.synchronize do
    @consumers.each do |_, rc|
      rc.update_queue_name_to(new_name) if rc.queue_name == old_name
    end
  end
end

#record_consumer_with(ch, consumer_tag, queue_name, callable, manual_ack, exclusive, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • consumer_tag (String)
  • queue_name (String)
  • callable (#call)
  • manual_ack (Boolean)
  • exclusive (Boolean)
  • arguments (Hash) 


139
140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/bunny/topology_registry.rb', line 139

def record_consumer_with(ch, consumer_tag, queue_name, callable, manual_ack, exclusive, arguments)
  @consumer_mutex.synchronize do
    cons = RecordedConsumer.new(ch, queue_name)
      .with_consumer_tag(consumer_tag)
      .with_callable(callable)
      .with_manual_ack(manual_ack)
      .with_exclusive(exclusive)
      .with_arguments(arguments)

    @consumers[consumer_tag] = cons
  end
end

#record_exchange(exchange) ⇒ Object

Parameters:



190
191
192
 
# File 'lib/bunny/topology_registry.rb', line 190

def record_exchange(exchange)
  @exchange_mutex.synchronize { @exchanges[exchange.name] = RecordedExchange::from(exchange) }
end

#record_exchange_binding_with(ch, source_name, destination_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • source_name (String)
  • destination_name (String)
  • routing_key (String)
  • arguments (Hash) 


293
294
295
296
297
298
299
300
301
 
# File 'lib/bunny/topology_registry.rb', line 293

def record_exchange_binding_with(ch, source_name, destination_name, routing_key, arguments)
  b = RecordedExchangeBinding.new(ch)
    .with_source(source_name)
    .with_destination(destination_name)
    .with_routing_key(routing_key)
    .with_arguments(arguments)

  @binding_mutex.synchronize { @exchange_bindings.add(b) }
end

#record_exchange_with(ch, name, type, durable, auto_delete, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • name (String)
  • type (String)
  • durable (Boolean)
  • auto_delete (Boolean)
  • arguments (Hash) 


200
201
202
203
204
205
206
207
208
 
# File 'lib/bunny/topology_registry.rb', line 200

def record_exchange_with(ch, name, type, durable, auto_delete, arguments)
  exchange = RecordedExchange.new(ch, name)
    .with_type(type)
    .with_durable(durable)
    .with_auto_delete(auto_delete)
    .with_arguments(arguments)

    @exchange_mutex.synchronize { @exchanges[exchange.name] = exchange }
end

#record_queue(queue) ⇒ Object

Parameters:



69
70
71
 
# File 'lib/bunny/topology_registry.rb', line 69

def record_queue(queue)
  @queue_mutex.synchronize { @queues[queue.name] = RecordedQueue::from(queue) }
end

#record_queue_binding_with(ch, exchange_name, queue_name, routing_key, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • exchange_name (String)
  • queue_name (String)
  • routing_key (String)
  • arguments (Hash) 


263
264
265
266
267
268
269
270
271
 
# File 'lib/bunny/topology_registry.rb', line 263

def record_queue_binding_with(ch, exchange_name, queue_name, routing_key, arguments)
  b = RecordedQueueBinding.new(ch)
    .with_source(exchange_name)
    .with_destination(queue_name)
    .with_routing_key(routing_key)
    .with_arguments(arguments)

  @binding_mutex.synchronize { @queue_bindings.add(b) }
end

#record_queue_name_change(old_name, new_name) ⇒ Object

Parameters:

  • old_name (String)
  • new_name (String) 


325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
 
# File 'lib/bunny/topology_registry.rb', line 325

def record_queue_name_change(old_name, new_name)
  # update the recorded queue itself
  @queue_mutex.synchronize do
    if (orig = @queues[old_name])
      @queues.delete(old_name)

      orig.update_name_to(new_name)
      @queues[new_name] = orig.dup
    end
  end

  self.propagate_queue_name_change_to_bindings(old_name, new_name)
  self.propagate_queue_name_change_to_consumers(old_name, new_name)
  # Make sure the original name is removed and won't be recovered
  # but do not cascade
  self.delete_recorded_queue_named_without_cascading(old_name)
end

#record_queue_with(ch, name, server_named, durable, auto_delete, exclusive, arguments) ⇒ Object

Parameters:

  • ch (Bunny::Channel)
  • name (String)
  • server_named (Boolean)
  • durable (Boolean)
  • auto_delete (Boolean)
  • exclusive (Boolean)
  • arguments (Hash) 


80
81
82
83
84
85
86
87
88
89
 
# File 'lib/bunny/topology_registry.rb', line 80

def record_queue_with(ch, name, server_named, durable, auto_delete, exclusive, arguments)
  queue = RecordedQueue.new(ch, name)
    .with_server_named(server_named)
    .with_durable(durable)
    .with_auto_delete(auto_delete)
    .with_exclusive(exclusive)
    .with_arguments(arguments)

    @queue_mutex.synchronize { @queues[queue.name] = queue }
end

#remove_recorded_bindings_with_exchange_destination(name) ⇒ Array<Bunny::RecordedBinding>

Parameters:

  • name (String)

Returns:



411
412
413
414
415
416
417
 
# File 'lib/bunny/topology_registry.rb', line 411

def remove_recorded_bindings_with_exchange_destination(name)
  @binding_mutex.synchronize do
    matches = self.exchange_bindings.filter { |b| b.destination == name }
    @exchange_bindings = Set.new(@exchange_bindings.reject { |b| b.destination == name })
    matches
  end
end

#remove_recorded_bindings_with_queue_destination(name) ⇒ Array<Bunny::RecordedBinding>

Parameters:

  • name (String)

Returns:



401
402
403
404
405
406
407
 
# File 'lib/bunny/topology_registry.rb', line 401

def remove_recorded_bindings_with_queue_destination(name)
  @binding_mutex.synchronize do
    matches = self.queue_bindings.filter { |b| b.destination == name }
    @queue_bindings = Set.new(@queue_bindings.reject { |b| b.destination == name })
    matches
  end
end

#remove_recorded_bindings_with_source(name) ⇒ Array<Bunny::RecordedBinding>

Parameters:

  • name (String)

Returns:



384
385
386
387
388
389
390
391
392
393
394
395
396
397
 
# File 'lib/bunny/topology_registry.rb', line 384

def remove_recorded_bindings_with_source(name)
  @binding_mutex.synchronize do
    matching_qbs = self.queue_bindings.filter { |b| b.source == name }
    matching_xbs = self.exchange_bindings.filter { |b| b.source == name }

    matches = matching_qbs + matching_xbs
    matches.each do |b|
      @queue_bindings.delete(b)
      @exchange_bindings.delete(b)
    end

    matches
  end
end

#reset! ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/bunny/topology_registry.rb', line 40

def reset!
  @queues = {}
  @exchanges = {}
  @queue_bindings = Set.new
  @exchange_bindings = Set.new
  @consumers = {}

  @queue_mutex = Monitor.new
  @exchange_mutex = Monitor.new
  @binding_mutex = Monitor.new
  @consumer_mutex = Monitor.new
end