Class: BBK::AMQP::Message

Inherits:
Object
  • Object
show all
Defined in:
lib/bbk/amqp/message.rb

Overview

Store information about consumed AMQP message

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(consumer, delivery_info, properties, body) ⇒ Message

Returns a new instance of Message.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/bbk/amqp/message.rb', line 10

def initialize(consumer, delivery_info, properties, body)
  @consumer = consumer
  @properties = properties.to_h.with_indifferent_access
  @body = body
  amqp_consumer = delivery_info[:consumer]
  @delivery_info = delivery_info.to_h.merge(
    message_consumer: consumer,
    protocols:        consumer.protocols,
    queue:            amqp_consumer&.queue_name
  )
  @headers = @properties.except(:headers).merge(properties[:headers].presence || {}).with_indifferent_access
  @payload = begin
    Oj.load(body).with_indifferent_access
  rescue StandardError
    {}.with_indifferent_access
  end
end

Instance Attribute Details

#body ⇒ Object (readonly)

Returns the value of attribute body.



8
9
10
 
# File 'lib/bbk/amqp/message.rb', line 8

def body
  @body
end

#consumer ⇒ Object (readonly)

Returns the value of attribute consumer.



8
9
10
 
# File 'lib/bbk/amqp/message.rb', line 8

def consumer
  @consumer
end

#delivery_info ⇒ Object (readonly)

Returns the value of attribute delivery_info.



8
9
10
 
# File 'lib/bbk/amqp/message.rb', line 8

def delivery_info
  @delivery_info
end

#headers ⇒ Object (readonly)

Returns the value of attribute headers.



8
9
10
 
# File 'lib/bbk/amqp/message.rb', line 8

def headers
  @headers
end

#payload ⇒ Object (readonly)

Returns the value of attribute payload.



8
9
10
 
# File 'lib/bbk/amqp/message.rb', line 8

def payload
  @payload
end

#properties ⇒ Object (readonly)

Returns the value of attribute properties.



8
9
10
 
# File 'lib/bbk/amqp/message.rb', line 8

def properties
  @properties
end

Instance Method Details

#ack(*args, answer: nil, **kwargs) ⇒ Object 



40
41
42
 
# File 'lib/bbk/amqp/message.rb', line 40

def ack(*args, answer: nil, **kwargs)
  consumer.ack(self, *args, answer: answer, **kwargs)
end

#clone ⇒ Object 



48
49
50
 
# File 'lib/bbk/amqp/message.rb', line 48

def clone
  self.class.new(consumer, delivery_info, properties, body)
end

#message_id ⇒ Object 



28
29
30
 
# File 'lib/bbk/amqp/message.rb', line 28

def message_id
  headers[:message_id]
end

#nack(*args, error: nil, **kwargs) ⇒ Object 



44
45
46
 
# File 'lib/bbk/amqp/message.rb', line 44

def nack(*args, error: nil, **kwargs)
  consumer.nack(self, *args, error: error, **kwargs)
end

#protocol ⇒ Object 



52
53
54
 
# File 'lib/bbk/amqp/message.rb', line 52

def protocol
  consumer&.protocols&.first
end

#reply_to ⇒ Object 



32
33
34
 
# File 'lib/bbk/amqp/message.rb', line 32

def reply_to
  headers[:reply_to]
end

#user_id ⇒ Object 



36
37
38
 
# File 'lib/bbk/amqp/message.rb', line 36

def user_id
  headers[:user_id]
end