Class: RightAws::SqsInterface
- Inherits:
-
RightAwsBase
- Object
- RightAwsBase
- RightAws::SqsInterface
- Includes:
- RightAwsBaseInterface
- Defined in:
- lib/sqs/right_sqs_interface.rb
Defined Under Namespace
Classes: SqsCreateQueueParser, SqsGetQueueAttributesParser, SqsGetVisibilityTimeoutParser, SqsListGrantsParser, SqsListQueuesParser, SqsReceiveMessagesParser, SqsSendMessagesParser, SqsStatusParser
Constant Summary collapse
- API_VERSION =
"2007-05-01"
- DEFAULT_HOST =
"queue.amazonaws.com"
- DEFAULT_PORT =
443
- DEFAULT_PROTOCOL =
'https'
- REQUEST_TTL =
30
- DEFAULT_VISIBILITY_TIMEOUT =
30
- @@bench =
AwsBenchmarkingBlock.new
- @@api =
API_VERSION
Constants included from RightAwsBaseInterface
RightAwsBaseInterface::DEFAULT_SIGNATURE_VERSION
Constants inherited from RightAwsBase
Instance Attribute Summary
Attributes included from RightAwsBaseInterface
#aws_access_key_id, #cache, #connection, #last_errors, #last_request, #last_request_id, #last_response, #logger, #params, #signature_version
Class Method Summary collapse
- .api ⇒ Object
- .bench_sqs ⇒ Object
- .bench_xml ⇒ Object
-
.queue_name_by_url(queue_url) ⇒ Object
Returns short queue name by url.
Instance Method Summary collapse
-
#add_grant(queue_url, grantee_email_address, permission = nil) ⇒ Object
Adds grants for user (identified by email he registered at Amazon).
-
#change_message_visibility(queue_url, message_id, visibility_timeout = 0) ⇒ Object
Changes message visibility timeout.
-
#clear_queue(queue_url) ⇒ Object
Removes all visible messages from queue.
-
#create_queue(queue_name, default_visibility_timeout = nil) ⇒ Object
Creates new queue.
-
#delete_message(queue_url, message_id) ⇒ Object
Deletes message from queue.
-
#delete_queue(queue_url, force_deletion = false) ⇒ Object
Deletes queue (queue must be empty or
force_deletion
must be set to true). -
#force_clear_queue(queue_url) ⇒ Object
Deletes queue then re-creates it (restores attributes also).
-
#force_delete_queue(queue_url) ⇒ Object
Deletes queue even if it has messages.
-
#generate_request(action, params = {}) ⇒ Object
Generates a request hash for the query API.
-
#generate_rest_request(method, param) ⇒ Object
Generates a request hash for the REST API.
-
#get_queue_attributes(queue_url, attribute = 'All') ⇒ Object
Retrieves the queue attribute(s).
-
#get_queue_length(queue_url) ⇒ Object
Returns approximate number of messages in queue.
-
#get_visibility_timeout(queue_url) ⇒ Object
Retrieves visibility timeout.
-
#initialize(aws_access_key_id = nil, aws_secret_access_key = nil, params = {}) ⇒ SqsInterface
constructor
Creates a new SqsInterface instance.
-
#list_grants(queue_url, grantee_email_address = nil, permission = nil) ⇒ Object
Retrieves hash of
grantee_id
=>perms
for this queue:. -
#list_queues(queue_name_prefix = nil) ⇒ Object
Lists all queues owned by this user that have names beginning with
queue_name_prefix
. -
#peek_message(queue_url, message_id) ⇒ Object
Peeks message from queue by message id.
-
#pop_message(queue_url) ⇒ Object
Pops (retrieves and deletes) first accessible message from queue.
-
#pop_messages(queue_url, number_of_messages = 1) ⇒ Object
Pops (retrieves and deletes) up to ‘number_of_messages’ from queue.
-
#queue_name_by_url(queue_url) ⇒ Object
Returns short queue name by url.
-
#queue_url_by_name(queue_name) ⇒ Object
Returns queue url by queue short name or
nil
if queue is not found. -
#receive_message(queue_url, visibility_timeout = nil) ⇒ Object
Reads first accessible message from queue.
-
#receive_messages(queue_url, number_of_messages = 1, visibility_timeout = nil) ⇒ Object
Retrieves a list of messages from queue.
-
#remove_grant(queue_url, grantee_email_address_or_id, permission = nil) ⇒ Object
Revokes permission from user.
-
#request_info(request, parser) ⇒ Object
Sends request to Amazon and parses the response Raises AwsError if any banana happened.
-
#send_message(queue_url, message) ⇒ Object
(also: #push_message)
Sends new message to queue.Returns ‘message_id’ or raises an exception.
-
#set_queue_attributes(queue_url, attribute, value) ⇒ Object
Sets queue attribute.
-
#set_visibility_timeout(queue_url, visibility_timeout = nil) ⇒ Object
Sets visibility timeout.
Methods included from RightAwsBaseInterface
#cache_hits?, caching, caching=, #caching?, #init, #multi_thread, #on_exception, #request_cache_or_info, #request_info_impl, #signed_service_params, #update_cache
Methods inherited from RightAwsBase
amazon_problems, amazon_problems=
Constructor Details
#initialize(aws_access_key_id = nil, aws_secret_access_key = nil, params = {}) ⇒ SqsInterface
Creates a new SqsInterface instance.
sqs = RightAws::SqsInterface.new('1E3GDYEOGFJPIT75KDT40','hgTHt68JY07JKUY08ftHYtERkjgtfERn57DFE379', {:multi_thread => true, :logger => Logger.new('/tmp/x.log')})
Params is a hash:
{:server => 'queue.amazonaws.com' # Amazon service host: 'queue.amazonaws.com'(default)
:port => 443 # Amazon service port: 80 or 443(default)
:multi_thread => true|false # Multi-threaded (connection per each thread): true or false(default)
:signature_version => '0' # The signature version : '0' or '1'(default)
:logger => Logger Object} # Logger instance: logs to STDOUT if omitted }
62 63 64 65 66 67 68 69 70 |
# File 'lib/sqs/right_sqs_interface.rb', line 62 def initialize(aws_access_key_id=nil, aws_secret_access_key=nil, params={}) init({ :name => 'SQS', :default_host => ENV['SQS_URL'] ? URI.parse(ENV['SQS_URL']).host : DEFAULT_HOST, :default_port => ENV['SQS_URL'] ? URI.parse(ENV['SQS_URL']).port : DEFAULT_PORT, :default_protocol => ENV['SQS_URL'] ? URI.parse(ENV['SQS_URL']).scheme : DEFAULT_PROTOCOL }, aws_access_key_id || ENV['AWS_ACCESS_KEY_ID'], aws_secret_access_key || ENV['AWS_SECRET_ACCESS_KEY'], params) end |
Class Method Details
.api ⇒ Object
46 47 48 |
# File 'lib/sqs/right_sqs_interface.rb', line 46 def self.api @@api end |
.bench_sqs ⇒ Object
41 42 43 |
# File 'lib/sqs/right_sqs_interface.rb', line 41 def self.bench_sqs @@bench.service end |
.bench_xml ⇒ Object
38 39 40 |
# File 'lib/sqs/right_sqs_interface.rb', line 38 def self.bench_xml @@bench.xml end |
.queue_name_by_url(queue_url) ⇒ Object
Returns short queue name by url.
RightSqs.queue_name_by_url('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> 'my_awesome_queue'
383 384 385 386 387 |
# File 'lib/sqs/right_sqs_interface.rb', line 383 def self.queue_name_by_url(queue_url) queue_url[/[^\/]*$/] rescue on_exception end |
Instance Method Details
#add_grant(queue_url, grantee_email_address, permission = nil) ⇒ Object
Adds grants for user (identified by email he registered at Amazon). Returns true
or an exception. Permission = ‘FULLCONTROL’ | ‘RECEIVEMESSAGE’ | ‘SENDMESSAGE’.
sqs.add_grant('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', '[email protected]', 'FULLCONTROL') #=> true
241 242 243 244 245 246 247 248 249 |
# File 'lib/sqs/right_sqs_interface.rb', line 241 def add_grant(queue_url, grantee_email_address, = nil) req_hash = generate_request('AddGrant', 'Grantee.EmailAddress' => grantee_email_address, 'Permission' => , :queue_url => queue_url) request_info(req_hash, SqsStatusParser.new(:logger => @logger)) rescue on_exception end |
#change_message_visibility(queue_url, message_id, visibility_timeout = 0) ⇒ Object
Changes message visibility timeout. Returns true
or an exception.
sqs.('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', '1234567890...0987654321', 10) #=> true
354 355 356 357 358 359 360 361 362 |
# File 'lib/sqs/right_sqs_interface.rb', line 354 def (queue_url, , visibility_timeout=0) req_hash = generate_request('ChangeMessageVisibility', 'MessageId' => , 'VisibilityTimeout' => visibility_timeout.to_s, :queue_url => queue_url) request_info(req_hash, SqsStatusParser.new(:logger => @logger)) rescue on_exception end |
#clear_queue(queue_url) ⇒ Object
Removes all visible messages from queue. Return true
or an exception.
sqs.clear_queue('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> true
413 414 415 416 417 418 |
# File 'lib/sqs/right_sqs_interface.rb', line 413 def clear_queue(queue_url) while (m = (queue_url)) ; end # delete all messages in queue true rescue on_exception end |
#create_queue(queue_name, default_visibility_timeout = nil) ⇒ Object
Creates new queue. Returns new queue link.
sqs.create_queue('my_awesome_queue') #=> 'http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue'
PS Some queue based requests may not become available until a couple of minutes after queue creation (permission grant and removal for example)
146 147 148 149 150 151 |
# File 'lib/sqs/right_sqs_interface.rb', line 146 def create_queue(queue_name, default_visibility_timeout=nil) req_hash = generate_request('CreateQueue', 'QueueName' => queue_name, 'DefaultVisibilityTimeout' => default_visibility_timeout || DEFAULT_VISIBILITY_TIMEOUT ) request_info(req_hash, SqsCreateQueueParser.new(:logger => @logger)) end |
#delete_message(queue_url, message_id) ⇒ Object
Deletes message from queue. Returns true
or an exception. Amazon returns true
on deletion of non-existent messages.
sqs.('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', '12345678904...0987654321') #=> true
341 342 343 344 345 346 347 348 |
# File 'lib/sqs/right_sqs_interface.rb', line 341 def (queue_url, ) req_hash = generate_request('DeleteMessage', 'MessageId' => , :queue_url => queue_url) request_info(req_hash, SqsStatusParser.new(:logger => @logger)) rescue on_exception end |
#delete_queue(queue_url, force_deletion = false) ⇒ Object
Deletes queue (queue must be empty or force_deletion
must be set to true). Queue is identified by url. Returns true
or an exception.
sqs.delete_queue('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue_2') #=> true
170 171 172 173 174 175 176 177 |
# File 'lib/sqs/right_sqs_interface.rb', line 170 def delete_queue(queue_url, force_deletion = false) req_hash = generate_request('DeleteQueue', 'ForceDeletion' => force_deletion.to_s, :queue_url => queue_url) request_info(req_hash, SqsStatusParser.new(:logger => @logger)) rescue on_exception end |
#force_clear_queue(queue_url) ⇒ Object
Deletes queue then re-creates it (restores attributes also). The fastest method to clear big queue or queue with invisible messages. Return true
or an exception.
sqs.force_clear_queue('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> true
PS This function is no longer supported. Amazon has changed the SQS semantics to require at least 60 seconds between queue deletion and creation. Hence this method will fail with an exception.
427 428 429 430 431 432 433 434 435 436 437 438 439 |
# File 'lib/sqs/right_sqs_interface.rb', line 427 def force_clear_queue(queue_url) queue_name = queue_name_by_url(queue_url) queue_attributes = get_queue_attributes(queue_url) force_delete_queue(queue_url) create_queue(queue_name) # hmmm... The next line is a trick. Amazon do not want change attributes immediately after queue creation # So we do 'empty' get_queue_attributes. Probably they need some time to allow attributes change. get_queue_attributes(queue_url) queue_attributes.each{ |attribute, value| set_queue_attributes(queue_url, attribute, value) } true rescue on_exception end |
#force_delete_queue(queue_url) ⇒ Object
Deletes queue even if it has messages. Return true
or an exception.
force_delete_queue('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> true
P.S. same as delete_queue('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', true)
446 447 448 449 450 |
# File 'lib/sqs/right_sqs_interface.rb', line 446 def force_delete_queue(queue_url) delete_queue(queue_url, true) rescue on_exception end |
#generate_request(action, params = {}) ⇒ Object
Generates a request hash for the query API
78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
# File 'lib/sqs/right_sqs_interface.rb', line 78 def generate_request(action, params={}) # :nodoc: # Sometimes we need to use queue uri (delete queue etc) # In that case we will use Symbol key: 'param[:queue_url]' service = params[:queue_url] ? URI(params[:queue_url]).path : '/' # remove unset(=optional) and symbolyc keys params.each{ |key, value| params.delete(key) if (value.nil? || key.is_a?(Symbol)) } # prepare output hash service_hash = { "Action" => action, "Expires" => (Time.now + REQUEST_TTL).utc.strftime("%Y-%m-%dT%H:%M:%SZ"), "AWSAccessKeyId" => @aws_access_key_id, "Version" => API_VERSION } service_hash.update(params) service_params = signed_service_params(@aws_secret_access_key, service_hash, :get, @params[:server], service) request = Net::HTTP::Get.new("#{AwsUtils::URLencode(service)}?#{service_params}") # prepare output hash { :request => request, :server => @params[:server], :port => @params[:port], :protocol => @params[:protocol], :proxy => @params[:proxy] } end |
#generate_rest_request(method, param) ⇒ Object
Generates a request hash for the REST API
101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 |
# File 'lib/sqs/right_sqs_interface.rb', line 101 def generate_rest_request(method, param) # :nodoc: queue_uri = param[:queue_url] ? URI(param[:queue_url]).path : '/' = param[:message] # extract message body if nesessary # remove unset(=optional) and symbolyc keys param.each{ |key, value| param.delete(key) if (value.nil? || key.is_a?(Symbol)) } # created request param_to_str = param.to_a.collect{|key,val| key.to_s + "=" + CGI::escape(val.to_s) }.join("&") param_to_str = "?#{param_to_str}" unless param_to_str.blank? request = "Net::HTTP::#{method.capitalize}".constantize.new("#{queue_uri}#{param_to_str}") request.body = if # set main headers request['content-md5'] = '' request['Content-Type'] = 'text/plain' request['Date'] = Time.now.httpdate # generate authorization string auth_string = "#{method.upcase}\n#{request['content-md5']}\n#{request['Content-Type']}\n#{request['Date']}\n#{CGI::unescape(queue_uri)}" signature = AwsUtils::sign(@aws_secret_access_key, auth_string) # set other headers request['Authorization'] = "AWS #{@aws_access_key_id}:#{signature}" request['AWS-Version'] = API_VERSION # prepare output hash { :request => request, :server => @params[:server], :port => @params[:port], :protocol => @params[:protocol], :proxy => @params[:proxy] } end |
#get_queue_attributes(queue_url, attribute = 'All') ⇒ Object
Retrieves the queue attribute(s). Returns a hash of attribute(s) or an exception.
sqs.get_queue_attributes('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> {"ApproximateNumberOfMessages"=>"0", "VisibilityTimeout"=>"30"}
183 184 185 186 187 188 189 190 |
# File 'lib/sqs/right_sqs_interface.rb', line 183 def get_queue_attributes(queue_url, attribute='All') req_hash = generate_request('GetQueueAttributes', 'Attribute' => attribute, :queue_url => queue_url) request_info(req_hash, SqsGetQueueAttributesParser.new(:logger => @logger)) rescue on_exception end |
#get_queue_length(queue_url) ⇒ Object
Returns approximate number of messages in queue.
sqs.get_queue_length('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> 3
403 404 405 406 407 |
# File 'lib/sqs/right_sqs_interface.rb', line 403 def get_queue_length(queue_url) get_queue_attributes(queue_url)['ApproximateNumberOfMessages'].to_i rescue on_exception end |
#get_visibility_timeout(queue_url) ⇒ Object
Retrieves visibility timeout.
sqs.get_visibility_timeout('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> 15
See also: get_queue_attributes
230 231 232 233 234 235 |
# File 'lib/sqs/right_sqs_interface.rb', line 230 def get_visibility_timeout(queue_url) req_hash = generate_request('GetVisibilityTimeout', :queue_url => queue_url ) request_info(req_hash, SqsGetVisibilityTimeoutParser.new(:logger => @logger)) rescue on_exception end |
#list_grants(queue_url, grantee_email_address = nil, permission = nil) ⇒ Object
Retrieves hash of grantee_id
=> perms
for this queue:
sqs.list_grants('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=>
{"000000000000000000000001111111111117476c7fea6efb2c3347ac3ab2792a"=>{:name=>"root", :perms=>["FULLCONTROL"]},
"00000000000000000000000111111111111e5828344600fc9e4a784a09e97041"=>{:name=>"myawesomefriend", :perms=>["FULLCONTROL"]}
257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 |
# File 'lib/sqs/right_sqs_interface.rb', line 257 def list_grants(queue_url, grantee_email_address=nil, = nil) req_hash = generate_request('ListGrants', 'Grantee.EmailAddress' => grantee_email_address, 'Permission' => , :queue_url => queue_url) response = request_info(req_hash, SqsListGrantsParser.new(:logger => @logger)) # One user may have up to 3 permission records for every queue. # We will join these records to one. result = {} response.each do |perm| id = perm[:id] # create hash for new user if unexisit result[id] = {:perms=>[]} unless result[id] # fill current grantee params result[id][:perms] << perm[:permission] result[id][:name] = perm[:name] end result end |
#list_queues(queue_name_prefix = nil) ⇒ Object
Lists all queues owned by this user that have names beginning with queue_name_prefix
. If queue_name_prefix
is omitted then retrieves a list of all queues.
sqs.create_queue('my_awesome_queue')
sqs.create_queue('my_awesome_queue_2')
sqs.list_queues('my_awesome') #=> ['http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue','http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue_2']
159 160 161 162 163 164 |
# File 'lib/sqs/right_sqs_interface.rb', line 159 def list_queues(queue_name_prefix=nil) req_hash = generate_request('ListQueues', 'QueueNamePrefix' => queue_name_prefix) request_info(req_hash, SqsListQueuesParser.new(:logger => @logger)) rescue on_exception end |
#peek_message(queue_url, message_id) ⇒ Object
Peeks message from queue by message id. Returns message in format of {:id=>'message_id', :body=>'message_body'}
or nil
.
sqs.('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', '1234567890...0987654321') #=>
{:id=>"12345678904GEZX9746N|0N9ED344VK5Z3SV1DTM0|1RVYH4X3TJ0987654321", :body=>"message_1"}
315 316 317 318 319 320 321 |
# File 'lib/sqs/right_sqs_interface.rb', line 315 def (queue_url, ) req_hash = generate_rest_request('GET', :queue_url => "#{queue_url}/#{CGI::escape }" ) = request_info(req_hash, SqsReceiveMessagesParser.new(:logger => @logger)) .blank? ? nil : [0] rescue on_exception end |
#pop_message(queue_url) ⇒ Object
Pops (retrieves and deletes) first accessible message from queue. Returns the message in format {:id=>'message_id', :body=>'message_body'}
or nil
.
sqs.('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=>
{:id=>"12345678904GEZX9746N|0N9ED344VK5Z3SV1DTM0|1RVYH4X3TJ0987654321", :body=>"message_1"}
487 488 489 490 491 492 |
# File 'lib/sqs/right_sqs_interface.rb', line 487 def (queue_url) = (queue_url) .blank? ? nil : [0] rescue on_exception end |
#pop_messages(queue_url, number_of_messages = 1) ⇒ Object
Pops (retrieves and deletes) up to ‘number_of_messages’ from queue. Returns an array of retrieved messages in format: [{:id=>'message_id', :body=>'message_body'}]
.
sqs.pop_messages('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', 3) #=>
[{:id=>"12345678904GEZX9746N|0N9ED344VK5Z3SV1DTM0|1RVYH4X3TJ0987654321", :body=>"message_1"}, ..., {}]
472 473 474 475 476 477 478 479 480 |
# File 'lib/sqs/right_sqs_interface.rb', line 472 def (queue_url, =1) = (queue_url, ) .each do || (queue_url, [:id]) end rescue on_exception end |
#queue_name_by_url(queue_url) ⇒ Object
Returns short queue name by url.
sqs.queue_name_by_url('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue') #=> 'my_awesome_queue'
393 394 395 396 397 |
# File 'lib/sqs/right_sqs_interface.rb', line 393 def queue_name_by_url(queue_url) self.class.queue_name_by_url(queue_url) rescue on_exception end |
#queue_url_by_name(queue_name) ⇒ Object
Returns queue url by queue short name or nil
if queue is not found
sqs.queue_url_by_name('my_awesome_queue') #=> 'http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue'
368 369 370 371 372 373 374 375 376 377 |
# File 'lib/sqs/right_sqs_interface.rb', line 368 def queue_url_by_name(queue_name) return queue_name if queue_name.include?('/') queue_urls = list_queues(queue_name) queue_urls.each do |queue_url| return queue_url if queue_name_by_url(queue_url) == queue_name end nil rescue on_exception end |
#receive_message(queue_url, visibility_timeout = nil) ⇒ Object
Reads first accessible message from queue. Returns message as a hash: {:id=>'message_id', :body=>'message_body'}
or nil
.
sqs.('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', 10) #=>
{:id=>"12345678904GEZX9746N|0N9ED344VK5Z3SV1DTM0|1RVYH4X3TJ0987654321", :body=>"message_1"}
457 458 459 460 461 462 |
# File 'lib/sqs/right_sqs_interface.rb', line 457 def (queue_url, visibility_timeout=nil) result = (queue_url, 1, visibility_timeout) result.blank? ? nil : result[0] rescue on_exception end |
#receive_messages(queue_url, number_of_messages = 1, visibility_timeout = nil) ⇒ Object
Retrieves a list of messages from queue. Returns an array of hashes in format: {:id=>'message_id', body=>'message_body'}
sqs.receive_messages('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue',10, 5) #=>
[{:id=>"12345678904GEZX9746N|0N9ED344VK5Z3SV1DTM0|1RVYH4X3TJ0987654321", :body=>"message_1"}, ..., {}]
P.S. Usually returns fewer messages than requested even if they are available.
299 300 301 302 303 304 305 306 307 308 |
# File 'lib/sqs/right_sqs_interface.rb', line 299 def (queue_url, =1, visibility_timeout=nil) return [] if == 0 req_hash = generate_rest_request('GET', 'NumberOfMessages' => , 'VisibilityTimeout' => visibility_timeout, :queue_url => "#{queue_url}/front" ) request_info(req_hash, SqsReceiveMessagesParser.new(:logger => @logger)) rescue on_exception end |
#remove_grant(queue_url, grantee_email_address_or_id, permission = nil) ⇒ Object
Revokes permission from user. Returns true
or an exception.
sqs.remove_grant('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', '[email protected]', 'FULLCONTROL') #=> true
281 282 283 284 285 286 287 288 289 290 |
# File 'lib/sqs/right_sqs_interface.rb', line 281 def remove_grant(queue_url, grantee_email_address_or_id, = nil) grantee_key = grantee_email_address_or_id.include?('@') ? 'Grantee.EmailAddress' : 'Grantee.ID' req_hash = generate_request('RemoveGrant', grantee_key => grantee_email_address_or_id, 'Permission' => , :queue_url => queue_url) request_info(req_hash, SqsStatusParser.new(:logger => @logger)) rescue on_exception end |
#request_info(request, parser) ⇒ Object
Sends request to Amazon and parses the response Raises AwsError if any banana happened
132 133 134 135 136 |
# File 'lib/sqs/right_sqs_interface.rb', line 132 def request_info(request, parser) # :nodoc: thread = @params[:multi_thread] ? Thread.current : Thread.main thread[:sqs_connection] ||= Rightscale::HttpConnection.new(:exception => AwsError, :logger => @logger) request_info_impl(thread[:sqs_connection], @@bench, request, parser) end |
#send_message(queue_url, message) ⇒ Object Also known as: push_message
Sends new message to queue.Returns ‘message_id’ or raises an exception.
sqs.('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', 'message_1') #=> "1234567890...0987654321"
327 328 329 330 331 332 333 334 |
# File 'lib/sqs/right_sqs_interface.rb', line 327 def (queue_url, ) req_hash = generate_rest_request('PUT', :message => , :queue_url => "#{queue_url}/back") request_info(req_hash, SqsSendMessagesParser.new(:logger => @logger)) rescue on_exception end |
#set_queue_attributes(queue_url, attribute, value) ⇒ Object
Sets queue attribute. Returns true
or an exception.
sqs.set_queue_attributes('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', "VisibilityTimeout", 10) #=> true
P.S. Amazon returns success even if the attribute does not exist. Also, attribute values may not be immediately available to other queries for some time after an update (see the SQS documentation for semantics).
199 200 201 202 203 204 205 206 207 |
# File 'lib/sqs/right_sqs_interface.rb', line 199 def set_queue_attributes(queue_url, attribute, value) req_hash = generate_request('SetQueueAttributes', 'Attribute' => attribute, 'Value' => value, :queue_url => queue_url) request_info(req_hash, SqsStatusParser.new(:logger => @logger)) rescue on_exception end |
#set_visibility_timeout(queue_url, visibility_timeout = nil) ⇒ Object
Sets visibility timeout. Returns true
or an exception.
sqs.set_visibility_timeout('http://queue.amazonaws.com/ZZ7XXXYYYBINS/my_awesome_queue', 15) #=> true
See also: set_queue_attributes
215 216 217 218 219 220 221 222 |
# File 'lib/sqs/right_sqs_interface.rb', line 215 def set_visibility_timeout(queue_url, visibility_timeout=nil) req_hash = generate_request('SetVisibilityTimeout', 'VisibilityTimeout' => visibility_timeout || DEFAULT_VISIBILITY_TIMEOUT, :queue_url => queue_url ) request_info(req_hash, SqsStatusParser.new(:logger => @logger)) rescue on_exception end |