Class: Hoze::PubSubMessage

Inherits:
Message
  • Object
show all
Defined in:
lib/hoze/pubsub/message.rb

Instance Attribute Summary

Attributes inherited from Message

#metadata, #payload, #source, #source_message, #timestamp

Instance Method Summary collapse

Constructor Details

#initialize(source_message, source) ⇒ PubSubMessage

Returns a new instance of PubSubMessage.



7
8
9
10
11
12
13
14
15
 
# File 'lib/hoze/pubsub/message.rb', line 7

def initialize source_message, source
  @payload = source_message.data
  @timestamp = source_message.published_at
  @metadata = source_message.attributes
  @metadata['tries'] = (@metadata['tries'].to_i || 0) + 1
  @source_message = source_message
  @source = source
  @acked = false
end

Instance Method Details

#ack! ⇒ Object

Acknowledge the message



23
24
25
26
 
# File 'lib/hoze/pubsub/message.rb', line 23

def ack!
  @source_message.ack! unless @acked
  @acked = true
end

#delay!(seconds) ⇒ Object

Ask for more time before acknowledging



18
19
20
 
# File 'lib/hoze/pubsub/message.rb', line 18

def delay! seconds
  @source_message.delay! seconds
end

#reject! ⇒ Object

Release the message



29
30
31
 
# File 'lib/hoze/pubsub/message.rb', line 29

def reject!
  @source_message.reject!
end

#retry! ⇒ Object

Retry the message



34
35
36
37
38
39
40
41
42
43
 
# File 'lib/hoze/pubsub/message.rb', line 34

def retry!
  ack!
  retry_config = @source.max_tries || 1
  if @metadata['tries'] < retry_config
    puts "RETRYING: #{@metadata['tries']} / #{retry_config}"
    @source.topic.publish_async(@payload, @metadata)
  else
    puts "Dead message #{@source_message.inspect}"
  end
end