Class: IronMQ::Queue

Inherits:
ResponseBase show all
Defined in:
lib/iron_mq/queues.rb

Instance Attribute Summary collapse

Attributes inherited from ResponseBase

#code

Instance Method Summary collapse

Methods inherited from ResponseBase

#[], #msg

Constructor Details

#initialize(client, queue_name) ⇒ Queue

Returns a new instance of Queue.



8
9
10
11
 
# File 'lib/iron_mq/queues.rb', line 8

def initialize(client, queue_name)
  @client = client
  @name = queue_name
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



6
7
8
 
# File 'lib/iron_mq/queues.rb', line 6

def name
  @name
end

#raw ⇒ Object (readonly)

Returns the value of attribute raw.



6
7
8
 
# File 'lib/iron_mq/queues.rb', line 6

def raw
  @raw
end

Instance Method Details

#add_alert(alert) ⇒ Object 



160
161
162
 
# File 'lib/iron_mq/queues.rb', line 160

def add_alert(alert)
  add_alerts([alert])
end

#add_alerts(alerts) ⇒ Object 



156
157
158
 
# File 'lib/iron_mq/queues.rb', line 156

def add_alerts(alerts)
  call_api_and_parse_response(:patch, '', queue: {alerts: alerts})
end

#add_subscriber(subscriber, options = {}) ⇒ Object

options for backward compatibility



114
115
116
 
# File 'lib/iron_mq/queues.rb', line 114

def add_subscriber(subscriber, options = {})
  add_subscribers([subscriber])
end

#add_subscribers(subscribers) ⇒ Object 



109
110
111
 
# File 'lib/iron_mq/queues.rb', line 109

def add_subscribers(subscribers)
  call_api_and_parse_response(:post, '/subscribers',{subscribers: subscribers})
end

#alerts ⇒ Object 



180
181
182
183
184
 
# File 'lib/iron_mq/queues.rb', line 180

def alerts
  load
  return nil unless @raw['alerts']
  to_alerts(@raw['alerts'])
end

#call_api_and_parse_response(meth, ext_path = '', options = {}, instantiate = true, ignore404 = false) ⇒ Object 



273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
 
# File 'lib/iron_mq/queues.rb', line 273

def call_api_and_parse_response(meth, ext_path = '', options = {},
                                instantiate = true, ignore404 = false)
  response =
    if meth.to_s == 'delete'
      headers = options.delete(:headers) ||
                options.delete('headers') ||
                Hash.new
      headers['Content-Type'] = 'application/json'
      @client.parse_response(@client.send(meth,
                                          "#{path(ext_path)}",
                                          options, headers))
    else
      @client.parse_response(@client.send(meth,
                                          "#{path(ext_path)}",
                                          options))
    end

  instantiate ? ResponseBase.new(response) : response
end

#clear ⇒ Object Also known as: clear_queue 



64
65
66
 
# File 'lib/iron_mq/queues.rb', line 64

def clear
  call_api_and_parse_response(:delete, '/messages', {}, false, true)
end

#clear_alerts ⇒ Object 



176
177
178
 
# File 'lib/iron_mq/queues.rb', line 176

def clear_alerts
  replace_alerts([])
end

#delete(message_id, reservation_id = nil) ⇒ Object

Backward compatibility



86
87
88
89
90
91
92
93
94
 
# File 'lib/iron_mq/queues.rb', line 86

def delete(message_id, reservation_id = nil)
  # API does not accept any options
  options = {}
  options['id'] = message_id
  unless reservation_id.nil?
    options['reservation_id'] = reservation_id
  end
  Message.new(self, options).delete
end

#delete_messages(ids) ⇒ Object

Accepts an array of message ids



97
98
99
 
# File 'lib/iron_mq/queues.rb', line 97

def delete_messages(ids)
  call_api_and_parse_response(:delete, '/messages', ids: ids)
end

#delete_queue ⇒ Object

Backward compatibility, better name is delete



71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/iron_mq/queues.rb', line 71

def delete_queue
  r = call_api_and_parse_response(:delete)
  @raw = nil
  return r
rescue Rest::HttpError => ex
  #if ex.code == 404
  #  Rest.logger.info('Delete got 404, safe to ignore.')
  #  # return ResponseBase as normal
  #  ResponseBase.new({'msg' => 'Deleted'}, 404)
  #else
    raise ex
  #end
end

#delete_reserved_messages(messages) ⇒ Object 



101
102
103
104
105
106
107
 
# File 'lib/iron_mq/queues.rb', line 101

def delete_reserved_messages(messages)
  ids = messages.map do |message|
    {id: message.id, reservation_id: message.reservation_id}
  end

  call_api_and_parse_response(:delete, '/messages', ids: ids)
end

#get_message(id) ⇒ Object 



243
244
245
246
 
# File 'lib/iron_mq/queues.rb', line 243

def get_message(id)
  resp = call_api_and_parse_response(:get, "/messages/#{id}", {}, false)
  Message.new(self, resp['message'])
end

#id ⇒ Object 



29
30
31
 
# File 'lib/iron_mq/queues.rb', line 29

def id
  load['id']
end

#info ⇒ Object 



13
14
15
 
# File 'lib/iron_mq/queues.rb', line 13

def info
  load
end

#load ⇒ Object

this is only run once if it hasn't been called before unless force is true, then it will force reload.



18
19
20
21
22
 
# File 'lib/iron_mq/queues.rb', line 18

def load
  reload if @raw.nil?

  @raw['queue']
end

#messages ⇒ Object

Backward compatibility



239
240
241
 
# File 'lib/iron_mq/queues.rb', line 239

def messages
  self
end

#peek_messages(options = {}) ⇒ Object Also known as: peek 



248
249
250
251
252
 
# File 'lib/iron_mq/queues.rb', line 248

def peek_messages(options = {})
  resp = call_api_and_parse_response(:get, '/messages', options)

  process_messages(resp['messages'], options)
end

#poll_messages(options = {}, &block) ⇒ Object Also known as: poll 



256
257
258
259
260
261
262
263
264
265
266
267
268
269
 
# File 'lib/iron_mq/queues.rb', line 256

def poll_messages(options = {}, &block)
  sleep_duration = options[:sleep_duration] || 1

  while true
    msg = get_messages(options.merge(:n => 1))
    if msg.nil?
      options[:break_if_nil] ? break : sleep(sleep_duration)
    else
      yield msg
      # Delete message after processing
      msg.delete
    end
  end
end

#post_messages(payload, options = {}) ⇒ Object Also known as: post 



186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/iron_mq/queues.rb', line 186

def post_messages(payload, options = {})
  batch = false

  instantiate = [options.delete(:instantiate),
                 options.delete('instantiate')].compact.first

  msgs = if payload.is_a?(Array)
           batch = true
           # FIXME: This maybe better to process Array of Objects the same way as for single message.
           #
           #          payload.map { |msg| options.merge(:body => msg) }
           #
           #        For now user must pass objects like `[{:body => msg1}, {:body => msg2}]`
           payload.map { |msg| msg.merge(options) }
         else
           [options.merge(body: payload)]
         end

  # Do not instantiate response
  res = call_api_and_parse_response(:post, '/messages',
                                    {messages: msgs}, false)

  if instantiate
    n = batch ? 2 : 1
    msg_ids = res['ids'].map { |id| {'id' => id} }

    process_messages(msg_ids, {n: n})
  else
    if batch
      # FIXME: Return Array of ResponseBase instead, it seems more clear than raw response
      #
      #          res['ids'].map { |id| ResponseBase.new({'id' => id, 'msg' => res['msg']}) }
      #
      ResponseBase.new(res) # Backward capable
    else
      ResponseBase.new({'id' => res['ids'][0], 'msg' => res['msg']})
    end
  end
end

#push_info ⇒ Object 



49
50
51
 
# File 'lib/iron_mq/queues.rb', line 49

def push_info
  load['push']
end

#push_queue? ⇒ Boolean

Returns:

  • (Boolean) 


45
46
47
 
# File 'lib/iron_mq/queues.rb', line 45

def push_queue?
  ['multicast', 'unicast'].include?(type)
end

#reload ⇒ Object 



24
25
26
27
 
# File 'lib/iron_mq/queues.rb', line 24

def reload
  @raw = call_api_and_parse_response(:get, '', {}, false, true)
  self
end

#remove_alert(alert) ⇒ Object 



168
169
170
 
# File 'lib/iron_mq/queues.rb', line 168

def remove_alert(alert)
  remove_alerts([alert])
end

#remove_alerts(alerts) ⇒ Object 



164
165
166
 
# File 'lib/iron_mq/queues.rb', line 164

def remove_alerts(alerts)
  call_api_and_parse_response(:delete, '/alerts', alerts: alerts)
end

#remove_subscriber(subscriber) ⇒ Object 



129
130
131
 
# File 'lib/iron_mq/queues.rb', line 129

def remove_subscriber(subscriber)
  remove_subscribers([subscriber])
end

#remove_subscribers(subscribers) ⇒ Object 



118
119
120
121
122
123
124
125
126
127
 
# File 'lib/iron_mq/queues.rb', line 118

def remove_subscribers(subscribers)
  call_api_and_parse_response(:delete,
                              '/subscribers',
                              {
                                subscribers: subscribers,
                                headers: {
                                  'Content-Type' => @client.content_type
                                }
                              })
end

#replace_alerts(alerts) ⇒ Object 



172
173
174
 
# File 'lib/iron_mq/queues.rb', line 172

def replace_alerts(alerts)
  call_api_and_parse_response(:put, '/alerts', alerts: alerts)
end

#replace_subscriber(subscriber) ⇒ Object 



144
145
146
 
# File 'lib/iron_mq/queues.rb', line 144

def replace_subscriber(subscriber)
  replace_subscribers([subscriber])
end

#replace_subscribers(subscribers) ⇒ Object 



133
134
135
136
137
138
139
140
141
142
 
# File 'lib/iron_mq/queues.rb', line 133

def replace_subscribers(subscribers)
  call_api_and_parse_response(:put,
                              '/subscribers',
                              {
                                  subscribers: subscribers,
                                  headers: {
                                      'Content-Type' => @client.content_type
                                  }
                              })
end

#reserve_messages(options = {}) ⇒ Object Also known as: get, get_messages, reserve 



228
229
230
231
 
# File 'lib/iron_mq/queues.rb', line 228

def reserve_messages(options = {})
  resp = call_api_and_parse_response(:post, '/reservations', options, false)
  process_messages(resp['messages'], options)
end

#size ⇒ Object 



33
34
35
 
# File 'lib/iron_mq/queues.rb', line 33

def size
  load['size'].to_i
end

#subscribers(options = {}) ⇒ Object

options was kept for backward compatibility



149
150
151
152
153
154
 
# File 'lib/iron_mq/queues.rb', line 149

def subscribers(options = {})
  load
  return [] if info['push'].nil? || info['push']['subscribers'].nil?

  info['push']['subscribers'].map { |s| Subscriber.new(s, self, options) }
end

#total_messages ⇒ Object 



37
38
39
 
# File 'lib/iron_mq/queues.rb', line 37

def total_messages
  load['total_messages'].to_i
end

#type ⇒ Object 



41
42
43
 
# File 'lib/iron_mq/queues.rb', line 41

def type
  load['type']
end

#update(options = {}) ⇒ Object Also known as: update_queue 



53
54
55
56
57
58
59
60
 
# File 'lib/iron_mq/queues.rb', line 53

def update(options={})
  res = call_api_and_parse_response(:put, '', {queue: options})

  oldinfo = @raw ? @raw['queue'] : {}
  @raw = res.raw
  @raw['queue'].merge!('size' => oldinfo['size'], 'total_messages' => oldinfo['total_messages'])
  res
end