Class: EventQ::Amazon::QueueClient
- Inherits:
-
Object
- Object
- EventQ::Amazon::QueueClient
- Defined in:
- lib/eventq_aws/aws_queue_client.rb
Instance Method Summary collapse
- #aws_safe_name(name) ⇒ Object
- #create_topic_arn(event_type) ⇒ Object
- #get_queue_arn(queue) ⇒ Object
-
#get_queue_url(queue) ⇒ Object
Returns the URL of the queue.
- #get_topic_arn(event_type) ⇒ Object
-
#initialize(options = {}) ⇒ QueueClient
constructor
A new instance of QueueClient.
- #keep_alive(connections: 15, interval: 1.5) ⇒ Object
-
#sns ⇒ Object
Returns the AWS SNS Client.
-
#sqs ⇒ Object
Returns the AWS SQS Client.
Constructor Details
#initialize(options = {}) ⇒ QueueClient
Returns a new instance of QueueClient.
5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 5 def initialize( = {}) if .has_key?(:aws_key) Aws.config[:credentials] = Aws::Credentials.new([:aws_key], [:aws_secret]) end if !.has_key?(:aws_account_number) raise ':aws_account_number option must be specified.'.freeze end @aws_account = [:aws_account_number] @sns_keep_alive_timeout = [:sns_keep_alive_timeout] || 30 @sns_continue_timeout = [:sns_continue_timeout] || 15 if .has_key?(:aws_region) @aws_region = [:aws_region] Aws.config[:region] = @aws_region else @aws_region = Aws.config[:region] end end |
Instance Method Details
#aws_safe_name(name) ⇒ Object
68 69 70 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 68 def aws_safe_name(name) return name[0..79].gsub(/[^a-zA-Z\d_\-]/,'') end |
#create_topic_arn(event_type) ⇒ Object
50 51 52 53 54 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 50 def create_topic_arn(event_type) _event_type = EventQ.create_event_type(event_type) response = sns.create_topic(name: aws_safe_name(_event_type)) return response.topic_arn end |
#get_queue_arn(queue) ⇒ Object
45 46 47 48 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 45 def get_queue_arn(queue) _queue_name = EventQ.create_queue_name(queue.name) return "arn:aws:sqs:#{@aws_region}:#{@aws_account}:#{aws_safe_name(_queue_name)}" end |
#get_queue_url(queue) ⇒ Object
Returns the URL of the queue. The queue will be created when it does
59 60 61 62 63 64 65 66 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 59 def get_queue_url(queue) _queue_name = EventQ.create_queue_name(queue.name) response= sqs.get_queue_url( queue_name: aws_safe_name(_queue_name), queue_owner_aws_account_id: @aws_account, ) return response.queue_url end |
#get_topic_arn(event_type) ⇒ Object
40 41 42 43 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 40 def get_topic_arn(event_type) _event_type = EventQ.create_event_type(event_type) return "arn:aws:sns:#{@aws_region}:#{@aws_account}:#{aws_safe_name(_event_type)}" end |
#keep_alive(connections: 15, interval: 1.5) ⇒ Object
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 72 def keep_alive(connections: 15, interval: 1.5) connections.times do Thread.new do while true do begin Seahorse::Client::NetHttp::ConnectionPool.pools.each do |cp| pool = cp.instance_variable_get(:@pool) pool.each do |k,v| cp.session_for(URI(k)) do |session| session.request(Net::HTTP::Get.new('/')) end end end rescue # swallow error and do nothing end sleep interval end end end end |
#sns ⇒ Object
Returns the AWS SNS Client
33 34 35 36 37 38 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 33 def sns @sns ||= Aws::SNS::Client.new( http_idle_timeout: @sns_keep_alive_timeout, http_continue_timeout: @sns_continue_timeout ) end |
#sqs ⇒ Object
Returns the AWS SQS Client
28 29 30 |
# File 'lib/eventq_aws/aws_queue_client.rb', line 28 def sqs @sqs ||= Aws::SQS::Client.new end |