#2055

Committed 15 Jul 2025 11:44AM UTC coverage: 95.67% (+0.09%) from 95.581%

Build # #2055

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1354 from nats-io/ordered-consumer-name-argh

[Fix] Simplified Ordered Consumer - Getting name early can cause NPE

Run Details

1 of 2 new or added lines in 1 file covered. (50.0%)

2 existing lines in 2 files now uncovered.

11844 of 12380 relevant lines covered (95.67%)

0.96 hits per line

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

83.72

/src/main/java/io/nats/client/impl/NatsMessageConsumerBase.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.JetStreamApiException;
import io.nats.client.MessageConsumer;
import io.nats.client.api.ConsumerInfo;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;

class NatsMessageConsumerBase implements MessageConsumer {
    protected NatsJetStreamPullSubscription sub;
    protected PullMessageManager pmm;
    protected final AtomicBoolean stopped;
    protected final AtomicBoolean finished;
    protected ConsumerInfo cachedConsumerInfo;
    protected String consumerName;

    NatsMessageConsumerBase(ConsumerInfo cachedConsumerInfo) {
        this.cachedConsumerInfo = cachedConsumerInfo;
        if (cachedConsumerInfo != null) {
            this.consumerName = cachedConsumerInfo.getName();
        }
        this.stopped = new AtomicBoolean(false);
        this.finished = new AtomicBoolean(false);
    }

    void setConsumerName(String consumerName) {
        this.consumerName = consumerName;
    }

    void initSub(NatsJetStreamPullSubscription sub) {
        this.sub = sub;
        this.consumerName = sub.getConsumerName();
        pmm = (PullMessageManager)sub.manager;
    }

    /**
     * {@inheritDoc}
     */
    public boolean isStopped() {
        return stopped.get();
    }

    /**
     * {@inheritDoc}
     */
    public boolean isFinished() {
        return finished.get();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String getConsumerName() {
        if (consumerName == null && cachedConsumerInfo != null) {
            consumerName = cachedConsumerInfo.getName();
        }
        return consumerName;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ConsumerInfo getConsumerInfo() throws IOException, JetStreamApiException {
        if (cachedConsumerInfo == null) {
            cachedConsumerInfo = sub.getConsumerInfo();
            consumerName = cachedConsumerInfo.getName();
        }
        return cachedConsumerInfo;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ConsumerInfo getCachedConsumerInfo() {
        return cachedConsumerInfo;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void stop() {
        stopped.set(true);
    }

    @Override
    public void close() throws Exception {
        stopped.set(true);
        shutdownSub();
    }

    protected void fullClose() {
        stopped.set(true);
        finished.set(true);
        shutdownSub();
    }

    protected void shutdownSub() {
        try {
            if (sub.isActive()) {
                if (sub.getNatsDispatcher() != null) {
                    sub.getDispatcher().unsubscribe(sub);
                }
                else {
                    sub.unsubscribe();
                }
            }
        }
        catch (Throwable ignore) {
            // nothing to do
        }
        if (pmm != null) {
            try {
                pmm.shutdownHeartbeatTimer();
            }
            catch (Throwable ignore) {
                // nothing to do
            }
        }
    }
}

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.JetStreamApiException;
17	import io.nats.client.MessageConsumer;
18	import io.nats.client.api.ConsumerInfo;
19
20	import java.io.IOException;
21	import java.util.concurrent.atomic.AtomicBoolean;
22
23	class NatsMessageConsumerBase implements MessageConsumer {
24	protected NatsJetStreamPullSubscription sub;
25	protected PullMessageManager pmm;
26	protected final AtomicBoolean stopped;
27	protected final AtomicBoolean finished;
28	protected ConsumerInfo cachedConsumerInfo;
29	protected String consumerName;
30
31	NatsMessageConsumerBase(ConsumerInfo cachedConsumerInfo) {	1✔
32	this.cachedConsumerInfo = cachedConsumerInfo;	1✔
33	if (cachedConsumerInfo != null) {	1✔
34	this.consumerName = cachedConsumerInfo.getName();	1✔
35	}
36	this.stopped = new AtomicBoolean(false);	1✔
37	this.finished = new AtomicBoolean(false);	1✔
38	}	1✔
39
40	void setConsumerName(String consumerName) {
41	this.consumerName = consumerName;	1✔
42	}	1✔
43
44	void initSub(NatsJetStreamPullSubscription sub) {
45	this.sub = sub;	1✔
46	this.consumerName = sub.getConsumerName();	1✔
47	pmm = (PullMessageManager)sub.manager;	1✔
48	}	1✔
49
50	/**
51	* {@inheritDoc}
52	*/
53	public boolean isStopped() {
54	return stopped.get();	×
55	}
56
57	/**
58	* {@inheritDoc}
59	*/
60	public boolean isFinished() {
61	return finished.get();	1✔
62	}
63
64	/**
65	* {@inheritDoc}
66	*/
67	@Override
68	public String getConsumerName() {
NEW 69	if (consumerName == null && cachedConsumerInfo != null) {	×
70	consumerName = cachedConsumerInfo.getName();	×
71	}
72	return consumerName;	×
73	}
74
75	/**
76	* {@inheritDoc}
77	*/
78	@Override
79	public ConsumerInfo getConsumerInfo() throws IOException, JetStreamApiException {
80	if (cachedConsumerInfo == null) {	1✔
81	cachedConsumerInfo = sub.getConsumerInfo();	1✔
82	consumerName = cachedConsumerInfo.getName();	1✔
83	}
84	return cachedConsumerInfo;	1✔
85	}
86
87	/**
88	* {@inheritDoc}
89	*/
90	@Override
91	public ConsumerInfo getCachedConsumerInfo() {
92	return cachedConsumerInfo;	×
93	}
94
95	/**
96	* {@inheritDoc}
97	*/
98	@Override
99	public void stop() {
100	stopped.set(true);	1✔
101	}	1✔
102
103	@Override
104	public void close() throws Exception {
105	stopped.set(true);	1✔
106	shutdownSub();	1✔
107	}	1✔
108
109	protected void fullClose() {
110	stopped.set(true);	1✔
111	finished.set(true);	1✔
112	shutdownSub();	1✔
113	}	1✔
114
115	protected void shutdownSub() {
116	try {
117	if (sub.isActive()) {	1✔
118	if (sub.getNatsDispatcher() != null) {	1✔
119	sub.getDispatcher().unsubscribe(sub);	1✔
120	}
121	else {
122	sub.unsubscribe();	1✔
123	}
124	}
125	}
126	catch (Throwable ignore) {	×
127	// nothing to do
128	}	1✔
129	if (pmm != null) {	1✔
130	try {
131	pmm.shutdownHeartbeatTimer();	1✔
132	}
133	catch (Throwable ignore) {	×
134	// nothing to do
135	}	1✔
136	}
137	}	1✔
138	}

nats-io / nats.java / #2055

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