Class: Legion::Transport::Message

Inherits:
Object
  • Object
show all
Extended by:
Logging::Helper
Includes:
Common
Defined in:
lib/legion/transport/message.rb

Constant Summary collapse

ENVELOPE_KEYS = 
%i[
  headers content_type content_encoding persistent expiration
  priority app_id user_id reply_to correlation_id message_id
  routing_key exchange type
].freeze

Constants included from Common

Common::NAMESPACE_BOUNDARIES

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Common

#channel_open?, #close, #close!, #deep_merge, #generate_consumer_tag, #open_channel, #options_builder

Constructor Details

#initialize(**options) ⇒ Message

Returns a new instance of Message.



14
15
16
17
 
# File 'lib/legion/transport/message.rb', line 14

def initialize(**options)
  @options = options
  validate
end

Class Method Details

.max_payload_bytes ⇒ Object 



19
20
21
22
23
24
 
# File 'lib/legion/transport/message.rb', line 19

def self.max_payload_bytes
  Legion::Settings[:transport][:max_payload_bytes]
rescue StandardError => e
  handle_exception(e, level: :warn, handled: true, operation: 'transport.message.max_payload_bytes')
  1_048_576
end

Instance Method Details

#app_id ⇒ Object 



59
60
61
 
# File 'lib/legion/transport/message.rb', line 59

def app_id
  @options[:app_id] || 'legion'
end

#channel ⇒ Object 



199
200
201
 
# File 'lib/legion/transport/message.rb', line 199

def channel
  Legion::Transport::Connection.channel
end

#content_encoding ⇒ Object 



183
184
185
 
# File 'lib/legion/transport/message.rb', line 183

def content_encoding
  'identity'
end

#content_type ⇒ Object 



179
180
181
 
# File 'lib/legion/transport/message.rb', line 179

def content_type
  'application/json'
end

#correlation_id ⇒ Object

ID of the message that this message is a reply to. Links subtasks back to the parent task.



78
79
80
 
# File 'lib/legion/transport/message.rb', line 78

def correlation_id
  @options[:correlation_id] || @options[:parent_id] || @options[:task_id]
end

#encode_message ⇒ Object 



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/legion/transport/message.rb', line 112

def encode_message
  message_payload = message
  message_payload = Legion::JSON.dump(message_payload) unless message_payload.is_a? String

  if encrypt?
    encrypted = Legion::Crypt.encrypt(message_payload)
    headers[:iv] = encrypted[:iv]
    @options[:content_encoding] = 'encrypted/cs'
    log.debug "Message encrypted with content_encoding=encrypted/cs class=#{self.class.name}"
    return encrypted[:enciphered_message]
  else
    @options[:content_encoding] = 'identity'
  end

  message_payload
end

#encrypt? ⇒ Boolean

Returns:

  • (Boolean) 


133
134
135
136
137
138
139
140
 
# File 'lib/legion/transport/message.rb', line 133

def encrypt?
  should_encrypt = if @options.key?(:encrypt)
                     @options[:encrypt]
                   else
                     Legion::Settings[:transport][:messages][:encrypt]
                   end
  should_encrypt && Legion::Settings[:crypt][:cs_encrypt_ready]
end

#encrypt_message(message, _type = 'cs') ⇒ Object 



129
130
131
 
# File 'lib/legion/transport/message.rb', line 129

def encrypt_message(message, _type = 'cs')
  Legion::Crypt.encrypt(message)
end

#exchange ⇒ Object 



147
148
149
 
# File 'lib/legion/transport/message.rb', line 147

def exchange
  Kernel.const_get(exchange_name)
end

#exchange_name ⇒ Object 



142
143
144
145
 
# File 'lib/legion/transport/message.rb', line 142

def exchange_name
  parts = derive_extension_parts
  "Legion::Extensions::#{parts.join('::')}::Transport::Exchanges::#{parts.first}"
end

#expiration ⇒ Object 



86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/legion/transport/message.rb', line 86

def expiration
  if @options.key? :expiration
    @options[:expiration]
  elsif @options.key? :ttl
    @options[:ttl]
  elsif Legion::Transport.settings[:messages].key? :expiration
    Legion::Transport.settings[:messages][:expiration]
  elsif Legion::Transport.settings[:messages].key? :ttl
    Legion::Transport.settings[:messages][:ttl]
  end
end

#headers ⇒ Object 



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/legion/transport/message.rb', line 151

def headers
  @options[:headers] ||= Concurrent::Hash.new
  @options[:headers]['legion_protocol_version'] ||= '2.0'
  inject_region_header
  inject_legion_region_header
  %i[task_id relationship_id trigger_namespace_id trigger_function_id parent_id master_id runner_namespace runner_class namespace_id function_id function
     chain_id debug].each do |header|
    next unless @options.key? header

    value = @options[header]
    @options[:headers][header] = case value
                                 when Integer, Float, TrueClass, FalseClass
                                   value
                                 else
                                   value.to_s
                                 end
  end
  inject_identity_headers
  @options[:headers]
rescue StandardError => e
  handle_exception(e, level: :error, handled: true, operation: 'transport.message.headers')
  {}
end

#message ⇒ Object 



104
105
106
 
# File 'lib/legion/transport/message.rb', line 104

def message
  @options.except(*ENVELOPE_KEYS)
end

#message_id ⇒ Object 



63
64
65
 
# File 'lib/legion/transport/message.rb', line 63

def message_id
  @options[:message_id] || @options[:task_id]
end

#persistent ⇒ Object 



82
83
84
 
# File 'lib/legion/transport/message.rb', line 82

def persistent
  @options[:persistent] || Legion::Transport.settings[:messages][:persistent]
end

#priority ⇒ Object 



175
176
177
 
# File 'lib/legion/transport/message.rb', line 175

def priority
  @options[:priority] || Legion::Transport.settings[:messages][:priority] || 0
end

#publish(options = @options) ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/legion/transport/message.rb', line 26

def publish(options = @options)
  raise unless @valid

  validate_payload_size
  ex_class = exchange
  exchange_dest = if ex_class.respond_to?(:cached_instance)
                    ex_class.cached_instance || ex_class.new
                  elsif ex_class.respond_to?(:new)
                    ex_class.new
                  else
                    ex_class
                  end
  exchange_dest.publish(encode_message,
                        routing_key:      routing_key || '',
                        content_type:     options[:content_type] || content_type,
                        content_encoding: options[:content_encoding] || content_encoding,
                        type:             options[:type] || type,
                        priority:         options[:priority] || priority,
                        expiration:       options[:expiration] || expiration,
                        headers:          headers,
                        persistent:       persistent,
                        message_id:       message_id,
                        correlation_id:   correlation_id,
                        app_id:           app_id,
                        timestamp:        timestamp)
  ex_name = exchange_dest.respond_to?(:name) ? exchange_dest.name : exchange_dest.to_s
  log.debug "Published to exchange=#{ex_name} routing_key=#{routing_key || ''} class=#{self.class.name}"
rescue Bunny::ConnectionClosedError, Bunny::ChannelAlreadyClosed, Bunny::ChannelError,
       Bunny::NetworkErrorWrapper, IOError, Timeout::Error => e
  handle_exception(e, level: :warn, handled: true, operation: 'transport.message.publish', spooled: true)
  spool_message(e)
end

#reply_to ⇒ Object 



72
73
74
 
# File 'lib/legion/transport/message.rb', line 72

def reply_to
  @options[:reply_to]
end

#routing_key ⇒ Object 



108
109
110
 
# File 'lib/legion/transport/message.rb', line 108

def routing_key
  @options[:routing_key] if @options.key? :routing_key
end

#timestamp ⇒ Object 



191
192
193
 
# File 'lib/legion/transport/message.rb', line 191

def timestamp
  Time.now.to_i
end

#type ⇒ Object 



187
188
189
 
# File 'lib/legion/transport/message.rb', line 187

def type
  'task'
end

#user_id ⇒ Object

user_id Sender’s identifier. www.rabbitmq.com/extensions.html#validated-user-id



68
69
70
 
# File 'lib/legion/transport/message.rb', line 68

def user_id
  @options[:user_id] || Legion::Transport.settings[:connection][:user]
end

#validate ⇒ Object 



195
196
197
 
# File 'lib/legion/transport/message.rb', line 195

def validate
  @valid = true
end