Package org.acemq.amqp.core

Class StreamReader<T>

java.lang.Object
org.acemq.amqp.core.StreamReader<T>
Type Parameters:
T - payload type
public final class StreamReader<T> extends Object
Says where to start reading a stream, and then starts. 

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

 mq.stream("orders.log", OrderPlaced.class)
         .fromLast(Duration.ofHours(1))
         .consume(message -> dashboard.show(message.payload()));

 mq.stream("orders.log", OrderPlaced.class)
         .fromOffset(resumeFrom)
         .consume(message -> ledger.post(message.payload()));

Nothing happens until consume(org.acemq.amqp.api.MessageHandler<T>). Every other method returns a new reader, so a half-built one can be passed around, kept as a field, or reused for two consumers reading the same stream from different places.

  • Method Details

    • fromFirst

      public StreamReader<T> fromFirst()
      Returns:
      a reader starting at the oldest message the stream still holds. On a stream with a year of history this reads the year, which is sometimes exactly right

    • fromLast

      public StreamReader<T> fromLast()
      Returns:
      a reader starting at the last chunk the stream holds, and everything after it

    • fromNext

      public StreamReader<T> fromNext()
      Returns:
      a reader taking only messages written after it attaches; the default

    • fromOffset

      public StreamReader<T> fromOffset(long offset)
      Parameters:
      offset - the exact position, normally one past the last offset recorded as handled
      Returns:
      a reader resuming there

    • from

      public StreamReader<T> from(Instant timestamp)
      Parameters:
      timestamp - the moment to start from
      Returns:
      a reader starting at the first message written at or after it

    • fromLast

      public StreamReader<T> fromLast(Duration age)
      Parameters:
      age - how far back to begin
      Returns:
      a reader replaying that period and then continuing live

    • from

      public StreamReader<T> from(StreamOffset offset)
      Parameters:
      offset - where to start
      Returns:
      a reader starting there

    • prefetch

      public StreamReader<T> prefetch(int prefetch)
      Parameters:
      prefetch - how many unacknowledged messages the broker may keep in flight
      Returns:
      a reader with that prefetch

    • skipFailures

      public StreamReader<T> skipFailures()
      Carries on past a message whose handler failed, counting it.

      The alternative to stopping, and the one to think about. Nothing else will report the gap; StreamConsumer.skipped() is the only record that it happened.

      Returns:
      a reader that skips failures

    • as

      public StreamReader<T> as(Codec codec)
      Parameters:
      codec - the format to read with, for formats whose bytes describe nothing
      Returns:
      a reader using that codec

    • consume

      public StreamConsumer consume(MessageHandler<T> handler)
      Starts reading.
      Parameters:
      handler - called for each message in offset order; returning normally advances the position, throwing applies the failure policy
      Returns:
      the running consumer, closed automatically when the connection closes

    • toString

      public String toString()
      Overrides:
      toString in class Object