Class: AceMQ::AMQP::Patterns::RoutingSlip

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

Overview

Where a message is going, and where it has been.

slip = Patterns::RoutingSlip.new
                          .step("orders-events", "order.validate", name: "validate")
                          .step("orders-events", "order.charge", name: "charge")
                          .step("orders-events", "order.ship", name: "ship")

slip.start(mq, order)

What it costs: no single place says what the whole route is at runtime, so a route that is wrong is discovered one hop at a time. Worth it when the steps vary per message, and not worth it when every message goes the same way — a fixed chain of consumers is simpler and easier to follow.

Constant Summary collapse

SLIP =

The two shapes an itinerary travels in.

SLIP is the JSON in acemq-routing-slip, which names its own destinations and needs nothing else to be read. ROUTE is Java's: x-acemq-route carrying the ordered step names of a Pipeline and nothing more, resolved by the consumer against a pipeline it has declared.

Ruby writes SLIP unless told otherwise — three of the five libraries write it and it is the self-describing one — and reads either. What ROUTE buys is that a Ruby step can stand in a pipeline a Java service declared, and hand the message on in a shape the next Java step can read.

:slip
ROUTE = 
:route

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(steps: [], done: [], form: SLIP, pipeline: nil, run_id: nil) ⇒ RoutingSlip

Returns a new instance of RoutingSlip.

Parameters:

  • steps (Array<Step>) (defaults to: []) —

    the stops still to come

  • done (Array<Step>) (defaults to: []) —

    the stops already made

  • form (Symbol) (defaults to: SLIP) —

    SLIP or ROUTE; how this one goes on the wire

  • pipeline (Pipeline, nil) (defaults to: nil) —

    which pipeline a ROUTE slip belongs to, since the header does not carry its own name

  • run_id (String, nil) (defaults to: nil) —

    identifies one run, across every hop



102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 102

def initialize(steps: [], done: [], form: SLIP, pipeline: nil, run_id: nil)
  @steps = steps
  @done = done
  @form = form
  @pipeline = pipeline
  # Minted the moment a slip becomes a route, and carried by every copy
  # after that, so the identifier is the same at every hop. Minting it
  # on the way out instead would give each hop a different one, which is
  # the opposite of what the header is for: it survives a dead-letter
  # and a replay so that one run can be followed across both.
  @run_id = form == ROUTE ? (run_id || SecureRandom.uuid) : run_id
end

Instance Attribute Details

#done ⇒ Object (readonly)

Returns the value of attribute done.



94
95
96
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 94

def done
  @done
end

#form ⇒ Object (readonly)

Returns the value of attribute form.



94
95
96
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 94

def form
  @form
end

#pipeline ⇒ Object (readonly)

Returns the value of attribute pipeline.



94
95
96
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 94

def pipeline
  @pipeline
end

#run_id ⇒ Object (readonly)

Returns the value of attribute run_id.



94
95
96
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 94

def run_id
  @run_id
end

#steps ⇒ Object (readonly)

Returns the value of attribute steps.



94
95
96
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 94

def steps
  @steps
end

Class Method Details

.from(envelope, pipeline: nil) ⇒ RoutingSlip?

Reads the itinerary off a message, in whichever form it is written.

The JSON slip is looked for first. On a message carrying both — which nothing in this family writes, but a gateway between two of them might — it wins, because it names its own destinations and is therefore readable whether or not this consumer happens to have declared the right pipeline.

Parameters:

  • envelope (Envelope)
  • pipeline (Pipeline, nil) (defaults to: nil) —

    the pipeline an x-acemq-route names the steps of. Without one, the steps and the position are still read, and there is nowhere to send the message onwards to.

Returns:

  • (RoutingSlip, nil) —

    nil when the message carries neither

Raises:

  • (FatalError) —

    when there is a slip and it cannot be read. Fatal rather than retryable: a slip that will not parse will not parse next time either, and a message going round the broker while nothing can tell where it is meant to go is the worst of both.



251
252
253
254
255
256
257
258
259
260
261
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 251

def self.from(envelope, pipeline: nil)
  raw = envelope.headers[SLIP_HEADER]
  return from_route(envelope, pipeline) if raw.nil?

  parsed = JSON.parse(raw.to_s)
  new(steps: Array(parsed["steps"]).map { |s| Step.from_wire(s) },
      done: Array(parsed["done"]).map { |s| Step.from_wire(s) })
rescue JSON::ParserError, TypeError, NoMethodError => e
  raise FatalError,
        "cannot read the routing slip on message #{envelope.id}: #{e.message}"
end

Instance Method Details

#advance ⇒ Object

A copy with the first step moved to done, stamped with the time.

done is carried rather than dropped so a slip that fails half way says how far it got. Whoever finds the message in a dead-letter queue is asking exactly that question.



139
140
141
142
143
144
145
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 139

def advance
  return self if finished?

  completed = @steps.first.dup
  completed.completed_at = Time.now.utc.strftime("%Y-%m-%dT%H:%M:%SZ")
  copy(steps: @steps.drop(1), done: @done + [completed])
end

#all ⇒ Object

The whole route, done and undone, in order.



151
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 151

def all = @done + @steps

#as(form, pipeline: @pipeline) ⇒ Object

A copy that will go on the wire in the other form.

A slip changes shape only when somebody asks it to. Left alone it keeps the shape it arrived in, which is what lets a Ruby step sit in the middle of a Java-declared pipeline without the run turning into something the next Java step cannot read.

Becoming a ROUTE re-resolves every step against the pipeline rather than keeping the destinations the slip was carrying. It has to: the header will carry only the names, so the exchange the next hop is published to has to be the one the pipeline declared, or the message goes somewhere nothing is listening and the run stops silently.

Parameters:

  • form (Symbol) —

    SLIP or ROUTE

  • pipeline (Pipeline, nil) (defaults to: @pipeline) —

    required to become a ROUTE, because the header carries step names and no exchange

Raises:

  • (ArgumentError) —

    with no pipeline, or when the pipeline has no step by one of the slip's names



171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 171

def as(form, pipeline: @pipeline)
  return copy(form: form) unless form == ROUTE

  if pipeline.nil?
    raise ArgumentError,
          "a routing slip written as x-acemq-route needs the pipeline its steps " \
          "belong to; the header carries their names and nothing else"
  end
  copy(form: form, pipeline: pipeline,
       steps: @steps.map { |step| pipeline.step_named(step.name) },
       done: @done.map { |step| pipeline.step_named(step.name) })
end

#finished? ⇒ Boolean

Whether every step has been done.

Returns:

  • (Boolean) 


132
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 132

def finished? = @steps.empty?

#next_step ⇒ Object

The stop this message is going to, or nil at the end of the route.



129
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 129

def next_step = @steps.first

#position ⇒ Object

Which step is next, counting from zero — Java's route-position.



148
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 148

def position = @done.size

#start(connection, payload, **fields) ⇒ Envelope

Sends a payload to the first stop.

Parameters:

  • connection (Connection)
  • payload (Object) —

    anything the codec will encode

  • fields (Hash) —

    envelope fields

Returns:

Raises:

  • (ArgumentError) 


190
191
192
193
194
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 190

def start(connection, payload, **fields)
  raise ArgumentError, "this routing slip has no steps in it" if finished?

  Patterns.send_to(connection, next_step, self, payload, fields)
end

#step(exchange, routing_key, name: nil) ⇒ Object

Adds a stop, and returns self.

Building mutates, in the one place this library allows it, for the same reason Topology does: a slip is assembled and then sent, and threading a new copy through four chained calls would buy an immutability nobody is using. #advance is the other half and does return a copy, because by then the slip is on a message and a message that changed under a handler is a message nothing can reason about.



123
124
125
126
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 123

def step(exchange, routing_key, name: nil)
  @steps << Step.new(exchange: exchange.to_s, routing_key: routing_key.to_s, name: name)
  self
end

#to_header ⇒ Object

The slip as it goes on the wire.



197
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 197

def to_header = JSON.generate(to_wire)

#to_headers ⇒ Hash

The application header a SLIP travels in, or nothing for a ROUTE.

Returns:

  • (Hash) 


209
210
211
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 209

def to_headers
  @form == ROUTE ? {} : { SLIP_HEADER => to_header }
end

#to_route ⇒ Hash

The reserved headers a ROUTE travels in, or nothing for a SLIP.

Kept apart from #to_headers because the two halves go to different places: an application header is passed to publish as headers:, and these are x-acemq- names, which the envelope owns and an application is refused.

Returns:

  • (Hash) 


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

def to_route
  return {} unless @form == ROUTE

  { Headers::ROUTE => all.map(&:name).join(","),
    Headers::ROUTE_POSITION => position,
    Headers::ROUTE_ID => @run_id }
end

#to_s ⇒ Object 



229
230
231
232
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 229

def to_s
  shape = @form == ROUTE ? "#{@pipeline&.name} " : ""
  "RoutingSlip[#{shape}done: #{@done.join(" -> ")} | next: #{@steps.join(" -> ")}]"
end

#to_wire ⇒ 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. 



200
201
202
203
204
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 200

def to_wire
  wire = { "steps" => @steps.map(&:to_wire) }
  wire["done"] = @done.map(&:to_wire) unless @done.empty?
  wire
end