37648054681

Committed 07 Oct 2026 03:55PM UTC coverage: 89.174% (-2.1%) from 91.29%

Build # 37648054681

Build Type

push

github

Committed by

scottf

Commit Message

NatsConnectionV3 (default) with OutboundBuffer, classic fixes, build-variants CI

- NatsConnectionV3 extends NatsConnection, selected by ConnectionImplementation (V3 default, Classic available via option, property or -PconnectionImplementation)
- OutboundBuffer: control lane + bounded data blocks; SUB control line limit check; kept-block limits from the first server's max payload
- Classic fixes: protocol buffer sizing, tryingToConnect CAS, lock-free pending getters, sendPing pong future cleanup, per-run reader/writer flag
- build-variants.yml replaces build-windows.yml
- Plans: connection V3, active/passive V3, testing.java overview, V2 bugs from review

Coverage Stats

4596 of 5421 branches covered (84.78%)

Branch coverage included in aggregate %.

1027 of 1185 new or added lines in 12 files covered. (86.67%)

161 existing lines in 4 files now uncovered.

11532 of 12665 relevant lines covered (91.05%)

4.03 hits per line

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

80.92

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

package io.synadia.client.impl;

import io.synadia.client.ForceReconnectOptions;
import io.synadia.client.HostnameResolveMode;
import io.synadia.client.Options;
import io.synadia.client.global.NatsSystemClock;
import io.synadia.client.utils.HappyEyeballsConnector;
import io.synadia.client.utils.NatsUri;
import io.synadia.client.utils.ScheduledTask;
import io.synadia.client.utils.WebSocket;
import org.jspecify.annotations.NonNull;

import javax.net.ssl.HandshakeCompletedListener;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static io.synadia.client.utils.NatsConstants.NANOS_PER_MILLI;
import static io.synadia.client.utils.NatsConstants.SECURE_WEBSOCKET_PROTOCOL;

/**
 * This class is not thread-safe.  Caller must ensure thread safety.
 */
public class SocketDataPort implements DataPort {

    protected NatsConnection connection;

    protected String host;
    protected int port;
    protected Socket socket;
    protected boolean isSecure = false;

    protected InputStream in;
    protected OutputStream out;

    // Write-timeout watch, gated on socketWriteTimeout > 0;
    // - when it is <= 0, there is no watch.
    private long writeTimeoutNanos;
    private long delayPeriodNanos;
    private ScheduledTask writeWatchTask;
    private final AtomicLong writeMustBeDoneBy = new AtomicLong(Long.MAX_VALUE);

    /**
     * Construct an unconnected data port. Data ports are instantiated by the connection,
     * then configured through afterConstruct and opened through connect.
     */
    public SocketDataPort() {}

    @Override
    public void afterConstruct(@NonNull Options options) {
        long millis = options.getSocketWriteTimeout();
        if (millis > 0) {
            writeTimeoutNanos = millis * NANOS_PER_MILLI;
            delayPeriodNanos = writeTimeoutNanos * 51 / 100;
        }
        // millis <= 0 -> writeTimeoutNanos stays 0 -> no write-timeout watch
    }

    @Override
    public void connect(@NonNull NatsConnection conn, @NonNull NatsUri nuri, long timeoutNanos) throws IOException {
        connection = conn;
        Options options = connection.getOptions();
        long timeout = timeoutNanos / 1_000_000; // convert to millis
        host = nuri.getHost();
        port = nuri.getPort();

        try {
            HostnameResolveMode mode = options.hostnameResolveMode();
            if (mode == HostnameResolveMode.HappyEyeballs) {
                socket = HappyEyeballsConnector.connect(
                    options.getExecutor(),
                    () -> createSocket(options),
                    host, port, (int) timeout
                );
            }
            else {
                socket = createSocket(options);
                InetSocketAddress inetSocketAddress;
                if (mode == HostnameResolveMode.Unresolved && !nuri.hostIsIpAddress()) {
                    inetSocketAddress = InetSocketAddress.createUnresolved(host, port);
                }
                else {
                    inetSocketAddress = new InetSocketAddress(host, port);
                }
                socket.connect(inetSocketAddress, (int) timeout);
            }

            if (options.getSocketReadTimeout() > 0) {
                socket.setSoTimeout((int) options.getSocketReadTimeout()); // SO_TIMEOUT is millis and int-typed; a read timeout never exceeds int range
            }

            if (options.getSocketSoLinger() > 0) {
                socket.setSoLinger(true, options.getSocketSoLinger());
            }

            if (options.getSocketReceiveBufferSize() > 0) {
                socket.setReceiveBufferSize(options.getSocketReceiveBufferSize());
            }

            if (options.getSocketSendBufferSize() > 0) {
                socket.setSendBufferSize(options.getSocketSendBufferSize());
            }

            if (nuri.isWebsocket()) {
                if (SECURE_WEBSOCKET_PROTOCOL.equalsIgnoreCase(nuri.getScheme())) {
                    upgradeToSecure();
                }
                try {
                    socket = new WebSocket(socket, host, options.getHttpRequestInterceptors(), nuri.getUri().getPath());
                } catch (Exception ex) {
                    socket.close();
                    throw ex;
                }
            }
            in = socket.getInputStream();
            out = socket.getOutputStream();
        }
        catch (Exception e) {
            if (socket != null) {
                try { socket.close(); } catch (Exception ignore) {}
            }
            socket = null;
            if (e instanceof IOException) {
                throw e;
            }
            throw new IOException(e);
        }

        if (writeTimeoutNanos > 0) {
            writeWatchTask = new ScheduledTask(connection.getScheduledExecutor(), delayPeriodNanos, TimeUnit.NANOSECONDS,
                () -> {
                    // if now is after when the write was supposed to be done by, the socket write is stuck
                    if (NatsSystemClock.nanoTime() > writeMustBeDoneBy.get()) {
                        writeWatchTask.shutdown(); // connection is going to be closed; no need to repeat this
                        connection.notifyErrorListener((c, el) -> el.socketWriteTimeout(c));
                        try {
                            connection.forceReconnect(ForceReconnectOptions.FORCE_CLOSE_INSTANCE);
                        }
                        catch (IOException e) {
                            // retry maybe?
                        }
                        catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            // This task is going to re-run anyway, so no point in throwing
                        }
                    }
                });
        }
    }

    /**
     * Upgrade the port to SSL. If it is already secured, this is a no-op.
     * If the data port type doesn't support SSL it should throw an exception.
     */
    public void upgradeToSecure() throws IOException {
        Options options = connection.getOptions();
        SSLContext context = options.getSslContext();

        SSLSocketFactory factory = context.getSocketFactory();

        SSLSocket sslSocket = (SSLSocket) factory.createSocket(socket, host, port, true);
        sslSocket.setUseClientMode(true);

        final CompletableFuture<Void> waitForHandshake = new CompletableFuture<>();
        final HandshakeCompletedListener hcl = (evt) -> waitForHandshake.complete(null);

        sslSocket.addHandshakeCompletedListener(hcl);
        sslSocket.startHandshake();

        try {
            waitForHandshake.get(options.getConnectionTimeout(), TimeUnit.MILLISECONDS);
        } catch (Exception ex) {
            connection.handleCommunicationIssue(ex);
            return;
        }
        finally {
            sslSocket.removeHandshakeCompletedListener(hcl);
        }

        socket = sslSocket;
        in = sslSocket.getInputStream();
        out = sslSocket.getOutputStream();
        isSecure = true;
    }

    public int read(byte[] dst, int off, int len) throws IOException {
        return in.read(dst, off, len);
    }

    public void write(byte[] src, int toWrite) throws IOException {
        if (writeTimeoutNanos > 0) {
            writeMustBeDoneBy.set(NatsSystemClock.nanoTime() + writeTimeoutNanos);
            out.write(src, 0, toWrite);
            writeMustBeDoneBy.set(Long.MAX_VALUE);
        }
        else {
            out.write(src, 0, toWrite);
        }
    }

    public void shutdownInput() throws IOException {
        // cannot call shutdownInput on sslSocket
        if (!isSecure && socket != null) {
            socket.shutdownInput();
        }
    }

    public void close() throws IOException {
        if (writeWatchTask != null) {
            writeWatchTask.shutdown();
        }
        if (socket != null) {
            socket.close();
        }
    }

    @Override
    public void forceClose() throws IOException {
        // socket can technically be null, like between states
        // practically it never will be, but guard it anyway
        if (socket != null) {
            try {
                // If we are being asked to force close, there is no need to linger.
                socket.setSoLinger(true, 0);
            }
            catch (SocketException e) {
                // don't want to fail if I couldn't set linger
            }
            close();
        }
    }

    public void flush() throws IOException {
        out.flush();
    }

    private Socket createSocket(Options options) throws SocketException {
        Socket socket;
        if (options.getProxy() != null) {
            socket = new Socket(options.getProxy());
        } else {
            socket = new Socket();
        }
        socket.setTcpNoDelay(true);
        socket.setReceiveBufferSize(2 * 1024 * 1024);
        socket.setSendBufferSize(2 * 1024 * 1024);
        return socket;
    }
}

1	package io.synadia.client.impl;
2
3	import io.synadia.client.ForceReconnectOptions;
4	import io.synadia.client.HostnameResolveMode;
5	import io.synadia.client.Options;
6	import io.synadia.client.global.NatsSystemClock;
7	import io.synadia.client.utils.HappyEyeballsConnector;
8	import io.synadia.client.utils.NatsUri;
9	import io.synadia.client.utils.ScheduledTask;
10	import io.synadia.client.utils.WebSocket;
11	import org.jspecify.annotations.NonNull;
12
13	import javax.net.ssl.HandshakeCompletedListener;
14	import javax.net.ssl.SSLContext;
15	import javax.net.ssl.SSLSocket;
16	import javax.net.ssl.SSLSocketFactory;
17	import java.io.IOException;
18	import java.io.InputStream;
19	import java.io.OutputStream;
20	import java.net.InetSocketAddress;
21	import java.net.Socket;
22	import java.net.SocketException;
23	import java.util.concurrent.CompletableFuture;
24	import java.util.concurrent.TimeUnit;
25	import java.util.concurrent.atomic.AtomicLong;
26
27	import static io.synadia.client.utils.NatsConstants.NANOS_PER_MILLI;
28	import static io.synadia.client.utils.NatsConstants.SECURE_WEBSOCKET_PROTOCOL;
29
30	/**
31	* This class is not thread-safe. Caller must ensure thread safety.
32	*/
33	public class SocketDataPort implements DataPort {
34
35	protected NatsConnection connection;
36
37	protected String host;
38	protected int port;
39	protected Socket socket;
40	protected boolean isSecure = false;	3 ✔
41
42	protected InputStream in;
43	protected OutputStream out;
44
45	// Write-timeout watch, gated on socketWriteTimeout > 0;
46	// - when it is <= 0, there is no watch.
47	private long writeTimeoutNanos;
48	private long delayPeriodNanos;
49	private ScheduledTask writeWatchTask;
50	private final AtomicLong writeMustBeDoneBy = new AtomicLong(Long.MAX_VALUE);	6 ✔
51
52	/**
53	* Construct an unconnected data port. Data ports are instantiated by the connection,
54	* then configured through afterConstruct and opened through connect.
55	*/
56	public SocketDataPort() {}	3 ✔
57
58	@Override
59	public void afterConstruct(@NonNull Options options) {
60	long millis = options.getSocketWriteTimeout();	3 ✔
61	if (millis > 0) {	4 ✔
62	writeTimeoutNanos = millis * NANOS_PER_MILLI;	5 ✔
63	delayPeriodNanos = writeTimeoutNanos * 51 / 100;	8 ✔
64	}
65	// millis <= 0 -> writeTimeoutNanos stays 0 -> no write-timeout watch
66	}	1 ✔
67
68	@Override
69	public void connect(@NonNull NatsConnection conn, @NonNull NatsUri nuri, long timeoutNanos) throws IOException {
70	connection = conn;	3 ✔
71	Options options = connection.getOptions();	4 ✔
72	long timeout = timeoutNanos / 1_000_000; // convert to millis	4 ✔
73	host = nuri.getHost();	4 ✔
74	port = nuri.getPort();	4 ✔
75
76	try {
77	HostnameResolveMode mode = options.hostnameResolveMode();	3 ✔
78	if (mode == HostnameResolveMode.HappyEyeballs) {	3 ✔
79	socket = HappyEyeballsConnector.connect(	5 ✔
80	options.getExecutor(),	10 ✔
81	() -> createSocket(options),	4 ✔
82	host, port, (int) timeout
83	);
84	}
85	else {
86	socket = createSocket(options);	5 ✔
87	InetSocketAddress inetSocketAddress;
88	if (mode == HostnameResolveMode.Unresolved && !nuri.hostIsIpAddress()) {	6 !
89	inetSocketAddress = InetSocketAddress.createUnresolved(host, port);	7 ✔
90	}
91	else {
92	inetSocketAddress = new InetSocketAddress(host, port);	8 ✔
93	}
94	socket.connect(inetSocketAddress, (int) timeout);	6 ✔
95	}
96
97	if (options.getSocketReadTimeout() > 0) {	5 !
98	socket.setSoTimeout((int) options.getSocketReadTimeout()); // SO_TIMEOUT is millis and int-typed; a read timeout never exceeds int range	×
99	}
100
101	if (options.getSocketSoLinger() > 0) {	3 !
102	socket.setSoLinger(true, options.getSocketSoLinger());	×
103	}
104
105	if (options.getSocketReceiveBufferSize() > 0) {	3 !
106	socket.setReceiveBufferSize(options.getSocketReceiveBufferSize());	×
107	}
108
109	if (options.getSocketSendBufferSize() > 0) {	3 !
110	socket.setSendBufferSize(options.getSocketSendBufferSize());	×
111	}
112
113	if (nuri.isWebsocket()) {	3 ✔
114	if (SECURE_WEBSOCKET_PROTOCOL.equalsIgnoreCase(nuri.getScheme())) {	5 ✔
115	upgradeToSecure();	2 ✔
116	}
117	try {
118	socket = new WebSocket(socket, host, options.getHttpRequestInterceptors(), nuri.getUri().getPath());	14 ✔
119	} catch (Exception ex) {	1 ✔
120	socket.close();	3 ✔
121	throw ex;	2 ✔
122	}	1 ✔
123	}
124	in = socket.getInputStream();	5 ✔
125	out = socket.getOutputStream();	5 ✔
126	}
127	catch (Exception e) {	1 ✔
128	if (socket != null) {	3 !
129	try { socket.close(); } catch (Exception ignore) {}	4 ✔
130	}
131	socket = null;	3 ✔
132	if (e instanceof IOException) {	3 ✔
133	throw e;	2 ✔
134	}
135	throw new IOException(e);	5 ✔
136	}	1 ✔
137
138	if (writeTimeoutNanos > 0) {	5 ✔
139	writeWatchTask = new ScheduledTask(connection.getScheduledExecutor(), delayPeriodNanos, TimeUnit.NANOSECONDS,	13 ✔
140	() -> {
141	// if now is after when the write was supposed to be done by, the socket write is stuck
142	if (NatsSystemClock.nanoTime() > writeMustBeDoneBy.get()) {	×
143	writeWatchTask.shutdown(); // connection is going to be closed; no need to repeat this	×
144	connection.notifyErrorListener((c, el) -> el.socketWriteTimeout(c));	×
145	try {
146	connection.forceReconnect(ForceReconnectOptions.FORCE_CLOSE_INSTANCE);	×
147	}
148	catch (IOException e) {	×
149	// retry maybe?
150	}
151	catch (InterruptedException e) {	×
152	Thread.currentThread().interrupt();	×
153	// This task is going to re-run anyway, so no point in throwing
154	}	×
155	}
156	});	×
157	}
158	}	1 ✔
159
160	/**
161	* Upgrade the port to SSL. If it is already secured, this is a no-op.
162	* If the data port type doesn't support SSL it should throw an exception.
163	*/
164	public void upgradeToSecure() throws IOException {
165	Options options = connection.getOptions();	4 ✔
166	SSLContext context = options.getSslContext();	3 ✔
167
168	SSLSocketFactory factory = context.getSocketFactory();	3 ✔
169
170	SSLSocket sslSocket = (SSLSocket) factory.createSocket(socket, host, port, true);	11 ✔
171	sslSocket.setUseClientMode(true);	3 ✔
172
173	final CompletableFuture<Void> waitForHandshake = new CompletableFuture<>();	4 ✔
174	final HandshakeCompletedListener hcl = (evt) -> waitForHandshake.complete(null);	8 ✔
175
176	sslSocket.addHandshakeCompletedListener(hcl);	3 ✔
177	sslSocket.startHandshake();	2 ✔
178
179	try {
180	waitForHandshake.get(options.getConnectionTimeout(), TimeUnit.MILLISECONDS);	6 ✔
181	} catch (Exception ex) {	×
182	connection.handleCommunicationIssue(ex);	×
183	return;	×
184	}
185	finally {
186	sslSocket.removeHandshakeCompletedListener(hcl);	3 ✔
187	}
188
189	socket = sslSocket;	3 ✔
190	in = sslSocket.getInputStream();	4 ✔
191	out = sslSocket.getOutputStream();	4 ✔
192	isSecure = true;	3 ✔
193	}	1 ✔
194
195	public int read(byte[] dst, int off, int len) throws IOException {
196	return in.read(dst, off, len);	7 ✔
197	}
198
199	public void write(byte[] src, int toWrite) throws IOException {
200	if (writeTimeoutNanos > 0) {	5 ✔
201	writeMustBeDoneBy.set(NatsSystemClock.nanoTime() + writeTimeoutNanos);	7 ✔
202	out.write(src, 0, toWrite);	6 ✔
203	writeMustBeDoneBy.set(Long.MAX_VALUE);	5 ✔
204	}
205	else {
206	out.write(src, 0, toWrite);	6 ✔
207	}
208	}	1 ✔
209
210	public void shutdownInput() throws IOException {
211	// cannot call shutdownInput on sslSocket
212	if (!isSecure && socket != null) {	3 !
UNCOV 213	socket.shutdownInput();	×
214	}
215	}	1 ✔
216
217	public void close() throws IOException {
218	if (writeWatchTask != null) {	3 ✔
219	writeWatchTask.shutdown();	3 ✔
220	}
221	if (socket != null) {	3 !
222	socket.close();	3 ✔
223	}
224	}	1 ✔
225
226	@Override
227	public void forceClose() throws IOException {
228	// socket can technically be null, like between states
229	// practically it never will be, but guard it anyway
230	if (socket != null) {	3 ✔
231	try {
232	// If we are being asked to force close, there is no need to linger.
233	socket.setSoLinger(true, 0);	5 ✔
234	}
235	catch (SocketException e) {	1 ✔
236	// don't want to fail if I couldn't set linger
237	}	1 ✔
238	close();	2 ✔
239	}
240	}	1 ✔
241
242	public void flush() throws IOException {
243	out.flush();	3 ✔
244	}	1 ✔
245
246	private Socket createSocket(Options options) throws SocketException {
247	Socket socket;
248	if (options.getProxy() != null) {	3 ✔
249	socket = new Socket(options.getProxy());	7 ✔
250	} else {
251	socket = new Socket();	4 ✔
252	}
253	socket.setTcpNoDelay(true);	3 ✔
254	socket.setReceiveBufferSize(2 * 1024 * 1024);	3 ✔
255	socket.setSendBufferSize(2 * 1024 * 1024);	3 ✔
256	return socket;	2 ✔
257	}
258	}

synadia-io / nats.java.v3 / 37648054681

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