36767740690

Committed 30 Sep 2026 07:44PM UTC coverage: 90.845%. Remained the same

Build # 36767740690

Build Type

push

github

Committed by

scottf

Commit Message

Publish snapshots without a staging repository

The nexus publish plugin defaults useStaging and packageGroup from the root project's version and group, which are unset because each subproject sets its own. Every snapshot publish therefore tried to open a release staging repository for an empty package group. useStaging now follows isRelease and packageGroup is io.synadia.

Also adds COVERALLS_MODULE_COVERAGE.md, the Coveralls findings for module-based builds.

Coverage Stats

4186 of 4830 branches covered (86.67%)

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

96.51

/core/src/main/java/io/synadia/client/impl/NatsServerPool.java

package io.synadia.client.impl;

import io.synadia.client.Options;
import io.synadia.client.ServerPool;
import io.synadia.client.utils.NatsConstants;
import io.synadia.client.utils.NatsUri;
import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;

import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.locks.ReentrantLock;

/**
 * The default {@link ServerPool} implementation. It holds the bootstrap servers from the options plus any
 * servers the nats-server advertises during connect, and hands them out in the order the connect and
 * reconnect logic should try them, honoring the no-randomize option.
 */
public class NatsServerPool implements ServerPool {

    protected final ReentrantLock listLock;
    protected List<ServerPoolEntry> entryList;
    protected Options options;
    protected int maxConnectAttempts;
    protected boolean hasSecureServer;
    protected NatsUri lastConnected;
    protected String defaultScheme;

    /**
     * Construct an empty pool. It is not usable until {@link #initialize(Options)} supplies the options
     * that provide the bootstrap servers.
     */
    public NatsServerPool() {
        listLock = new ReentrantLock();
        entryList = new ArrayList<>(); // this gets updated occasionally
        options = Options.builder().build(); // this will get updated when initialize is called
    }

    /**
     * {@inheritDoc}
     */
    public void initialize(@NonNull Options opts) {
        // 1. Hold on to options as we need them for settings
        options = opts;

        // 2. maxConnectAttempts accounts for the first connect attempt and also reconnect attempts
        maxConnectAttempts = options.getMaxReconnects() < 0 ? Integer.MAX_VALUE : options.getMaxReconnects() + 1;

        // 3. Add all the bootstrap to the server list and prepare list for next
        //    FYI bootstrap will always have at least the default url
        listLock.lock();
        try {
            for (NatsUri optionNuri : options.getNatsServerUris()) {
                // If optionNuri is not found in the list being built, add to the list
                // If optionNuri equivalent is found in the list and is secure, just use it
                boolean notAlreadyInList = true;
                for (ServerPoolEntry entry : entryList) {
                    if (optionNuri.equivalent(entry.nuri)) {
                        if (optionNuri.isSecure()) {
                            entryList.remove(entry);
                        }
                        else {
                            notAlreadyInList = false;
                        }
                        break;
                    }
                }
                if (notAlreadyInList) {
                    if (defaultScheme == null && !optionNuri.getScheme().equals(NatsConstants.NATS_PROTOCOL)) {
                        defaultScheme = optionNuri.getScheme();
                    }
                    entryList.add(new ServerPoolEntry(optionNuri, false));
                }
            }

            // prepare list for next, intentionally within lock
            afterListChanged();
        }
        finally {
            listLock.unlock();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean acceptDiscoveredUrls(@NonNull List<@NonNull String> discoveredServers) {
        // 1. If ignored discovered servers, don't do anything b/c never want
        //    anything but the explicit, which is already loaded.
        // 2. return false == no new servers discovered
        if (options.isIgnoreDiscoveredServers()) {
            return false;
        }

        listLock.lock();
        try {
            // 2. Build a list for discovered
            //    - since we'll need the NatsUris later
            //    - and to have a list to use to prune removed gossiped servers
            List<NatsUri> discovered = new ArrayList<>();
            for (String d : discoveredServers) {
                try {
                    discovered.add(new NatsUri(d, defaultScheme));
                } catch (URISyntaxException ignore) {
                    // should never actually happen
                }
            }

            // 3. Start a new server list, loading in current order from the current list, and keeping
            //    - the last connected
            //    - all non-gossiped
            //    - any found in the new discovered list
            //      - for any new discovered, we also remove them from
            //        that list so step there are no dupes for step #4
            //      - This also maintains the Srv state of an already known discovered
            List<ServerPoolEntry> newEntryList = new ArrayList<>();
            for (ServerPoolEntry entry : entryList) {
                int ix = findEquivalent(discovered, entry.nuri);
                if (ix != -1 || entry.nuri.equals(lastConnected) || !entry.isGossiped) {
                    newEntryList.add(entry);
                    if (ix != -1) {
                        discovered.remove(ix);
                    }
                }
            }

            // 4. Add all left over from the new discovered list
            boolean discoveryContainedUnknowns = false;
            if (!discovered.isEmpty()) {
                discoveryContainedUnknowns = true;
                for (NatsUri d : discovered) {
                    newEntryList.add(new ServerPoolEntry(d, true));
                }
            }

            // 5. replace the list with the new one
            entryList = newEntryList;

            // 6. prepare list for next
            afterListChanged();

            // 7.
            return discoveryContainedUnknowns;
        }
        finally {
            listLock.unlock();
        }
    }

    protected void afterListChanged() {
        // 1. randomize if needed and allowed
        if (entryList.size() > 1 && !options.isNoRandomize()) {
            Collections.shuffle(entryList, ThreadLocalRandom.current());
        }

        // 2. calculate hasSecureServer and find the index of lastConnected
        hasSecureServer = false;
        int lastConnectedIx = -1;
        for (int ix = 0; ix < entryList.size(); ix++) {
            NatsUri nuri = entryList.get(ix).nuri;
            hasSecureServer |= nuri.isSecure();
            if (nuri.equals(lastConnected)) {
                lastConnectedIx = ix;
            }
        }

        // C. put the last connected server at the end of the list
        if (lastConnectedIx != -1) {
            entryList.add(entryList.remove(lastConnectedIx));
        }
    }

    @Override
    @Nullable
    public NatsUri peekNextServer() {
        listLock.lock();
        try {
            return entryList.isEmpty() ? null : entryList.get(0).nuri;
        }
        finally {
            listLock.unlock();
        }
    }

    @Override
    @Nullable
    public NatsUri nextServer() {
        // 0. The list is already managed for qualified by connectFailed
        // 1. Get the first item in the list, update it's time, add back to the end of list
        listLock.lock();
        try {
            if (!entryList.isEmpty()) {
                ServerPoolEntry entry = entryList.remove(0);
                entry.lastAttempt = System.currentTimeMillis();
                entryList.add(entry);
                return entry.nuri;
            }
            return null;
        }
        finally {
            listLock.unlock();
        }
    }

    @Override
    @Nullable
    public List<String> resolveHostToIps(@NonNull String host, boolean maxOneResult, boolean includeIPV6) {
        return NatsHostResolver.resolveHostToIps(host, maxOneResult, includeIPV6);
    }

    @Override
    public void connectSucceeded(@NonNull NatsUri nuri) {
        // 1. Work from the end because nextServer moved the one being tried to the end
        // 2. If we find the server in the list...
        //    2.1. remember it and
        //    2.2. reset failed attempts
        listLock.lock();
        try {
            for (int x = entryList.size() - 1; x >= 0 ; x--) {
                ServerPoolEntry entry = entryList.get(x);
                if (entry.nuri.equals(nuri)) {
                    lastConnected = nuri;
                    entry.failedAttempts = 0;
                    return;
                }
            }
        }
        finally {
            listLock.unlock();
        }
    }

    @Override
    public void connectFailed(@NonNull NatsUri nuri) {
        // 1. Work from the end because nextServer moved the one being tried to the end
        // 2. If we find the server in the list...
        //    2.1. increment failed attempts
        //    2.2. if failed attempts reaches max, remove it from the list
        //         otherwise just move it to the end of the list
        listLock.lock();
        try {
            for (int x = entryList.size() - 1; x >= 0 ; x--) {
                ServerPoolEntry entry = entryList.get(x);
                if (entry.nuri.equals(nuri)) {
                    entryList.remove(x);
                    if (++entry.failedAttempts < maxConnectAttempts) {
                        entryList.add(entry);
                    }
                    return;
                }
            }
        }
        finally {
            listLock.unlock();
        }
    }

    @Override
    @NonNull
    public List<String> getServerList() {
        listLock.lock();
        try {
            List<String> list = new ArrayList<>();
            for (ServerPoolEntry entry : entryList) {
                list.add(entry.nuri.toString());
            }
            return list;
        }
        finally {
            listLock.unlock();
        }
    }

    @Override
    public boolean hasSecureServer() {
        return hasSecureServer;
    }

    protected int findEquivalent(List<NatsUri> list, NatsUri toFind) {
        for (int i = 0; i < list.size(); i++) {
            NatsUri nuri = list.get(i);
            if (nuri.equivalent(toFind)) {
                return i;
            }
        }
        return -1;
    }
}

1	package io.synadia.client.impl;
2
3	import io.synadia.client.Options;
4	import io.synadia.client.ServerPool;
5	import io.synadia.client.utils.NatsConstants;
6	import io.synadia.client.utils.NatsUri;
7	import org.jspecify.annotations.NonNull;
8	import org.jspecify.annotations.Nullable;
9
10	import java.net.URISyntaxException;
11	import java.util.ArrayList;
12	import java.util.Collections;
13	import java.util.List;
14	import java.util.concurrent.ThreadLocalRandom;
15	import java.util.concurrent.locks.ReentrantLock;
16
17	/**
18	* The default {@link ServerPool} implementation. It holds the bootstrap servers from the options plus any
19	* servers the nats-server advertises during connect, and hands them out in the order the connect and
20	* reconnect logic should try them, honoring the no-randomize option.
21	*/
22	public class NatsServerPool implements ServerPool {
23
24	protected final ReentrantLock listLock;
25	protected List<ServerPoolEntry> entryList;
26	protected Options options;
27	protected int maxConnectAttempts;
28	protected boolean hasSecureServer;
29	protected NatsUri lastConnected;
30	protected String defaultScheme;
31
32	/**
33	* Construct an empty pool. It is not usable until {@link #initialize(Options)} supplies the options
34	* that provide the bootstrap servers.
35	*/
36	public NatsServerPool() {	2 ✔
37	listLock = new ReentrantLock();	5 ✔
38	entryList = new ArrayList<>(); // this gets updated occasionally	5 ✔
39	options = Options.builder().build(); // this will get updated when initialize is called	4 ✔
40	}	1 ✔
41
42	/**
43	* {@inheritDoc}
44	*/
45	public void initialize(@NonNull Options opts) {
46	// 1. Hold on to options as we need them for settings
47	options = opts;	3 ✔
48
49	// 2. maxConnectAttempts accounts for the first connect attempt and also reconnect attempts
50	maxConnectAttempts = options.getMaxReconnects() < 0 ? Integer.MAX_VALUE : options.getMaxReconnects() + 1;	13 ✔
51
52	// 3. Add all the bootstrap to the server list and prepare list for next
53	// FYI bootstrap will always have at least the default url
54	listLock.lock();	3 ✔
55	try {
56	for (NatsUri optionNuri : options.getNatsServerUris()) {	12 ✔
57	// If optionNuri is not found in the list being built, add to the list
58	// If optionNuri equivalent is found in the list and is secure, just use it
59	boolean notAlreadyInList = true;	2 ✔
60	for (ServerPoolEntry entry : entryList) {	11 ✔
61	if (optionNuri.equivalent(entry.nuri)) {	5 ✔
62	if (optionNuri.isSecure()) {	3 ✔
63	entryList.remove(entry);	6 ✔
64	}
65	else {
66	notAlreadyInList = false;	2 ✔
67	}
68	break;	1 ✔
69	}
70	}	1 ✔
71	if (notAlreadyInList) {	2 ✔
72	if (defaultScheme == null && !optionNuri.getScheme().equals(NatsConstants.NATS_PROTOCOL)) {	8 ✔
73	defaultScheme = optionNuri.getScheme();	4 ✔
74	}
75	entryList.add(new ServerPoolEntry(optionNuri, false));	9 ✔
76	}
77	}	1 ✔
78
79	// prepare list for next, intentionally within lock
80	afterListChanged();	2 ✔
81	}
82	finally {
83	listLock.unlock();	3 ✔
84	}
85	}	1 ✔
86
87	/**
88	* {@inheritDoc}
89	*/
90	@Override
91	public boolean acceptDiscoveredUrls(@NonNull List<@NonNull String> discoveredServers) {
92	// 1. If ignored discovered servers, don't do anything b/c never want
93	// anything but the explicit, which is already loaded.
94	// 2. return false == no new servers discovered
95	if (options.isIgnoreDiscoveredServers()) {	4 ✔
96	return false;	2 ✔
97	}
98
99	listLock.lock();	3 ✔
100	try {
101	// 2. Build a list for discovered
102	// - since we'll need the NatsUris later
103	// - and to have a list to use to prune removed gossiped servers
104	List<NatsUri> discovered = new ArrayList<>();	4 ✔
105	for (String d : discoveredServers) {	10 ✔
106	try {
107	discovered.add(new NatsUri(d, defaultScheme));	9 ✔
108	} catch (URISyntaxException ignore) {	×
109	// should never actually happen
110	}	1 ✔
111	}	1 ✔
112
113	// 3. Start a new server list, loading in current order from the current list, and keeping
114	// - the last connected
115	// - all non-gossiped
116	// - any found in the new discovered list
117	// - for any new discovered, we also remove them from
118	// that list so step there are no dupes for step #4
119	// - This also maintains the Srv state of an already known discovered
120	List<ServerPoolEntry> newEntryList = new ArrayList<>();	4 ✔
121	for (ServerPoolEntry entry : entryList) {	11 ✔
122	int ix = findEquivalent(discovered, entry.nuri);	6 ✔
123	if (ix != -1 \|\| entry.nuri.equals(lastConnected) \|\| !entry.isGossiped) {	12 ✔
124	newEntryList.add(entry);	4 ✔
125	if (ix != -1) {	3 ✔
126	discovered.remove(ix);	4 ✔
127	}
128	}
129	}	1 ✔
130
131	// 4. Add all left over from the new discovered list
132	boolean discoveryContainedUnknowns = false;	2 ✔
133	if (!discovered.isEmpty()) {	3 ✔
134	discoveryContainedUnknowns = true;	2 ✔
135	for (NatsUri d : discovered) {	10 ✔
136	newEntryList.add(new ServerPoolEntry(d, true));	8 ✔
137	}	1 ✔
138	}
139
140	// 5. replace the list with the new one
141	entryList = newEntryList;	3 ✔
142
143	// 6. prepare list for next
144	afterListChanged();	2 ✔
145
146	// 7.
147	return discoveryContainedUnknowns;	4 ✔
148	}
149	finally {
150	listLock.unlock();	3 ✔
151	}
152	}
153
154	protected void afterListChanged() {
155	// 1. randomize if needed and allowed
156	if (entryList.size() > 1 && !options.isNoRandomize()) {	9 ✔
157	Collections.shuffle(entryList, ThreadLocalRandom.current());	4 ✔
158	}
159
160	// 2. calculate hasSecureServer and find the index of lastConnected
161	hasSecureServer = false;	3 ✔
162	int lastConnectedIx = -1;	2 ✔
163	for (int ix = 0; ix < entryList.size(); ix++) {	9 ✔
164	NatsUri nuri = entryList.get(ix).nuri;	7 ✔
165	hasSecureServer \|= nuri.isSecure();	7 ✔
166	if (nuri.equals(lastConnected)) {	5 ✔
167	lastConnectedIx = ix;	2 ✔
168	}
169	}
170
171	// C. put the last connected server at the end of the list
172	if (lastConnectedIx != -1) {	3 ✔
173	entryList.add(entryList.remove(lastConnectedIx));	9 ✔
174	}
175	}	1 ✔
176
177	@Override
178	@Nullable
179	public NatsUri peekNextServer() {
180	listLock.lock();	3 ✔
181	try {
182	return entryList.isEmpty() ? null : entryList.get(0).nuri;	15 ✔
183	}
184	finally {
185	listLock.unlock();	3 ✔
186	}
187	}
188
189	@Override
190	@Nullable
191	public NatsUri nextServer() {
192	// 0. The list is already managed for qualified by connectFailed
193	// 1. Get the first item in the list, update it's time, add back to the end of list
194	listLock.lock();	3 ✔
195	try {
196	if (!entryList.isEmpty()) {	4 ✔
197	ServerPoolEntry entry = entryList.remove(0);	6 ✔
198	entry.lastAttempt = System.currentTimeMillis();	3 ✔
199	entryList.add(entry);	5 ✔
200	return entry.nuri;	5 ✔
201	}
202	return null;	4 ✔
203	}
204	finally {
205	listLock.unlock();	3 ✔
206	}
207	}
208
209	@Override
210	@Nullable
211	public List<String> resolveHostToIps(@NonNull String host, boolean maxOneResult, boolean includeIPV6) {
212	return NatsHostResolver.resolveHostToIps(host, maxOneResult, includeIPV6);	5 ✔
213	}
214
215	@Override
216	public void connectSucceeded(@NonNull NatsUri nuri) {
217	// 1. Work from the end because nextServer moved the one being tried to the end
218	// 2. If we find the server in the list...
219	// 2.1. remember it and
220	// 2.2. reset failed attempts
221	listLock.lock();	3 ✔
222	try {
223	for (int x = entryList.size() - 1; x >= 0 ; x--) {	10 !
224	ServerPoolEntry entry = entryList.get(x);	6 ✔
225	if (entry.nuri.equals(nuri)) {	5 ✔
226	lastConnected = nuri;	3 ✔
227	entry.failedAttempts = 0;	3 ✔
228	return;	1 ✔
229	}
230	}
231	}
232	finally {
233	listLock.unlock();	3 ✔
234	}
235	}	×
236
237	@Override
238	public void connectFailed(@NonNull NatsUri nuri) {
239	// 1. Work from the end because nextServer moved the one being tried to the end
240	// 2. If we find the server in the list...
241	// 2.1. increment failed attempts
242	// 2.2. if failed attempts reaches max, remove it from the list
243	// otherwise just move it to the end of the list
244	listLock.lock();	3 ✔
245	try {
246	for (int x = entryList.size() - 1; x >= 0 ; x--) {	8 !
247	ServerPoolEntry entry = entryList.get(x);	6 ✔
248	if (entry.nuri.equals(nuri)) {	5 !
249	entryList.remove(x);	5 ✔
250	if (++entry.failedAttempts < maxConnectAttempts) {	10 ✔
251	entryList.add(entry);	5 ✔
252	}
253	return;	1 ✔
254	}
255	}
256	}
257	finally {
258	listLock.unlock();	3 ✔
259	}
260	}	×
261
262	@Override
263	@NonNull
264	public List<String> getServerList() {
265	listLock.lock();	3 ✔
266	try {
267	List<String> list = new ArrayList<>();	4 ✔
268	for (ServerPoolEntry entry : entryList) {	11 ✔
269	list.add(entry.nuri.toString());	6 ✔
270	}	1 ✔
271	return list;	4 ✔
272	}
273	finally {
274	listLock.unlock();	3 ✔
275	}
276	}
277
278	@Override
279	public boolean hasSecureServer() {
280	return hasSecureServer;	3 ✔
281	}
282
283	protected int findEquivalent(List<NatsUri> list, NatsUri toFind) {
284	for (int i = 0; i < list.size(); i++) {	8 ✔
285	NatsUri nuri = list.get(i);	5 ✔
286	if (nuri.equivalent(toFind)) {	4 ✔
287	return i;	2 ✔
288	}
289	}
290	return -1;	2 ✔
291	}
292	}

synadia-io / nats.java.v3 / 36767740690

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