Class: Qurd::Listener

Inherits:
Object
  • Object
show all
Includes:
Mixins::AwsClients, Mixins::Configuration
Defined in:
lib/qurd/listener.rb

Overview

Provide an interface for interacting with configured queues and AWS.

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Mixins::Configuration

#qurd_config, #qurd_configuration, #qurd_logger, #qurd_logger!

Methods included from Mixins::AwsClients

#aws_client, #aws_retryable

Constructor Details

#initialize(attrs = {}) ⇒ Listener

Returns a new instance of Listener.

Parameters:

  • attrs (Hash) (defaults to: {})

Options Hash (attrs):

  • :aws_credentials (Aws::Credentials)
  • :name (String)
  • :queues (Array<String|Regexp>) —

    An array of SQS names and Regexps

  • :region (String)
  • :visibility_timeout (String)
  • :wait_time (String) 


44
45
46
47
48
49
50
51
52
 
# File 'lib/qurd/listener.rb', line 44

def initialize(attrs = {})
  @aws_credentials = attrs[:aws_credentials]
  @name = attrs[:name]
  @region = attrs[:region]
  @visibility_timeout = attrs[:visibility_timeout]
  @wait_time = attrs[:wait_time]
  @queues = convert_queues attrs[:queues]
  configure_queues
end

Instance Attribute Details

#aws_credentials ⇒ Aws::Credentials (readonly)

The AWS credentials for the account

Returns:

  • (Aws::Credentials) 


28
29
30
 
# File 'lib/qurd/listener.rb', line 28

def aws_credentials
  @aws_credentials
end

#message ⇒ Qurd::Message (readonly)

The message reveived

Returns:



28
29
30
31
32
33
34
 
# File 'lib/qurd/listener.rb', line 28

attr_reader :aws_credentials,
:message,
:name,
:queues,
:region,
:visibility_timeout,
:wait_time

#name ⇒ String (readonly)

The name of the executor

Returns:



28
29
30
31
32
33
34
 
# File 'lib/qurd/listener.rb', line 28

attr_reader :aws_credentials,
:message,
:name,
:queues,
:region,
:visibility_timeout,
:wait_time

#queues ⇒ Array<String> (readonly)

An array of AWS SQS URLs for the account

Returns:



28
29
30
31
32
33
34
 
# File 'lib/qurd/listener.rb', line 28

attr_reader :aws_credentials,
:message,
:name,
:queues,
:region,
:visibility_timeout,
:wait_time

#region ⇒ String (readonly)

The AWS region for the account

Returns:



28
29
30
31
32
33
34
 
# File 'lib/qurd/listener.rb', line 28

attr_reader :aws_credentials,
:message,
:name,
:queues,
:region,
:visibility_timeout,
:wait_time

#visibility_timeout ⇒ String (readonly)

Returns:



28
29
30
31
32
33
34
 
# File 'lib/qurd/listener.rb', line 28

attr_reader :aws_credentials,
:message,
:name,
:queues,
:region,
:visibility_timeout,
:wait_time

#wait_time ⇒ Object (readonly)

Returns the value of attribute wait_time.



28
29
30
31
32
33
34
 
# File 'lib/qurd/listener.rb', line 28

attr_reader :aws_credentials,
:message,
:name,
:queues,
:region,
:visibility_timeout,
:wait_time

Instance Method Details

#inspect ⇒ Object 



104
105
106
 
# File 'lib/qurd/listener.rb', line 104

def inspect
  format('<Qurd::Listener:%x name:%s>',  object_id, name)
end

#listen ⇒ Array<Thread>

Create one thread per queue, receive messages from it and process each message received

Returns:

  • (Array<Thread>) 


76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/qurd/listener.rb', line 76

def listen
  threads = queue_threads do |qurl, _context|
    loop do
      begin
        msgs = aws_client(:SQS).receive_message(
          queue_url: qurl,
          wait_time_seconds: wait_time,
          visibility_timeout: visibility_timeout
        )
        threads = process_messages(qurl, msgs)
        joins = threads.map do |thread|
          thread.join(qurd_configuration.listen_timeout)
        end
        if joins.compact.count != threads.count
          nthreads = threads.count - joins.compact.count
          lock_counter { @counter.thread_timeouts += nthreads }
          qurd_logger.warn('Some threads timed out')
        end
      rescue Aws::Errors::ServiceError => e
        lock_counter { @counter.aws_service_errors += 1 }
        qurd_logger.error("Aws raised #{e}")
      end
    end
  end
  threads << stats_thread
end

#queue_threads(&_block) {|url, ctx| ... } ⇒ Object

Create a thread for each queue URL, a context denoting the listener name and the queue URL.

Parameters:

  • _block (Proc) —

    the proc each thread should run

Yield Parameters:

  • url (String) —

    the url of the queue

  • ctx (Cabin::Context) —

    the logging context



59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/qurd/listener.rb', line 59

def queue_threads(&_block)
  queues.map do |qurl|
    @mutex = Mutex.new
    @counter = Hashie::Mash.new({thread_timeouts: 0, aws_service_errors: 0, successes: 0, failures: 0, messages: 0})

    qurd_logger.debug("Creating thread for #{qurl}")
    Thread.new(qurl) do |url|
      ctx = qurd_config.get_context(name: name, queue_name: url[/([^\/]+)$/])
      qurd_logger.debug('Thread running')
      yield url, ctx
    end
  end
end