Module: Smash::CloudPowers::Synapse::Queue

Includes:
AwsResources, Helper
Included in:
Smash::CloudPowers::SelfAwareness, Smash::CloudPowers::Synapse, Board
Defined in:
lib/cloud_powers/synapse/queue/board.rb,
lib/cloud_powers/synapse/queue/queue.rb

Defined Under Namespace

Classes: Board

Instance Method Summary collapse

Methods included from AwsResources

#ec2, #image, #kinesis, #region, #s3, #sns, #sqs

Methods included from Zenv

#env_vars, #file_tree_search, #i_vars, #project_root, #project_root=, #system_vars, #zfind

Methods included from Helper

#attr_map!, #called_from, #create_logger, #errors, #format_error_message, #log_file, #logger, #smart_retry, #task_path, #task_require_path, #to_camel, #to_i_var, #to_pascal, #to_ruby_file_name, #to_snake, #update_message_body, #valid_json?

Methods included from Auth

creds, region

Instance Method Details

#board_name(url) ⇒ Object

This method can be used to parse a queue name from its address. It can be handy if you need the name of a queue but you don't want the overhead of creating a Board object.



13
14
15
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 13

def board_name(url)
  url.to_s.split('/').last
end

#build_queue(name) ⇒ Object

This method builds a Queue::Board object for you to use but doesn't invoke the #create! method, so no API call is made to create the queue on SQS. This can be used if the board already exists. @params: name : name of the queue you want to interact with @returns: Queue::Board



33
34
35
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 33

def build_queue(name)
  Board.build(to_camel(name))
end

#create_queue!(name) ⇒ Object

This method allows you to create a queue on SQS without explicitly creating a Board object @params: name : The name of the queue to be created @returns: Queue::Board



20
21
22
23
24
25
26
27
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 20

def create_queue!(name)
  begin
    Board.create!(to_camel(name))
  rescue Aws::SQS::Errors::QueueDeletedRecently => e
    sleep 5
    retry
  end
end

#delete_queue_message(queue, opts = {}) ⇒ Object

Deletes a queue message without caring about reading/interacting with the message. This is usually used for progress tracking, ie; a Neuron takes a message from the Backlog, moves it to WIP and deletes it from Backlog. Then repeats these steps for the remaining States in the Workflow @params: queue [, opts ] queue is the name of the queue to interact with opts is a configuration hash for the SQS::QueuePoller



43
44
45
46
47
48
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 43

def delete_queue_message(queue, opts = {})
  poll(queue, opts) do |msg, stats|
    poller(queue).delete_message(msg)
    throw :stop_polling
  end
end

#get_queue_message_count(board_url) ⇒ Object

This method is used to gain the approximate count of messages in a given queue @params: board_url : The URL for the board you need to get a count from @returns: float representation of the count



53
54
55
56
57
58
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 53

def get_queue_message_count(board_url)
  sqs.get_queue_attributes(
    queue_url: board_url,
    attribute_names: ['ApproximateNumberOfMessages']
  ).attributes['ApproximateNumberOfMessages'].to_f
end

#pluck_queue_message(board) ⇒ Object

@params: board<String|symbol>: The name of the board



62
63
64
65
66
67
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 62

def pluck_queue_message(board)
  poll(board) do |msg, poller|
    poller.delete_message(msg)
    return valid_json?(msg.body) ? JSON.parse(msg.body) : msg.body
  end
end

#poll(board, opts = {}) ⇒ Object

Polls the given board with the given options hash and a block that interacts with the message that is retrieved from the queue @params: board [, opts ] board is the name of the queue you want to poll opts can have any AWS::SQS polling option &block is the block that is used to interact with the message that was retrieved



76
77
78
79
80
81
82
83
84
85
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 76

def poll(board, opts = {})
  this_poller = queue_poller(board)
  results = nil
  this_poller.poll(opts) do |msg|
    results = yield msg, this_poller if block_given?
    this_poller.delete_message(msg)
    throw :stop_polling
  end
  results
end

#queue_exists?(name) ⇒ Boolean

Checks SQS for the existence of this queue

Returns:

  • (Boolean) 


105
106
107
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 105

def queue_exists?(name)
  !sqs.list_queues(queue_name_prefix: name).queue_urls.empty?
end

#queue_poller(board_name) ⇒ Object

This method can be used to gain a SQS::QueuePoller. It creates a Board object, the Board then sends the API call to SQS to create the queue and sets an instance variable, using the board's name, to the Board object itself @params: board_name : name of the Queue you want to gain a QueuePoller for @returns: @<board_name:Queue::Board>



92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 92

def queue_poller(board_name)
  board = Board.create!(board_name)

  unless instance_variable_defined?(board.i_var)
    instance_variable_set(
      board.i_var,
      board
    )
  end
  instance_variable_get(board.i_var).poller
end

#queue_search(name) ⇒ Object

Searches for a queue based on the name @params: name @returns: queue_urls 



112
113
114
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 112

def queue_search(name)
  sqs.list_queues(queue_name_prefix: name).queue_urls
end

#send_queue_message(address, message) ⇒ Object

Sends a given message to a given queue @params: address : address of the queue you want to interact with @returns: queue_urls <Array>



119
120
121
122
123
124
 
# File 'lib/cloud_powers/synapse/queue/queue.rb', line 119

def send_queue_message(address, message)
  sqs.send_message(
    queue_url: address,
    message_body: message
  )
end