Streams

A stream is an append-only log the broker keeps until retention removes it. Reading one does not empty it: every consumer holds its own position, and the same message can be read again tomorrow, by someone else, from the beginning.

That sounds like a queue with better retention. It is not, and the differences are silent rather than loud.

mq.declareStream("orders.log", Duration.ofDays(7), 10L * 1024 * 1024 * 1024);

mq.stream("orders.log", OrderPlaced.class)
        .fromFirst()
        .consume(message -> projection.apply(message.payload()));

Both retention limits may be null, which is legal, logs a warning, and is almost always a mistake — a stream with no limit grows until the disk is full, and a full disk is a broker-wide alarm that blocks every publisher on the node.

How the age is spelled

The age goes to the broker as x-max-age in the largest unit that states it exactly: whole days as 7D, else whole hours as 1h, else whole minutes as 90m, else whole seconds as 90s — the spelling Go, Python and Ruby use, so the same stream can be declared from any of them. Releases before 0.7.13 wrote seconds, 3600s for an hour.

That matters for a stream that already exists. The broker compares the argument as a string (confirmed on RabbitMQ 3.13), so a stream first declared by an earlier Java release keeps 3600s, and declaring it again from 0.7.13 is refused with PRECONDITION_FAILED — with an age, because the spelling differs, and without one, because the broker also refuses a declaration that leaves out an argument the stream has. Delete the stream and declare it again to pick up the new spelling.

Segment size

A stream is stored as a sequence of segment files, and retention happens a whole segment at a time: nothing is discarded until an entire segment can be. That makes the segment size the granularity of every other retention setting — a stream told to keep an hour, in segments large enough to hold a day, keeps a day.

mq.declareStream("orders.log", Duration.ofDays(7), 10L * 1024 * 1024 * 1024,
        100L * 1024 * 1024);   // 100 MB segments

Absent unless you ask. The broker has its own default and it is tuned for the broker's storage rather than for any particular stream, so nothing is sent unless a size is named. That matters beyond tidiness: a declaration carrying an argument that the first declaration of the same stream did not is a PRECONDITION_FAILED for whoever declares second, so a default invented here would break every stream first declared by a Go, Python or Ruby service. Those three leave it out the same way, and the argument is the same x-stream-max-segment-size-bytes in all four.

Where to start reading

reader.fromFirst();                          // everything still held
reader.fromLast();                           // the last message, then onwards
reader.fromNext();                           // only what arrives from now on
reader.fromOffset(41_337);                   // an exact position
reader.from(Instant.now().minus(1, HOURS));  // by wall-clock time
reader.fromLast(Duration.ofDays(3));

A reader never told where to start reads from fromNext(). That is right for a new consumer joining a live system and wrong for a projection — one built without fromFirst() silently skips its own history and looks healthy while being wrong. State the position rather than inheriting it.

Resuming

The broker does not remember your position. That is what makes streams cheap, and it is what makes checkpointing your job:

try (StreamConsumer consumer = mq.stream("orders.log", OrderPlaced.class)
        .fromOffset(checkpoints.load("projection-a") + 1)
        .consume(message -> projection.apply(message.payload()))) {

    consumer.lastHandledOffset()
            .ifPresent(offset -> checkpoints.save("projection-a", offset));
}

Resume from one more than lastHandledOffset(). Saving the checkpoint in the same transaction as the projection's own writes is what makes the pair exactly once; anywhere else is at-least-once, which is fine if the handler is idempotent.

A lost connection

The client reconnects on its own, and a reader carries on from where it was rather than from where it began. A recovered reader starts at the oldest entry it was given and had not finished, or just after the newest it finished: the same entries a queue would redeliver, and no others. So a fromFirst() reader does not read the stream again from the start, and a fromNext() reader does not skip what was appended while it was away. One that had been given nothing yet still starts where you told it to.

What a stream cannot do

A stream never removes a message, and nearly every failure-handling tool in this library is built on moving one:

So a failing handler has two honest outcomes and you must pick:

mq.stream("orders.log", T.class).fromFirst().consume(handler);                 // stop
mq.stream("orders.log", T.class).fromFirst().skipFailures().consume(handler);  // skip

Stopping is the default because it is recoverable — the message is still there and a consumer restarted at the same offset sees it again after a fix.

skipFailures() deserves thought: nothing else records the gap. consumer.skipped() is the only evidence, and no dead-letter queue holds a copy. Alert on it or it is invisible.

Prefetch

A stream consumer always has one — 100 here unless you say otherwise, never zero. RabbitMQ closes the channel for a stream consumer without a prefetch, because it is the only backpressure a stream has.

mq.stream("orders.log", OrderPlaced.class).fromFirst().prefetch(200).consume(handler);

The number is this library's choice, not the contract

The default is not part of the cross-language contract, and the five libraries deliberately disagree about it:

Nothing is wrong with either number. Prefetch is a trade of memory against throughput — how many undelivered messages a consumer is willing to hold in order to avoid waiting on a round trip for each one — and the right answer depends on payload size and handler speed, both of which are properties of your application rather than of the protocol. A library picking a different default does not make a stream written by one unreadable by another: the offset, the retention arguments and the message on the wire are the contract, and prefetch is a consumer-side setting that never leaves the channel.

So do not read across from another AceMQ service and expect the same number, and do not treat a difference between two languages here as a bug to be filed. If the value matters to you, state it — the same advice as the reading position:

mq.stream("orders.log", OrderPlaced.class)
        .fromFirst()
        .prefetch(50)          // say it rather than inherit it
        .consume(handler);

Large payloads want a smaller number, because prefetch multiplies them in memory. A fast handler on small messages wants a larger one, because the round trip starts to dominate. A prefetch below 1 is refused rather than silently corrected, since the broker would close the channel with a PRECONDITION_FAILED that does not explain itself.

Requirements

Streams need RabbitMQ 3.9 or later over amqp://. Anything else refuses rather than degrades — falling back to a classic queue would lose replay, retention and every consumer's independent position, which are the three reasons to want one.

The in-memory test kit does not implement streams, so stream tests need a real broker.

When to use one

Use a stream when more than one consumer needs the same messages, when history must be re-readable, or when a projection has to be rebuildable from scratch: event sourcing, audit logs, analytics fan-out.

Stay with a queue for work that is done once and finished. Streams have no dead-letter queue, no retry ladder and no competing-consumer semantics, and rebuilding those on top of one is how a simple job becomes a distributed systems project.