#2042

Committed 04 Jul 2025 01:42PM UTC coverage: 95.608% (-0.003%) from 95.611%

Build # #2042

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1348 from nats-io/fix-what-i-broke-2-21-3

Fix heartbeat timer handling broken when replacing timer with scheduler.

Run Details

44 of 50 new or added lines in 7 files covered. (88.0%)

6 existing lines in 4 files now uncovered.

11798 of 12340 relevant lines covered (95.61%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.73

/src/main/java/io/nats/client/impl/NatsIterableConsumer.java

// Copyright 2020-2023 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at:
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package io.nats.client.impl;

import io.nats.client.*;
import io.nats.client.api.ConsumerInfo;

import java.io.IOException;
import java.time.Duration;

class NatsIterableConsumer extends NatsMessageConsumer implements IterableConsumer {

    NatsIterableConsumer(SimplifiedSubscriptionMaker subscriptionMaker, ConsumerInfo cachedConsumerInfo, ConsumeOptions opts) throws IOException, JetStreamApiException {
        super(subscriptionMaker, cachedConsumerInfo, opts, null, null);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Message nextMessage(Duration timeout) throws InterruptedException, JetStreamStatusCheckedException {
        try {
            Message msg = sub.nextMessage(timeout);
            if (msg != null && stopped.get() && pmm.noMorePending()) {
                finished.set(true);
            }
            return msg;
        }
        catch (JetStreamStatusException e) {
            throw new JetStreamStatusCheckedException(e);
        }
        catch (IllegalStateException i) {
            // this happens if the consumer is stopped, since it is
            // drained/unsubscribed, so don't pass it on if it's expected
            return null;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Message nextMessage(long timeoutMillis) throws InterruptedException, JetStreamStatusCheckedException {
        return nextMessage(Duration.ofMillis(timeoutMillis));
    }
}

1	// Copyright 2020-2023 The NATS Authors
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the License.
4	// You may obtain a copy of the License at:
5	//
6	// http://www.apache.org/licenses/LICENSE-2.0
7	//
8	// Unless required by applicable law or agreed to in writing, software
9	// distributed under the License is distributed on an "AS IS" BASIS,
10	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11	// See the License for the specific language governing permissions and
12	// limitations under the License.
13
14	package io.nats.client.impl;
15
16	import io.nats.client.*;
17	import io.nats.client.api.ConsumerInfo;
18
19	import java.io.IOException;
20	import java.time.Duration;
21
22	class NatsIterableConsumer extends NatsMessageConsumer implements IterableConsumer {
23
24	NatsIterableConsumer(SimplifiedSubscriptionMaker subscriptionMaker, ConsumerInfo cachedConsumerInfo, ConsumeOptions opts) throws IOException, JetStreamApiException {
25	super(subscriptionMaker, cachedConsumerInfo, opts, null, null);	1✔
26	}	1✔
27
28	/**
29	* {@inheritDoc}
30	*/
31	@Override
32	public Message nextMessage(Duration timeout) throws InterruptedException, JetStreamStatusCheckedException {
33	try {
34	Message msg = sub.nextMessage(timeout);	1✔
35	if (msg != null && stopped.get() && pmm.noMorePending()) {	1✔
UNCOV 36	finished.set(true);	×
37	}
38	return msg;	1✔
39	}
40	catch (JetStreamStatusException e) {	×
41	throw new JetStreamStatusCheckedException(e);	×
42	}
43	catch (IllegalStateException i) {	1✔
44	// this happens if the consumer is stopped, since it is
45	// drained/unsubscribed, so don't pass it on if it's expected
46	return null;	1✔
47	}
48	}
49
50	/**
51	* {@inheritDoc}
52	*/
53	@Override
54	public Message nextMessage(long timeoutMillis) throws InterruptedException, JetStreamStatusCheckedException {
55	return nextMessage(Duration.ofMillis(timeoutMillis));	1✔
56	}
57	}

nats-io / nats.java / #2042

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous