Class: AceMQ::AMQP::Patterns::ClaimCheckCodec

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

Overview

A codec that puts a large payload aside and sends the key instead.

store = Patterns::FilesystemClaimCheckStore.new("/mnt/claims")
checked = Patterns::ClaimCheckCodec.wrapping(JSONCodec.new, store)

mq = Connection.open(url: url, codec: checked)
mq.publish(document, to: "document.stored")

Only when it is worth it

Below DEFAULT_THRESHOLD the payload travels inline, exactly as it would without this codec. That matters more than it sounds: offloading a two-hundred-byte message turns one broker round trip into a store round trip and a broker round trip, so an unconditional claim check makes the common case slower to fix the rare one.

The framing therefore says which of the two it is, and a consumer handles both without being told. That is what allows the threshold to be changed, or this codec to be introduced, without a flag day: messages written before the change are still readable after it.

What is on the wire

0xAC  0x01  0x00  payload   inline, and identical to what the delegate wrote
0xAC  0x01  0x01  key       a claim check

Three bytes, and the third is how a consumer decides whether it is holding a payload or a reference to one. Nothing else distinguishes them, which is why the framing rather than a header is the contract: a header can be stripped by a shovel or a federation link, and the body cannot.

These are the same three bytes the Java, and any other AceMQ, claim check writes, and the key is the store's key as bare UTF-8 — not a URI, not a scheme, nothing wrapped around it. A Ruby consumer pointed at the same store therefore reads a document a Java publisher checked in, and the other way round.

The content type is the delegate's, unchanged — unlike encryption, where the bytes really are something else. A claim-checked message is still a document; it is a document that is somewhere else, and a consumer that lacks the store gets a clear failure rather than a parser error.

Constant Summary collapse

MAGIC =

Marks this codec's framing.

0xAC
VERSION =

The framing version. One byte, so a later shape can be introduced without every consumer having to be redeployed the same afternoon.

0x01
INLINE =

The payload is in the message.

0x00
CHECKED =

The message carries a key, and the payload is in the store.

0x01
HEADER =

How many bytes the framing takes.

3
DEFAULT_THRESHOLD =

Below this, payloads travel inline.

64 KiB: comfortably above an ordinary event and comfortably below the size at which a broker starts to care. RabbitMQ will accept far larger, which is the problem — nothing refuses a 40 MB message, it simply makes everything worse afterwards.

64 * 1024

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(delegate, store, threshold: DEFAULT_THRESHOLD) ⇒ ClaimCheckCodec

Returns a new instance of ClaimCheckCodec.

See Also:



309
310
311
312
313
314
315
316
317
318
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 309

def initialize(delegate, store, threshold: DEFAULT_THRESHOLD)
  @delegate = Codec.check!(delegate)
  @store = store
  unless threshold.is_a?(Integer) && !threshold.negative?
    raise ArgumentError, "a threshold cannot be negative, was #{threshold.inspect}"
  end

  @threshold = threshold
  freeze
end

Instance Attribute Details

#delegate ⇒ Object (readonly)

The wrapped codec, whose output is what gets stored or inlined.



321
322
323
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 321

def delegate
  @delegate
end

#threshold ⇒ Object (readonly)

Bytes at or above which a payload is offloaded.



324
325
326
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 324

def threshold
  @threshold
end

Class Method Details

.framed?(body) ⇒ Boolean

Whether a body carries this codec's framing.

Parameters:

  • body (String)

Returns:

  • (Boolean) 


390
391
392
393
394
395
396
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 390

def self.framed?(body)
  body = body.to_s
  return false if body.bytesize < HEADER

  body.getbyte(0) == MAGIC && body.getbyte(1) == VERSION &&
    [INLINE, CHECKED].include?(body.getbyte(2))
end

.key_of(body) ⇒ String?

Reads the key a message refers to, without fetching it.

For the operator looking at a dead-letter queue: which object does this need, and is it still in the store? Answering that from the message alone is the difference between a five-minute check and restoring a backup.

Parameters:

  • body (String) —

    a message body

Returns:

  • (String, nil) —

    the key, or nil when the payload travelled inline or this codec did not write the message



379
380
381
382
383
384
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 379

def self.key_of(body)
  body = body.to_s
  return nil unless framed?(body) && body.getbyte(2) == CHECKED

  body.byteslice(HEADER, body.bytesize - HEADER).force_encoding(Encoding::UTF_8)
end

.wrapping(delegate, store, threshold: DEFAULT_THRESHOLD) ⇒ ClaimCheckCodec

Parameters:

  • delegate (#encode) —

    the codec that turns payloads into bytes

  • store (ClaimCheckStore) —

    where large payloads go

  • threshold (Integer) (defaults to: DEFAULT_THRESHOLD) —

    payloads of at least this many bytes are offloaded; zero offloads everything, which is occasionally what a store-backed audit trail wants

Returns:



304
305
306
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 304

def self.wrapping(delegate, store, threshold: DEFAULT_THRESHOLD)
  new(delegate, store, threshold: threshold)
end

Instance Method Details

#can_decode?(content_type) ⇒ Boolean

Whatever the delegate accepts. A claim check does not change what the message is.

Returns:

  • (Boolean) 


367
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 367

def can_decode?(content_type) = @delegate.can_decode?(content_type)

#content_type ⇒ Object

The delegate's. A claim-checked document is still a document.



327
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 327

def content_type = @delegate.content_type

#decode(body, content_type = "") ⇒ Object

Decodes a message, fetching the payload when it was offloaded.

Parameters:

  • body (String) —

    the bytes as they arrived

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

    what the sender said, when the delegate wants to know

Returns:

  • (Object) —

    the payload

Raises:

  • (DecodeError) —

    when the store no longer holds the payload



351
352
353
354
355
356
357
358
359
360
361
362
363
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 351

def decode(body, content_type = "")
  body = body.to_s
  # Written before this codec was introduced, or by a publisher that
  # does not use it. Reading it as the delegate would is the only useful
  # answer, and it is what makes adding a claim check to a live queue
  # safe.
  return delegate_decode(body, content_type) unless self.class.framed?(body)

  rest = body.byteslice(HEADER, body.bytesize - HEADER)
  rest = fetch(rest.force_encoding(Encoding::UTF_8)) if body.getbyte(2) == CHECKED

  delegate_decode(restore_encoding(rest), content_type)
end

#encode(payload) ⇒ String

Encodes a payload, offloading it when it is large enough to be worth it.

Parameters:

  • payload (Object)

Returns:

  • (String) —

    the framed bytes to publish



334
335
336
337
338
339
340
341
342
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 334

def encode(payload)
  encoded = @delegate.encode(payload).to_s
  # bytesize rather than length. A string of 40,000 characters with an
  # accent in it is more than 40,000 bytes on the wire, and the
  # threshold is a statement about what the broker has to carry.
  return frame(INLINE, encoded) if encoded.bytesize < @threshold

  frame(CHECKED, @store.put(encoded).to_s)
end

#to_s ⇒ Object 



398
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 398

def to_s = "ClaimCheckCodec(#{@delegate.class.name}, above #{@threshold} bytes)"