why jetstream subscribe routine always receive all message history ?

[icoco]

some key method code bellows:

 a class Broker { 
...
 public JetStream getStream(String subject) {
        if (null != jetStream) {
            return jetStream;
        }
        try {
            Connection nc = getConnect();
            jetStream = nc.jetStream();

            // Create the stream
            NatsJsUtils.createStreamOrUpdateSubjects(nc, getNatsOptions().stream, subject);
            return jetStream;
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error("Failed create JetSteam, error:{}", ex);
        }
        return null;
    }


 public synchronized Dispatcher getDispatcher() throws Exception {
        if (null != dispatcher) return dispatcher;

        dispatcher = getConnect().createDispatcher();
        traceIfNeeds(dispatcher);
        return dispatcher;

    }
... 
public JetStreamSubscription subscribeStream(String subject, Consumer<Message> onMessage) throws Exception {
        Dispatcher disp = getDispatcher();

        MessageHandler handler = (msg) -> {
                 **system.out.print("receive message :" + msg);**
        };

        boolean autoAck = true;

        JetStreamSubscription jetStreamSubscription = getStream(subject).subscribe(subject, disp, handler, autoAck);

        return jetStreamSubscription;

    }
}

test code:

program: 

 broker.subscribeStream(‘bus.bus’,(x)->{});

the subject 'bus.bus' has been publish message before .

then while program run, the message handler always trigger and output '"receive message : xxx'

looks the stream replay . what is the mechanism？

[scottf]

Where is the code for getStream(subject) As far as I know this returns a JetStreamContext. When you subscribe this way you are making a brand new consumer to the stream. As you've configured the subscription, your consumer starts at the first message of the stream.

Please have a look at the top of the readme https://github.com/nats-io/nats.java?tab=readme-ov-file#examples-and-other-documentation there are links to many examples.

[icoco]

anyway, thanks for your reply will check the demo. thanks!

[icoco]

most PUB/SUB framework is quit simple, subscribe meaning listen now and future, maybe replay stream is great idea, but it obviously not common use case.

default do not load history message is more easy to use

[scottf]

You can construct a ConsumerConfiguration that starts at a given sequence or time, for instance

ConsumerConfiguration.builder()
.filterSubject(...)
.deliverPolicy(DeliverPolicy.ByStartSequence)
.startSequence(42)
.build()

You can also use a "durable" consumer. It will track the last message it reads.

ConsumerConfiguration.builder()
.filterSubject(...)
.durable("durable-name)
.build()

The difference is the first you need to track the sequence yourself, the second is that the server tracks it but a the cost of resources, which isn't really a problem unless you have more than thousands of consumers.