36784688561

Committed 30 Sep 2026 10:17PM UTC coverage: 90.863% (+0.006%) from 90.857%

Build # 36784688561

Build Type

push

github

Committed by

scottf

Commit Message

Coveralls failures do not fail module builds

Run 36783609042 failed the service job on a Coveralls reporter checksum mismatch, which also skipped its publish. Both Coveralls steps are continue-on-error, as in v2.

Coverage Stats

4189 of 4830 branches covered (86.73%)

Branch coverage included in aggregate %.

10649 of 11500 relevant lines covered (92.6%)

4.12 hits per line

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

81.98

/jetstream/src/main/java/io/synadia/client/impl/NatsMessageConsumer.java

package io.synadia.client.impl;

import io.synadia.client.Message;
import io.synadia.client.MessageHandler;
import io.synadia.client.api.ConsumerInfo;
import io.synadia.client.api.JetStreamException;


class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManagerObserver {
    protected final ConsumeOptions consumeOpts;
    protected final SimplifiedSubscriptionMaker subscriptionMaker;
    protected final NatsDispatcher userDispatcher;
    protected final MessageHandler userMessageHandler;

    protected final int thresholdMessages;
    protected final long thresholdBytes;
    protected final boolean isTrackingBytes;

    protected int pendingReceivedMessages;
    protected long pendingReceivedBytes;
    protected boolean noReceivedArePending;
    protected boolean forcePull;
    protected int pendingProcessedMessages;
    protected long pendingProcessedBytes;
    protected boolean processedHasCrossedThreshold;

    NatsMessageConsumer(SimplifiedSubscriptionMaker subscriptionMaker,
                        ConsumerInfo cachedConsumerInfo,
                        ConsumeOptions consumeOpts,
                        NatsDispatcher userDispatcher,
                        final MessageHandler userMessageHandler) throws JetStreamException, InterruptedException
    {
        super(cachedConsumerInfo);

        this.subscriptionMaker = subscriptionMaker;
        this.consumeOpts = consumeOpts;
        this.userDispatcher = userDispatcher;
        this.userMessageHandler = userMessageHandler;

        int bm = consumeOpts.getBatchSize();
        long bb = consumeOpts.getBatchBytes();
        int rePullMessages = Math.max(1, bm * consumeOpts.getThresholdPercent() / 100);
        long rePullBytes = bb == 0 ? 0 : Math.max(1, bb * consumeOpts.getThresholdPercent() / 100);
        // The threshold is what the pending count has to drop below for the next pull. A batch of 1
        // gives a re-pull size of 1 and so a threshold of 0, which pending can never go below, so the
        // consumer pulled once and then stopped. Never let the threshold fall under 1.
        int thresholdM = bm - rePullMessages;
        thresholdMessages = thresholdM < 1 ? 1 : thresholdM;
        long thresholdB = bb - rePullBytes;
        thresholdBytes = bb == 0 ? Integer.MIN_VALUE : (thresholdB < 1 ? 1 : thresholdB);
        isTrackingBytes = rePullBytes > 0;
        doSub(true);
    }

    protected void fullResetPending() {
        pendingReceivedMessages = 0;
        pendingReceivedBytes = 0;
        noReceivedArePending = true;
        forcePull = true;
        pendingProcessedMessages = 0;
        pendingProcessedBytes = 0;
        processedHasCrossedThreshold = true;
    }

    protected void statusAdjustPending(int messages, long bytes) {
        pendingReceivedMessages = Math.max(0, pendingReceivedMessages - messages);
        pendingReceivedBytes = Math.max(0, pendingReceivedBytes - bytes);
        noReceivedArePending = pendingReceivedMessages == 0 || (isTrackingBytes && pendingReceivedBytes == 0);
        forcePull = true;
        pendingProcessedMessages = Math.max(0, pendingProcessedMessages - messages);
        pendingProcessedBytes = Math.max(0, pendingProcessedBytes - bytes);
        processedHasCrossedThreshold = pendingProcessedMessages < thresholdMessages || (isTrackingBytes && pendingProcessedBytes < thresholdBytes);
    }

    protected void aboutToPull(int messages, long bytes) {
        pendingReceivedMessages += messages;
        pendingReceivedBytes += bytes;
        noReceivedArePending = false;
        forcePull = false;
        pendingProcessedMessages += messages;
        pendingProcessedBytes += bytes;
        processedHasCrossedThreshold = false;
    }

    protected void updateProcessed(Message msg) {
        pendingProcessedMessages = Math.max(0, pendingProcessedMessages - 1);
        if (pendingProcessedMessages < thresholdMessages) {
            processedHasCrossedThreshold = true;
        }
        if (isTrackingBytes) {
            pendingProcessedBytes = Math.max(0, pendingProcessedBytes - msg.consumeByteCount());
            processedHasCrossedThreshold |= pendingProcessedBytes < thresholdBytes;
        }
        afterPendingUpdated();
    }

    @Override
    public void messageReceived(Message msg) {
        pendingReceivedMessages = Math.max(0, pendingReceivedMessages - 1);
        if (pendingReceivedMessages == 0) {
            noReceivedArePending = true;
        }
        if (isTrackingBytes) {
            pendingReceivedBytes = Math.max(0, pendingReceivedBytes - msg.consumeByteCount());
            noReceivedArePending |= pendingReceivedBytes == 0;
        }
        afterPendingUpdated();
    }

    @Override
    public void pullCompletedWithStatus(int messages, long bytes) {
        if (messages == -1) {
            // status without Nats-Pending-* headers
            fullResetPending();
        }
        else {
            statusAdjustPending(messages, bytes);
        }
        afterPendingUpdated();
    }

    @Override
    public void pullTerminatedByError() {
        if (stopped.get()) {
            fullClose();
        }
        else {
            try {
                shutdownSub();
                doSub(false);
            }
            catch (JetStreamException e) {
                resetOnException();
            }
            catch (InterruptedException e) {
                // Runs on a NATS-owned thread (heartbeat timer pool or message delivery) that
                // can't propagate the checked interrupt. Realistically only our own close
                // (shutdownNow) interrupts here, and the stopped/fullClose path handles that
                // teardown; a stray interrupt is transient, so recovering is correct. Restore
                // the flag as standard hygiene (a no-op on a pooled worker), then recover.
                Thread.currentThread().interrupt();
                resetOnException();
            }
        }
    }

    void doSub(boolean first) throws JetStreamException, InterruptedException {
        MessageHandler mh = userMessageHandler == null ? null : msg -> {
            try {
                userMessageHandler.onMessage(msg);
            }
            finally {
                updateProcessed(msg);
            }
        };

        // Whether the subscribe below succeeded, which is what decides if this attempt owns a
        // subscription that has to be cleaned up when something after it fails. It cannot be inferred
        // from the "sub" field: on the reset path that still refers to the previous, already shut down
        // subscription, and shutdownSub() dereferences it without a null check.
        boolean subInitialized = false;
        try {
            stopped.set(false);
            finished.set(false);
            super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null), !first);
            subInitialized = true;
            fullResetPending();
            rePull();
        }
        catch (JetStreamException | RuntimeException e) {
            if (subInitialized) {
                // The "subscribe" succeeded and something after it did not, so this attempt is being
                // abandoned while holding a live subscription. Nothing else will clean it up - on the
                // first == true path the consumer never reaches the caller, so both the subscription
                // and its heartbeat timer would be unreachable and leak.
                shutdownSub();
            }
            // The unchecked case matters as much as the checked one. A connection or dispatcher that is
            // closing or draining throws IllegalStateException out of the subscribe, and on the reset
            // path (first == false) shutdownSub() has already run. Letting that escape leaves the
            // consumer with no subscription and no heartbeat timer while stopped and finished both
            // still report false - it silently stops delivering, permanently. Recovering is correct:
            // resetOnException re-arms the heartbeat, which alarms again and retries the reset.
            if (first) {
                throw e;
            }
            resetOnException();
        }
    }

    private void resetOnException() {
        fullResetPending();
        pmm.updateLastMessageReceived();
        pmm.initOrResetHeartbeatTimer();
    }

    protected void afterPendingUpdated() {
        if (stopped.get()) {
            pmm.shutdownHeartbeatTimer();
            if (noReceivedArePending) {
                fullClose();
            }
        }
        else if (forcePull || processedHasCrossedThreshold) {
            rePull();
        }
    }

    protected void rePull() {
        int rePullMessages = Math.max(1, consumeOpts.getBatchSize() - pendingProcessedMessages);
        long rePullBytes = consumeOpts.getBatchBytes() == 0 ? 0 : consumeOpts.getBatchBytes() - pendingProcessedBytes;
        PinnablePullRequestOptions pro = new PinnablePullRequestOptions(pmm.currentPinId,
            PullRequestOptions.builder(rePullMessages)
                .maxBytes(rePullBytes)
                .expiresIn(consumeOpts.getExpiresInMillis())
                .idleHeartbeat(consumeOpts.getIdleHeartbeat())
                .group(consumeOpts.getGroup())
                .priority(consumeOpts.getPriority())
                .minPending(consumeOpts.getMinPending())
                .minAckPending(consumeOpts.getMinAckPending()));
        aboutToPull(rePullMessages, rePullBytes);
        sub._pull(pro, consumeOpts.raiseStatusWarnings(), this);
    }
}

1	package io.synadia.client.impl;
2
3	import io.synadia.client.Message;
4	import io.synadia.client.MessageHandler;
5	import io.synadia.client.api.ConsumerInfo;
6	import io.synadia.client.api.JetStreamException;
7
8
9	class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManagerObserver {
10	protected final ConsumeOptions consumeOpts;
11	protected final SimplifiedSubscriptionMaker subscriptionMaker;
12	protected final NatsDispatcher userDispatcher;
13	protected final MessageHandler userMessageHandler;
14
15	protected final int thresholdMessages;
16	protected final long thresholdBytes;
17	protected final boolean isTrackingBytes;
18
19	protected int pendingReceivedMessages;
20	protected long pendingReceivedBytes;
21	protected boolean noReceivedArePending;
22	protected boolean forcePull;
23	protected int pendingProcessedMessages;
24	protected long pendingProcessedBytes;
25	protected boolean processedHasCrossedThreshold;
26
27	NatsMessageConsumer(SimplifiedSubscriptionMaker subscriptionMaker,
28	ConsumerInfo cachedConsumerInfo,
29	ConsumeOptions consumeOpts,
30	NatsDispatcher userDispatcher,
31	final MessageHandler userMessageHandler) throws JetStreamException, InterruptedException
32	{
33	super(cachedConsumerInfo);	3 ✔
34
35	this.subscriptionMaker = subscriptionMaker;	3 ✔
36	this.consumeOpts = consumeOpts;	3 ✔
37	this.userDispatcher = userDispatcher;	3 ✔
38	this.userMessageHandler = userMessageHandler;	3 ✔
39
40	int bm = consumeOpts.getBatchSize();	3 ✔
41	long bb = consumeOpts.getBatchBytes();	3 ✔
42	int rePullMessages = Math.max(1, bm * consumeOpts.getThresholdPercent() / 100);	9 ✔
43	long rePullBytes = bb == 0 ? 0 : Math.max(1, bb * consumeOpts.getThresholdPercent() / 100);	7 !
44	// The threshold is what the pending count has to drop below for the next pull. A batch of 1
45	// gives a re-pull size of 1 and so a threshold of 0, which pending can never go below, so the
46	// consumer pulled once and then stopped. Never let the threshold fall under 1.
47	int thresholdM = bm - rePullMessages;	4 ✔
48	thresholdMessages = thresholdM < 1 ? 1 : thresholdM;	8 ✔
49	long thresholdB = bb - rePullBytes;	4 ✔
50	thresholdBytes = bb == 0 ? Integer.MIN_VALUE : (thresholdB < 1 ? 1 : thresholdB);	8 !
51	isTrackingBytes = rePullBytes > 0;	7 !
52	doSub(true);	3 ✔
53	}	1 ✔
54
55	protected void fullResetPending() {
56	pendingReceivedMessages = 0;	3 ✔
57	pendingReceivedBytes = 0;	3 ✔
58	noReceivedArePending = true;	3 ✔
59	forcePull = true;	3 ✔
60	pendingProcessedMessages = 0;	3 ✔
61	pendingProcessedBytes = 0;	3 ✔
62	processedHasCrossedThreshold = true;	3 ✔
63	}	1 ✔
64
65	protected void statusAdjustPending(int messages, long bytes) {
66	pendingReceivedMessages = Math.max(0, pendingReceivedMessages - messages);	8 ✔
67	pendingReceivedBytes = Math.max(0, pendingReceivedBytes - bytes);	8 ✔
68	noReceivedArePending = pendingReceivedMessages == 0 \|\| (isTrackingBytes && pendingReceivedBytes == 0);	11 !
69	forcePull = true;	3 ✔
70	pendingProcessedMessages = Math.max(0, pendingProcessedMessages - messages);	8 ✔
71	pendingProcessedBytes = Math.max(0, pendingProcessedBytes - bytes);	8 ✔
72	processedHasCrossedThreshold = pendingProcessedMessages < thresholdMessages \|\| (isTrackingBytes && pendingProcessedBytes < thresholdBytes);	13 !
73	}	1 ✔
74
75	protected void aboutToPull(int messages, long bytes) {
76	pendingReceivedMessages += messages;	6 ✔
77	pendingReceivedBytes += bytes;	6 ✔
78	noReceivedArePending = false;	3 ✔
79	forcePull = false;	3 ✔
80	pendingProcessedMessages += messages;	6 ✔
81	pendingProcessedBytes += bytes;	6 ✔
82	processedHasCrossedThreshold = false;	3 ✔
83	}	1 ✔
84
85	protected void updateProcessed(Message msg) {
86	pendingProcessedMessages = Math.max(0, pendingProcessedMessages - 1);	8 ✔
87	if (pendingProcessedMessages < thresholdMessages) {	5 ✔
88	processedHasCrossedThreshold = true;	3 ✔
89	}
90	if (isTrackingBytes) {	3 !
91	pendingProcessedBytes = Math.max(0, pendingProcessedBytes - msg.consumeByteCount());	×
92	processedHasCrossedThreshold \|= pendingProcessedBytes < thresholdBytes;	×
93	}
94	afterPendingUpdated();	2 ✔
95	}	1 ✔
96
97	@Override
98	public void messageReceived(Message msg) {
99	pendingReceivedMessages = Math.max(0, pendingReceivedMessages - 1);	8 ✔
100	if (pendingReceivedMessages == 0) {	3 ✔
101	noReceivedArePending = true;	3 ✔
102	}
103	if (isTrackingBytes) {	3 !
104	pendingReceivedBytes = Math.max(0, pendingReceivedBytes - msg.consumeByteCount());	×
105	noReceivedArePending \|= pendingReceivedBytes == 0;	×
106	}
107	afterPendingUpdated();	2 ✔
108	}	1 ✔
109
110	@Override
111	public void pullCompletedWithStatus(int messages, long bytes) {
112	if (messages == -1) {	3 !
113	// status without Nats-Pending-* headers
114	fullResetPending();	×
115	}
116	else {
117	statusAdjustPending(messages, bytes);	4 ✔
118	}
119	afterPendingUpdated();	2 ✔
120	}	1 ✔
121
122	@Override
123	public void pullTerminatedByError() {
124	if (stopped.get()) {	4 !
125	fullClose();	×
126	}
127	else {
128	try {
129	shutdownSub();	2 ✔
130	doSub(false);	3 ✔
131	}
132	catch (JetStreamException e) {	×
133	resetOnException();	×
134	}
135	catch (InterruptedException e) {	×
136	// Runs on a NATS-owned thread (heartbeat timer pool or message delivery) that
137	// can't propagate the checked interrupt. Realistically only our own close
138	// (shutdownNow) interrupts here, and the stopped/fullClose path handles that
139	// teardown; a stray interrupt is transient, so recovering is correct. Restore
140	// the flag as standard hygiene (a no-op on a pooled worker), then recover.
141	Thread.currentThread().interrupt();	×
142	resetOnException();	×
143	}	1 ✔
144	}
145	}	1 ✔
146
147	void doSub(boolean first) throws JetStreamException, InterruptedException {
148	MessageHandler mh = userMessageHandler == null ? null : msg -> {	8 ✔
149	try {
150	userMessageHandler.onMessage(msg);	4 ✔
151	}
152	finally {
153	updateProcessed(msg);	3 ✔
154	}
155	};	1 ✔
156
157	// Whether the subscribe below succeeded, which is what decides if this attempt owns a
158	// subscription that has to be cleaned up when something after it fails. It cannot be inferred
159	// from the "sub" field: on the reset path that still refers to the previous, already shut down
160	// subscription, and shutdownSub() dereferences it without a null check.
161	boolean subInitialized = false;	2 ✔
162	try {
163	stopped.set(false);	4 ✔
164	finished.set(false);	4 ✔
165	super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null), !first);	16 ✔
166	subInitialized = true;	2 ✔
167	fullResetPending();	2 ✔
168	rePull();	2 ✔
169	}
170	catch (JetStreamException \| RuntimeException e) {	1 ✔
171	if (subInitialized) {	2 ✔
172	// The "subscribe" succeeded and something after it did not, so this attempt is being
173	// abandoned while holding a live subscription. Nothing else will clean it up - on the
174	// first == true path the consumer never reaches the caller, so both the subscription
175	// and its heartbeat timer would be unreachable and leak.
176	shutdownSub();	2 ✔
177	}
178	// The unchecked case matters as much as the checked one. A connection or dispatcher that is
179	// closing or draining throws IllegalStateException out of the subscribe, and on the reset
180	// path (first == false) shutdownSub() has already run. Letting that escape leaves the
181	// consumer with no subscription and no heartbeat timer while stopped and finished both
182	// still report false - it silently stops delivering, permanently. Recovering is correct:
183	// resetOnException re-arms the heartbeat, which alarms again and retries the reset.
184	if (first) {	2 ✔
185	throw e;	2 ✔
186	}
187	resetOnException();	2 ✔
188	}	1 ✔
189	}	1 ✔
190
191	private void resetOnException() {
192	fullResetPending();	2 ✔
193	pmm.updateLastMessageReceived();	3 ✔
194	pmm.initOrResetHeartbeatTimer();	3 ✔
195	}	1 ✔
196
197	protected void afterPendingUpdated() {
198	if (stopped.get()) {	4 ✔
199	pmm.shutdownHeartbeatTimer();	3 ✔
200	if (noReceivedArePending) {	3 ✔
201	fullClose();	3 ✔
202	}
203	}
204	else if (forcePull \|\| processedHasCrossedThreshold) {	6 ✔
205	rePull();	2 ✔
206	}
207	}	1 ✔
208
209	protected void rePull() {
210	int rePullMessages = Math.max(1, consumeOpts.getBatchSize() - pendingProcessedMessages);	9 ✔
211	long rePullBytes = consumeOpts.getBatchBytes() == 0 ? 0 : consumeOpts.getBatchBytes() - pendingProcessedBytes;	9 !
212	PinnablePullRequestOptions pro = new PinnablePullRequestOptions(pmm.currentPinId,	6 ✔
213	PullRequestOptions.builder(rePullMessages)	2 ✔
214	.maxBytes(rePullBytes)	3 ✔
215	.expiresIn(consumeOpts.getExpiresInMillis())	4 ✔
216	.idleHeartbeat(consumeOpts.getIdleHeartbeat())	4 ✔
217	.group(consumeOpts.getGroup())	4 ✔
218	.priority(consumeOpts.getPriority())	4 ✔
219	.minPending(consumeOpts.getMinPending())	4 ✔
220	.minAckPending(consumeOpts.getMinAckPending()));	4 ✔
221	aboutToPull(rePullMessages, rePullBytes);	4 ✔
222	sub._pull(pro, consumeOpts.raiseStatusWarnings(), this);	9 ✔
223	}	1 ✔
224	}

synadia-io / nats.java.v3 / 36784688561

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