36791038761

Committed 30 Sep 2026 11:25PM UTC coverage: 90.839% (-0.03%) from 90.87%

Build # 36791038761

Build Type

push

github

Committed by

scottf

Commit Message

Windows build tests only the affected modules

- build-windows.yml gets the same changes job as the module builds; the Windows job is skipped when no module is affected and otherwise runs :<module>:test for each affected module in one job, so nats-server is built once
- test report upload paths point at each module's build directory
- affected-modules.sh treats a build-windows.yml change as affecting every module

Coverage Stats

4187 of 4830 branches covered (86.69%)

Branch coverage included in aggregate %.

10647 of 11500 relevant lines covered (92.58%)

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 / 36791038761

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