Class: Surfliner::Mq::Router

Inherits:
Object
  • Object
show all
Defined in:
lib/surfliner/mq/router.rb

Overview

Queue subscriber that routes messages by routing key to multiple handlers

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic, queue_config: QueueConfig.from_env) ⇒ Router

Initializes a new Router and subscribes it to the specified queue on the specified topic.

Parameters:

  • topic (Mq::Topic) —

    The topic to listen on

  • queue_config (Mq::QueueConfig) (defaults to: QueueConfig.from_env) —

    The queue to subscribe to



19
20
21
22
23
 
# File 'lib/surfliner/mq/router.rb', line 19

def initialize(topic, queue_config: QueueConfig.from_env)
  @topic = topic
  @queue = topic.queue(queue_config)
  @consumer = queue.subscribe(&method(:on_delivery))
end

Instance Attribute Details

#consumer ⇒ Bunny::Consumer (readonly)

Returns the consumer (used to unsubscribe).

Returns:

  • (Bunny::Consumer) —

    the consumer (used to unsubscribe)



12
13
14
 
# File 'lib/surfliner/mq/router.rb', line 12

def consumer
  @consumer
end

#queue ⇒ Bunny::Queue (readonly)

Returns the queue.

Returns:

  • (Bunny::Queue) —

    the queue



9
10
11
 
# File 'lib/surfliner/mq/router.rb', line 9

def queue
  @queue
end

#topic ⇒ Topic (readonly)

Returns the topic.

Returns:



6
7
8
 
# File 'lib/surfliner/mq/router.rb', line 6

def topic
  @topic
end

Instance Method Details

#add_handler(routing_key) {|payload_json| ... } ⇒ Object

Adds a handler for the specified routing key. Handlers will receive only messages with the specified routing key. Multiple handler blocks can be added to the same queue and/or routing key.

Parameters:

  • routing_key (String) —

    the routing key to filter messages by

Yield Parameters:

  • payload_json (String) —

    each payload



31
32
33
34
35
36
37
38
 
# File 'lib/surfliner/mq/router.rb', line 31

def add_handler(routing_key, &block)
  raise "Can't add handlers after router shutdown" if consumer.nil?

  log("adding handler #{block}")
  handlers = handlers_for(routing_key)
  queue.bind(topic.exchange, routing_key:) if handlers.empty?
  handlers << block
end

#shutdown ⇒ Object

Removes all handlers and unsubscribes from the queue.



41
42
43
44
45
46
 
# File 'lib/surfliner/mq/router.rb', line 41

def shutdown
  log("shutting down")
  handlers_by_routing_key.clear
  consumer.cancel
  @consumer = nil
end