Streams

A stream keeps its messages after they are read. That is the whole difference from a queue, and everything else follows from it: two readers can be at different places in the same data, a new reader can start from the beginning, and nothing is removed by being consumed.

await mq.DeclareStreamAsync("events", TimeSpan.FromDays(7), 10_000_000_000);

using var reader = await mq.Stream<OrderPlaced>("events")
    .FromFirst()
    .Prefetch(100)
    .ConsumeAsync(async message => await _projection.ApplyAsync(message.Payload));

Declare with a limit

await mq.DeclareStreamAsync(name, maxAge: TimeSpan.FromDays(7), maxLengthBytes: 10_000_000_000);

A stream keeps everything written to it until one of these removes it. Declaring one without a limit is declaring a queue that grows until the disk is full, and a full disk stops the whole broker, not just this stream. Both arguments accept null, and both being null is almost always a mistake.

maxAge is written as x-max-age in the largest unit that holds it exactly — 1h, 90m, 7D, and seconds only when nothing larger fits — the same spelling Java, Go, Python and Ruby use. The broker compares the argument as a string, so a stream declared from here and from another language has to agree on it. Up to 0.7.9 this library wrote whole seconds (3600s for an hour). A stream that already exists with that spelling keeps it — the broker never rewrites an argument — so redeclaring it with DeclareStreamAsync is refused with PRECONDITION_FAILED - inequivalent arg 'x-max-age'. Until it can be deleted and recreated, declare it through DeclareQueueAsync with StreamArguments.MaxAge set to the old string.

Segment size

await mq.DeclareStreamAsync(name,
    maxAge: TimeSpan.FromDays(7), maxLengthBytes: 10_000_000_000, segmentBytes: 500_000_000);

segmentBytes is how large each of the stream's files on disk gets, and it sets x-stream-max-segment-size-bytes. Retention happens a whole segment at a time, so a very large segment makes retention coarse: nothing is discarded until an entire segment can be, and a stream bounded at 1 GB with 500 MB segments keeps rather more than 1 GB.

There is no default, on purpose. The broker has one, it is the right one nearly always, and a library that picked its own would make a stream declared from C# quietly different from the same stream declared from Go, Python or Ruby — where the option is opt-in too. The difference would not be quiet for long: a queue redeclared with a different argument is refused rather than adjusted, so a stream this library had silently stamped a segment size onto could not be redeclared from anywhere else.

The three argument names are on StreamArguments (MaxAge, MaxLengthBytes, SegmentBytes) for a caller declaring through Topology rather than through DeclareStreamAsync.

Where to start reading

.FromFirst()                          // everything still held
.FromLast()                           // the last message written, and onward
.FromNext()                           // only what is written from now on
.FromOffset(12345)                    // an exact offset
.FromTime(DateTimeOffset.UtcNow.AddHours(-1))
.FromLast(TimeSpan.FromHours(1))      // however far back an hour reaches

The offset is a consumer setting, not a queue setting — two readers of the same stream sit at different places, so it cannot belong to the queue.

FromNext() is the default, because it is the only one that does not replay history the first time a reader starts. A reader that defaults to FromFirst() reprocesses everything the moment it is deployed, which is a surprise the first time and an outage if the stream is large.

Prefetch is required

Streams need one, and the default here is 100. A stream will otherwise hand over its entire history as fast as the network allows.

What it is for

An audit log, an event log a new consumer needs to catch up on, or anything where "what happened" matters after it has been handled.

It is the wrong shape for distributing work. Consuming does not remove anything, so two workers reading the same stream both do the same job. Work distribution wants a queue.

When a handler fails

The reader stops at the entry it could not handle. IsActive turns false, Failed counts it, and AceMqDiagnostics.StreamReaderStopped is reported with the offset. Nothing is retried and nothing is moved, because a stream has nowhere to retry to: republishing would append the entry to the log a second time, and every other reader would then read it twice. The entry is still in the stream — fix the handler and start a reader from that offset.

SkipFailures() carries on past the entry instead, with a copy in {stream}.dlq. That is a projection that has a gap in it, which is sometimes what you want and is never what you want by accident.

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.

An entry the handler was part-way through when the connection went is handed to it again, as it would be from a queue. That is within one process; across a restart, see below.

Offsets are not stored for you

A reader starts where you tell it to, every time it starts. Nothing is remembered between runs, so a restarting reader with FromFirst() reprocesses everything and one with FromNext() misses whatever arrived while it was down. The Java library behaves the same way; this is a property of the design, not a gap in the port.

To resume exactly where a previous run stopped, record the offset as you go and start from it:

using var reader = await mq.Stream<OrderPlaced>("events")
    .FromOffset(await _checkpoints.LastAsync("projection"))
    .ConsumeAsync(async message =>
    {
        await _projection.ApplyAsync(message.Payload);
        await _checkpoints.SaveAsync("projection", message.Headers["x-stream-offset"]);
    });

The consumer also reports where it has got to:

reader.LastHandledOffset   // null until the first message is handled
reader.Handled
reader.Failed
reader.Skipped

Saving the checkpoint after handling means a crash between the two replays the last message. Saving it before means a crash loses one. Neither is avoidable without a transaction spanning the broker and your store, so pick the one your handler can tolerate — and if it can tolerate neither, make the handler idempotent on the envelope's id.