Class: Mimi::Messaging::SQS_SNS::ReplyConsumer

Inherits:
Object
  • Object
show all
Defined in:
lib/mimi/messaging/sqs_sns/reply_consumer.rb

Overview

ReplyConsumer listens on a particular SQS queue for replies and passes them to registered Queues (see Ruby ::Queue class).

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(adapter, reply_queue_name) ⇒ ReplyConsumer

Returns a new instance of ReplyConsumer.



13
14
15
16
17
18
19
20
21
22
 
# File 'lib/mimi/messaging/sqs_sns/reply_consumer.rb', line 13

def initialize(adapter, reply_queue_name)
  @mutex = Mutex.new
  @queues = {}
  @adapter = adapter
  @reply_queue_name = reply_queue_name
  @consumer = TemporaryQueueConsumer.new(adapter, reply_queue_name) do |message|
    dispatch_message(message)
  end
  @reply_queue_url = @consumer.queue_url
end

Instance Attribute Details

#reply_queue_name ⇒ Object (readonly)

Returns the value of attribute reply_queue_name.



11
12
13
 
# File 'lib/mimi/messaging/sqs_sns/reply_consumer.rb', line 11

def reply_queue_name
  @reply_queue_name
end

#reply_queue_url ⇒ Object (readonly)

Returns the value of attribute reply_queue_url.



11
12
13
 
# File 'lib/mimi/messaging/sqs_sns/reply_consumer.rb', line 11

def reply_queue_url
  @reply_queue_url
end

Instance Method Details

#register_request_id(request_id) ⇒ Queue

Register a new request_id to listen for.

Whenever the message with given request_id will be received, it will be dispatched to a returned Queue.

Parameters:

  • request_id (String)

Returns:

  • (Queue) —

    a new Queue object registered for this request_id



38
39
40
41
42
43
44
 
# File 'lib/mimi/messaging/sqs_sns/reply_consumer.rb', line 38

def register_request_id(request_id)
  queue = TimeoutQueue.new
  @mutex.synchronize do
    queue = @queues[request_id] ||= queue
  end
  queue
end

#stop ⇒ Object 



24
25
26
27
28
 
# File 'lib/mimi/messaging/sqs_sns/reply_consumer.rb', line 24

def stop
  @consumer.stop
rescue StandardError => e
  raise Mimi::Messaging::Error, "Failed to stop reply consumer: #{e}"
end