Class: AceMQ::AMQP::Envelope

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/envelope.rb

Overview

What travels with a message, besides the message.

Frozen once built, because an envelope describes a message that has already been published or received: changing one in place would change what a log line said after it was written. #with returns a modified copy for the cases where a new message derives from an old one.

Defaults are applied when the envelope is built rather than when it is read, so two libraries reading the same message agree without having to agree on a second set of rules: correlation_id falls back to id, an attempt starts at 1, a version starts at 1, and type falls back to the routing key, and to DEFAULT_TYPE when that is empty too.

Constant Summary collapse

DEFAULT_TYPE =

The type of a message published with neither a type nor a routing key, as through a fanout exchange. The word the other four libraries write.

"message"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(id: SecureRandom.uuid, type: "", version: 1, correlation_id: nil, causation_id: "", attempt: 1, first_seen: Time.now, origin: "", error: "", claim: "", headers: {}, route: {}) ⇒ Envelope

Returns a new instance of Envelope.

Parameters:

  • id (String) (defaults to: SecureRandom.uuid) —

    the message identifier

  • type (String) (defaults to: "") —

    the logical type

  • version (Integer) (defaults to: 1) —

    the schema version

  • correlation_id (String) (defaults to: nil) —

    defaults to the id

  • causation_id (String) (defaults to: "") —

    the message that caused this one

  • attempt (Integer) (defaults to: 1) —

    the delivery attempt, from 1

  • first_seen (Time) (defaults to: Time.now) —

    when it was first published

  • origin (String) (defaults to: "") —

    service@host

  • error (String) (defaults to: "") —

    why it was dead-lettered

  • claim (String) (defaults to: "") —

    where the payload is, when it is stored elsewhere

  • headers (Hash) (defaults to: {}) —

    the application's own headers

  • route (Hash) (defaults to: {}) —

    the x-acemq-route headers, when the message is travelling a declared pipeline. Carried opaquely and passed through every hop: an envelope has to hold them because they are reserved names and so cannot go in headers, and it does not have to know what they mean. Patterns::RoutingSlip is what reads and writes them.

Raises:

  • (ArgumentError) —

    if a reserved name is in headers



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/acemq/amqp/envelope.rb', line 59

def initialize(id: SecureRandom.uuid, type: "", version: 1,
               correlation_id: nil, causation_id: "", attempt: 1,
               first_seen: Time.now, origin: "", error: "", claim: "",
               headers: {}, route: {})
  self.class.refuse_reserved_names(headers)

  @id = id
  @type = type
  @version = [version, 1].max
  @correlation_id = correlation_id.nil? || correlation_id.empty? ? id : correlation_id
  @causation_id = causation_id
  @attempt = [attempt, 1].max
  @first_seen = first_seen
  @origin = origin
  @error = error
  @claim = claim
  @headers = headers.freeze
  @route = route.freeze
  freeze
end

Instance Attribute Details

#attempt ⇒ Object (readonly)

Returns the value of attribute attempt.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def attempt
  @attempt
end

#causation_id ⇒ Object (readonly)

Returns the value of attribute causation_id.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def causation_id
  @causation_id
end

#claim ⇒ Object (readonly)

Returns the value of attribute claim.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def claim
  @claim
end

#correlation_id ⇒ Object (readonly)

Returns the value of attribute correlation_id.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def correlation_id
  @correlation_id
end

#error ⇒ Object (readonly)

Returns the value of attribute error.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def error
  @error
end

#first_seen ⇒ Object (readonly)

Returns the value of attribute first_seen.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def first_seen
  @first_seen
end

#headers ⇒ Object (readonly)

Returns the value of attribute headers.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def headers
  @headers
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def id
  @id
end

#origin ⇒ Object (readonly)

Returns the value of attribute origin.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def origin
  @origin
end

#route ⇒ Object (readonly)

Returns the value of attribute route.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def route
  @route
end

#type ⇒ Object (readonly)

Returns the value of attribute type.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def type
  @type
end

#version ⇒ Object (readonly)

Returns the value of attribute version.



39
40
41
 
# File 'lib/acemq/amqp/envelope.rb', line 39

def version
  @version
end

Class Method Details

.from_headers(raw, routing_key = "") ⇒ Envelope

Reads an envelope off a delivery.

Anything missing takes its default, and anything unreadable takes its default too: a message from a producer that wrote x-acemq-attempt as a string is still a message, and refusing to deliver it would hand the application an outage rather than a message.

Parameters:

  • raw (Hash, nil) —

    the delivery's headers

  • routing_key (String) (defaults to: "") —

    what it arrived on

Returns:



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/acemq/amqp/envelope.rb', line 90

def self.from_headers(raw, routing_key = "")
  raw = (raw || {}).transform_keys(&:to_s)
  application = raw.reject { |name, _| Headers.reserved?(name) }

  id = text(raw[Headers::ID])
  id = SecureRandom.uuid if id.empty?

  new(
    id: id,
    type: presence(text(raw[Headers::TYPE]), presence(routing_key, DEFAULT_TYPE)),
    version: number(raw[Headers::VERSION], 1),
    correlation_id: presence(text(raw[Headers::CORRELATION]), id),
    causation_id: text(raw[Headers::CAUSATION]),
    attempt: number(raw[Headers::ATTEMPT], 1),
    first_seen: from_millis(number(raw[Headers::FIRST_SEEN], millis(Time.now))),
    origin: text(raw[Headers::ORIGIN]),
    error: text(raw[Headers::ERROR]),
    claim: text(raw[Headers::CLAIM]),
    headers: application,
    route: raw.slice(*Headers::ROUTE_HEADERS)
  )
end

.from_millis(value) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



204
205
206
 
# File 'lib/acemq/amqp/envelope.rb', line 204

def self.from_millis(value)
  Time.at(value / 1000.0)
end

.millis(time) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



199
200
201
 
# File 'lib/acemq/amqp/envelope.rb', line 199

def self.millis(time)
  (time.to_f * 1000).to_i
end

.number(value, fallback) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

A header as an integer, or the default when it is not one.



221
222
223
224
225
226
227
228
 
# File 'lib/acemq/amqp/envelope.rb', line 221

def self.number(value, fallback)
  return fallback if value.nil? || value == true || value == false
  return value if value.is_a?(Integer)

  Integer(value.to_s, 10)
rescue ArgumentError, TypeError
  fallback
end

.refuse_reserved_names(headers) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Reserved names in an application's own hash are refused rather than dropped: silently discarding a header somebody set is worse than saying no, and these would otherwise be written twice and read back inconsistently.

Raises:

  • (ArgumentError) 


189
190
191
192
193
194
195
196
 
# File 'lib/acemq/amqp/envelope.rb', line 189

def self.refuse_reserved_names(headers)
  offending = headers.keys.map(&:to_s).select { |name| Headers.reserved?(name) }.sort
  return if offending.empty?

  raise ArgumentError,
        "these header names belong to AceMQ and cannot be set by hand: " \
        "#{offending.join(", ")}"
end

.text(value) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

A header as a string. Some clients put strings on the wire as bytes, and to_s on those gives something no other language will recognise.



212
213
214
215
216
 
# File 'lib/acemq/amqp/envelope.rb', line 212

def self.text(value)
  return "" if value.nil?

  value.to_s
end

Instance Method Details

#==(other) ⇒ Object 



179
180
181
 
# File 'lib/acemq/amqp/envelope.rb', line 179

def ==(other)
  other.is_a?(Envelope) && to_headers == other.to_headers
end

#age ⇒ Float

How long since the message was first published, in seconds.

The basis for giving up on age rather than on attempts, which is the honest limit when a queue has been paused: a message can be on attempt two and four days old.

Returns:

  • (Float) 


175
176
177
 
# File 'lib/acemq/amqp/envelope.rb', line 175

def age
  Time.now - first_seen
end

#to_headers(routing_key = "") ⇒ Hash

The AMQP headers for this envelope.

Parameters:

  • routing_key (String) (defaults to: "") —

    used when no type was given

Returns:

  • (Hash) —

    the reserved headers, plus the application's own



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/acemq/amqp/envelope.rb', line 127

def to_headers(routing_key = "")
  written = {
    Headers::ID => id,
    Headers::TYPE => type_on(routing_key),
    Headers::VERSION => version,
    Headers::CORRELATION => correlation_id.empty? ? id : correlation_id,
    Headers::ATTEMPT => attempt,
    Headers::FIRST_SEEN => self.class.millis(first_seen)
  }

  # Absent rather than empty. A header carrying "" is a header somebody
  # has to write a special case for at the other end.
  { Headers::CAUSATION => causation_id, Headers::ORIGIN => origin,
    Headers::ERROR => error, Headers::CLAIM => claim }.each do |name, value|
    written[name] = value unless value.nil? || value.empty?
  end

  written.merge(route).merge(headers)
end

#type_on(routing_key = "") ⇒ String

The type this envelope goes out with under a routing key: its own, else the key, else DEFAULT_TYPE. The one rule the header and the message.type tag on a publish are both read off.

Parameters:

  • routing_key (String) (defaults to: "")

Returns:

  • (String) 


119
120
121
 
# File 'lib/acemq/amqp/envelope.rb', line 119

def type_on(routing_key = "")
  [type, routing_key.to_s].find { |t| !t.empty? } || DEFAULT_TYPE
end

#with(**changes) ⇒ Envelope

A copy with fields changed.

Parameters:

  • changes (Hash) —

    the fields to replace

Returns:



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/acemq/amqp/envelope.rb', line 151

def with(**changes)
  Envelope.new(
    id: changes.fetch(:id, id),
    type: changes.fetch(:type, type),
    version: changes.fetch(:version, version),
    correlation_id: changes.fetch(:correlation_id, correlation_id),
    causation_id: changes.fetch(:causation_id, causation_id),
    attempt: changes.fetch(:attempt, attempt),
    first_seen: changes.fetch(:first_seen, first_seen),
    origin: changes.fetch(:origin, origin),
    error: changes.fetch(:error, error),
    claim: changes.fetch(:claim, claim),
    headers: changes.fetch(:headers, headers),
    route: changes.fetch(:route, route)
  )
end