Package org.acemq.amqp.core

Interface StreamConsumer

All Superinterfaces:
AutoCloseable
public interface StreamConsumer extends AutoCloseable
A running read of a stream.

Separate from MessageConsumer on purpose, because a stream is not a queue with different settings — it is a different model, and a single type covering both would be a type where half the methods throw.

What a stream does not have:

  • No dead-letter queue. The retry ladder does not apply. A stream never moves a message anywhere, because it never removes one.
  • No requeue and no selective reject. There is nothing to put back.
  • No destructive read. Acknowledging advances this consumer's position. Every other consumer still sees the message, and so will this one if it starts again from an earlier offset.

So a handler that fails has only two honest outcomes, and StreamOptions makes the choice explicit: stop, or skip and count. There is no third option that quietly works.

  • Method Details

    • queue

      String queue()
      Returns:
      the stream being read

    • isRunning

      boolean isRunning()
      Returns:
      whether this consumer is still attached

    • lastHandledOffset

      OptionalLong lastHandledOffset()
      Returns:
      the offset of the last message handled successfully, or empty if none has been. This is the number to persist if the consumer must resume where it left off; resume from StreamOffset.at(long) with one more than this

    • handled

      long handled()
      Returns:
      how many messages this consumer has handled successfully

    • failed

      long failed()
      Returns:
      how many handlers have failed

    • skipped

      long skipped()
      Returns:
      how many messages were passed over after a failure, which is only ever non-zero when the failure policy is to skip. Data in a stream that nothing processed is worth alerting on

    • stoppedBy

      Optional<Throwable> stoppedBy()
      Returns:
      why this consumer stopped, when it stopped because a handler failed and the policy was to stop

    • close

      void close()
      Specified by:
      close in interface AutoCloseable