Package org.acemq.amqp.patterns

Class ClaimCheckCodec

java.lang.Object
org.acemq.amqp.patterns.ClaimCheckCodec
All Implemented Interfaces:
Codec
public final class ClaimCheckCodec extends Object implements Codec
Keeps large payloads off the broker. 

 Codec checked = ClaimCheckCodec.wrapping(new JsonCodec(), store);

 mq.publisher("policies", "document.stored", Document.class).as(checked);
 mq.consume("policies.documents", Document.class, checked, m -> read(m.payload()));

A scanned medical report is tens of megabytes. Putting it on a queue is possible and is a mistake: it fills the broker's memory, it is copied to every bound queue, it makes a dead-letter queue impossible to inspect, and it turns a broker into a filesystem with worse tools. What travels instead is a claim check — the payload goes to a ClaimCheckStore, and the message carries the key.

Only when it is worth it

Below the 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

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.

  • Field Details

    • DEFAULT_THRESHOLD

      public static final int 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.

      See Also:

  • Method Details

    • wrapping

      public static ClaimCheckCodec wrapping(Codec delegate, ClaimCheckStore store)
      Parameters:
      delegate - the codec that turns objects into bytes
      store - where large payloads go
      Returns:
      a codec offloading anything above DEFAULT_THRESHOLD

    • wrapping

      public static ClaimCheckCodec wrapping(Codec delegate, ClaimCheckStore store, int threshold)
      Parameters:
      delegate - the codec that turns objects into bytes
      store - where large payloads go
      threshold - payloads of at least this many bytes are offloaded; zero offloads everything, which is occasionally what a store-backed audit trail wants
      Returns:
      a codec offloading anything at or above the threshold

    • contentType

      public String contentType()
      Specified by:
      contentType in interface Codec
      Returns:
      the content type this codec produces, for example application/json

    • encode

      public byte[] encode(Object payload)
      Specified by:
      encode in interface Codec
      Parameters:
      payload - object to encode; never null
      Returns:
      encoded bytes

    • decode

      public <T> T decode(byte[] body, Class<T> target)
      Specified by:
      decode in interface Codec
      Type Parameters:
      T - target type
      Parameters:
      body - encoded bytes
      target - type to produce
      Returns:
      the decoded payload

    • canDecode

      public boolean canDecode(@Nullable String contentType)
      Specified by:
      canDecode in interface Codec
      Parameters:
      contentType - content type of a received message, or null when the sender did not set one
      Returns:
      whether this codec can decode it

    • keyOf

      public static @Nullable String keyOf(byte[] body)
      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 - a message body
      Returns:
      the key, or null when the payload travelled inline or this codec did not write the message

    • delegate

      public Codec delegate()
      Returns:
      the wrapped codec, whose output is what gets stored or inlined

    • toString

      public String toString()
      Overrides:
      toString in class Object