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

LEGION_PROTOCOL_VERSION = 
'2.0'
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.



16
17
18
19
 
# File 'lib/legion/transport/message.rb', line 16

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

Class Method Details

.max_payload_bytes ⇒ Object 



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

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 



169
170
171
 
# File 'lib/legion/transport/message.rb', line 169

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

#channel ⇒ Object 



310
311
312
 
# File 'lib/legion/transport/message.rb', line 310

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

#confirm_publish(exchange_dest, options) ⇒ Object 



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/legion/transport/message.rb', line 104

def confirm_publish(exchange_dest, options)
  return :accepted unless options[:publisher_confirm] == true

  confirm_channel = publish_channel(exchange_dest)
  return :accepted unless confirm_channel.respond_to?(:wait_for_confirms)

  timeout = options[:publish_confirm_timeout_ms]
  confirmed = if timeout
                confirm_channel.wait_for_confirms(timeout.to_f / 1000.0)
              else
                confirm_channel.wait_for_confirms
              end
  confirmed == false ? :nacked : :accepted
rescue Timeout::Error => e
  handle_exception(e, level: :warn, handled: true, operation: 'transport.message.confirm_publish')
  :confirm_timeout
end

#content_encoding ⇒ Object 



294
295
296
 
# File 'lib/legion/transport/message.rb', line 294

def content_encoding
  'identity'
end

#content_type ⇒ Object 



290
291
292
 
# File 'lib/legion/transport/message.rb', line 290

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.



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

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

#encode_message ⇒ Object 



222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/legion/transport/message.rb', line 222

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) 


243
244
245
246
247
248
249
250
 
# File 'lib/legion/transport/message.rb', line 243

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 



239
240
241
 
# File 'lib/legion/transport/message.rb', line 239

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

#exchange ⇒ Object 



257
258
259
 
# File 'lib/legion/transport/message.rb', line 257

def exchange
  Kernel.const_get(exchange_name)
end

#exchange_name ⇒ Object 



252
253
254
255
 
# File 'lib/legion/transport/message.rb', line 252

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

#expiration ⇒ Object 



196
197
198
199
200
201
202
203
204
205
206
 
# File 'lib/legion/transport/message.rb', line 196

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 



261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
 
# File 'lib/legion/transport/message.rb', line 261

def headers
  @options[:headers] ||= Concurrent::Hash.new
  @options[:headers]['legion_protocol_version'] ||= LEGION_PROTOCOL_VERSION
  inject_legion_version_header
  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

#install_return_listener(exchange_dest, options, return_state) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/legion/transport/message.rb', line 128

def install_return_listener(exchange_dest, options, return_state)
  return unless options[:mandatory] == true

  return_channel = publish_channel(exchange_dest)
  return unless return_channel.respond_to?(:on_return)

  expected_correlation_id = correlation_id
  expected_message_id = message_id
  return_channel.on_return do |return_info, properties, _content|
    next if properties.respond_to?(:correlation_id) && properties.correlation_id &&
            expected_correlation_id && properties.correlation_id != expected_correlation_id
    next if properties.respond_to?(:message_id) && properties.message_id &&
            expected_message_id && properties.message_id != expected_message_id

    return_state[:returned] = true
    return_state[:reply_code] = return_info.reply_code if return_info.respond_to?(:reply_code)
    return_state[:reply_text] = return_info.reply_text if return_info.respond_to?(:reply_text)
  end
end

#message ⇒ Object 



214
215
216
 
# File 'lib/legion/transport/message.rb', line 214

def message
  @options.except(*ENVELOPE_KEYS)
end

#message_id ⇒ Object 



173
174
175
 
# File 'lib/legion/transport/message.rb', line 173

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

#persistent ⇒ Object 



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

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

#prepare_publisher_confirms(exchange_dest, options) ⇒ Object 



95
96
97
98
99
100
101
102
 
# File 'lib/legion/transport/message.rb', line 95

def prepare_publisher_confirms(exchange_dest, options)
  return unless options[:publisher_confirm] == true

  confirm_channel = publish_channel(exchange_dest)
  return unless confirm_channel.respond_to?(:confirm_select)

  confirm_channel.confirm_select
end

#priority ⇒ Object 



286
287
288
 
# File 'lib/legion/transport/message.rb', line 286

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

#publish(options = nil) ⇒ Object 



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
 
# File 'lib/legion/transport/message.rb', line 28

def publish(options = nil)
  raise unless @valid

  publish_options = options ? @options.merge(options) : @options
  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
  return_state = {}
  install_return_listener(exchange_dest, publish_options, return_state)
  prepare_publisher_confirms(exchange_dest, publish_options)
  exchange_dest.publish(encode_message,
                        **publish_envelope_options(publish_options))
  result = publish_result(exchange_dest, publish_options, return_state)
  return result if return_publish_result?(publish_options)

  nil
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: spool_enabled?(publish_options))
  spool_message(e, publish_options) if spool_enabled?(publish_options)
  publish_failure_result(spool_enabled?(publish_options) ? :spooled : :failed, e, publish_options)
end

#publish_channel(exchange_dest) ⇒ Object 



122
123
124
125
126
 
# File 'lib/legion/transport/message.rb', line 122

def publish_channel(exchange_dest)
  return exchange_dest.channel if exchange_dest.respond_to?(:channel)

  channel
end

#publish_envelope_options(options) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/legion/transport/message.rb', line 58

def publish_envelope_options(options)
  {
    routing_key:      options[: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:       options.key?(:persistent) ? options[:persistent] : persistent,
    message_id:       message_id,
    correlation_id:   correlation_id,
    reply_to:         reply_to,
    app_id:           app_id,
    timestamp:        timestamp
  }.tap do |envelope|
    envelope[:mandatory] = true if options[:mandatory] == true
  end
end

#publish_failure_result(status, error, options = @options) ⇒ Object 



157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/legion/transport/message.rb', line 157

def publish_failure_result(status, error, options = @options)
  {
    status:         status,
    accepted:       false,
    error_class:    error.class.name,
    error:          error.message,
    routing_key:    options[:routing_key] || routing_key || '',
    message_id:     message_id,
    correlation_id: correlation_id
  }
end

#publish_result(exchange_dest, options, return_state) ⇒ Object 



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/legion/transport/message.rb', line 78

def publish_result(exchange_dest, options, return_state)
  confirmed_status = confirm_publish(exchange_dest, options)
  status = return_state[:returned] ? :unroutable : confirmed_status
  ex_name = exchange_dest.respond_to?(:name) ? exchange_dest.name : exchange_dest.to_s
  log.debug "Published to exchange=#{ex_name} routing_key=#{options[:routing_key] || routing_key || ''} class=#{self.class.name}"
  {
    status:            status,
    accepted:          status == :accepted,
    exchange:          ex_name,
    routing_key:       options[:routing_key] || routing_key || '',
    message_id:        message_id,
    return_reply_code: return_state[:reply_code],
    return_reply_text: return_state[:reply_text],
    correlation_id:    correlation_id
  }.compact
end

#reply_to ⇒ Object 



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

def reply_to
  @options[:reply_to]
end

#return_publish_result?(options) ⇒ Boolean

Returns:

  • (Boolean) 


152
153
154
155
 
# File 'lib/legion/transport/message.rb', line 152

def return_publish_result?(options)
  options[:return_result] == true || options[:mandatory] == true || options[:publisher_confirm] == true ||
    options[:spool] == false
end

#routing_key ⇒ Object 



218
219
220
 
# File 'lib/legion/transport/message.rb', line 218

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

#spool_enabled?(options) ⇒ Boolean

Returns:

  • (Boolean) 


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

def spool_enabled?(options)
  options.fetch(:spool, true) != false
end

#timestamp ⇒ Object 



302
303
304
 
# File 'lib/legion/transport/message.rb', line 302

def timestamp
  Time.now.to_i
end

#type ⇒ Object 



298
299
300
 
# File 'lib/legion/transport/message.rb', line 298

def type
  'task'
end

#user_id ⇒ Object

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



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

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

#validate ⇒ Object 



306
307
308
 
# File 'lib/legion/transport/message.rb', line 306

def validate
  @valid = true
end