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

71.78

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


// Copyright 2015-2026 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.synadia.client.impl;

import io.synadia.client.Options;
import io.synadia.client.StatisticsCollector;
import io.synadia.client.utils.ByteArrayBuilder;

import java.io.IOException;
import java.nio.BufferOverflowException;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import static io.synadia.client.OptionsConstants.MAX_MESSAGES_IN_NETWORK_BUFFER;
import static io.synadia.client.impl.MarkerMessage.END_RECONNECT;
import static io.synadia.client.utils.ByteArrayBuilder.bufferAllocSize;
import static io.synadia.client.utils.NatsConstants.CR;
import static io.synadia.client.utils.NatsConstants.LF;

/**
 * The connection's writer thread. It drains the outgoing message queues into the data port,
 * switching between the normal queue and the reconnect queue as the connection state changes.
 */
public class NatsConnectionWriter implements Runnable {
    enum Mode {
        Normal, Reconnect, WaitingForEndReconnect
    }
    private static final int BUFFER_BLOCK_SIZE = 256;

    private final NatsConnection connection;

    private final ReentrantLock writerLock;
    private Future<Boolean> stopped;
    private Future<DataPort> dataPortFuture;
    private DataPort dataPort;
    // A new flag per start(). The thread started by start() checks and clears only the flag it was
    // started with, so a late thread from a previous socket cannot stop or disturb the current one.
    private volatile AtomicBoolean running;
    private final AtomicReference<Mode> mode;
    private final ReentrantLock startStopLock;

    private byte[] sendBuffer;
    private final AtomicInteger sendBufferLength;

    private final WriterMessageQueue normalOutgoing;
    private final WriterMessageQueue reconnectOutgoing;
    private final long reconnectBufferSize;

    NatsConnectionWriter(NatsConnection connection) {
        this.connection = connection;
        writerLock = new ReentrantLock();

        this.running = new AtomicBoolean(false);
        mode = new AtomicReference<>(Mode.Normal);
        this.startStopLock = new ReentrantLock();
        this.stopped = new CompletableFuture<>();
        ((CompletableFuture<Boolean>)this.stopped).complete(Boolean.TRUE); // we are stopped on creation

        Options options = connection.getOptions();
        int sbl = bufferAllocSize(options.getBufferSize(), BUFFER_BLOCK_SIZE);
        sendBufferLength = new AtomicInteger(sbl);
        sendBuffer = new byte[sbl];
        reconnectBufferSize = options.getReconnectBufferSize();

        normalOutgoing = new WriterMessageQueue(
            options.getMaxMessagesInOutgoingQueue(),
            options.isDiscardMessagesWhenOutgoingQueueFull(),
            options.getWriteQueuePushTimeout()
        );

        // The "reconnect" buffer contains internal messages
        // Using this constructor makes it unbounded and without "discard when full"
        reconnectOutgoing = new WriterMessageQueue(options.getWriteQueuePushTimeout());
    }

    // Should only be called if the current thread has exited.
    // Use the Future from stop() to determine if it is ok to call this.
    // This method resets that future so mistiming can result in badness.
    void start(Future<DataPort> dataPortFuture) {
        this.startStopLock.lock();
        try {
            this.dataPortFuture = dataPortFuture;
            AtomicBoolean runFlag = new AtomicBoolean(true);
            this.running = runFlag;
            this.normalOutgoing.resume();
            this.reconnectOutgoing.resume();
            this.stopped = connection.getWriterExecutor().submit(() -> run(runFlag), Boolean.TRUE);
        } finally {
            this.startStopLock.unlock();
        }
    }

    // May be called several times on an error.
    // Returns a future that is completed when the thread completes, not when this
    // method does.
    Future<Boolean> stop() {
        AtomicBoolean runFlag = running;
        if (runFlag.get()) {
            runFlag.set(false);
            startStopLock.lock();
            try {
                this.normalOutgoing.pause();
                this.reconnectOutgoing.pause();
                this.normalOutgoing.filter();
            }
            finally {
                this.startStopLock.unlock();
            }
        }
        return this.stopped;
    }

    boolean isRunning() {
        return running.get();
    }

    void sendMessageBatch(NatsMessage msg, DataPort dataPort, StatisticsCollector stats) throws IOException {
        writerLock.lock();
        try {
            int sendPosition = 0;
            int sbl = sendBufferLength.get();

            while (msg != null) {
                if (msg == END_RECONNECT) {
                    mode.set(Mode.Normal);
                    break;
                }
                long size = msg.getSizeInBytes();

                if (sendPosition + size > sbl) {
                    if (sendPosition > 0) {
                        dataPort.write(sendBuffer, sendPosition);
                        stats.registerWrite(sendPosition);
                        sendPosition = 0;
                    }
                    if (size > sbl) { // have to resize b/c can't fit 1 message
                        sbl = bufferAllocSize((int) size, BUFFER_BLOCK_SIZE);
                        sendBufferLength.set(sbl);
                        sendBuffer = new byte[sbl];
                    }
                }

                ByteArrayBuilder bab = msg.getProtocolBab();
                int babLen = bab.length();
                System.arraycopy(bab.internalArray(), 0, sendBuffer, sendPosition, babLen);
                sendPosition += babLen;

                sendBuffer[sendPosition++] = CR;
                sendBuffer[sendPosition++] = LF;

                if (!msg.isProtocol()) { // because a protocol message does not have headers or data
                    sendPosition += msg.copyNotEmptyHeaders(sendPosition, sendBuffer);

                    byte[] bytes = msg.getData(); // guaranteed to not be null
                    if (bytes.length > 0) {
                        System.arraycopy(bytes, 0, sendBuffer, sendPosition, bytes.length);
                        sendPosition += bytes.length;
                    }

                    sendBuffer[sendPosition++] = CR;
                    sendBuffer[sendPosition++] = LF;
                }

                stats.incrementOut(size);

                if (msg.flushImmediatelyAfterPublish) {
                    dataPort.write(sendBuffer, sendPosition);
                    dataPort.flush();
                    stats.registerWrite(sendPosition);
                    sendPosition = 0;
                }

                msg = msg.next;
            }

            // no need to write if there are no bytes
            if (sendPosition > 0) {
                dataPort.write(sendBuffer, sendPosition);
                stats.registerWrite(sendPosition);
            }
        }
        finally {
            writerLock.unlock();
        }
    }

    @Override
    public void run() {
        run(running);
    }

    private void run(AtomicBoolean runFlag) {
        long outgoingTimeoutMillis = 2 * 60 * 1000L; // 2 minutes; can be long since no one is sending
        long reconnectTimeoutMillis = 1L; // This should be short, since we are trying to get the reconnect through

        try {
            dataPort = this.dataPortFuture.get(); // Will wait for the future to complete
            StatisticsCollector stats = this.connection.getStatisticsCollector();

            while (runFlag.get() && !Thread.interrupted()) {
                NatsMessage msg;
                if (mode.get() == Mode.Normal) {
                    msg = this.normalOutgoing.accumulate(sendBufferLength.get(), MAX_MESSAGES_IN_NETWORK_BUFFER, outgoingTimeoutMillis);
                }
                else {
                    msg = this.reconnectOutgoing.accumulate(sendBufferLength.get(), MAX_MESSAGES_IN_NETWORK_BUFFER, reconnectTimeoutMillis);
                }
                if (msg != null) {
                    sendMessageBatch(msg, dataPort, stats);
                }
            }
        } catch (IOException | BufferOverflowException io) {
            // if already not running, an IOE is not unreasonable in a transition state
            if (runFlag.get()) {
                this.connection.handleCommunicationIssue(io);
            }
        } catch (CancellationException | ExecutionException ex) {
            // Exit
        } catch (InterruptedException ex) {
            // Exit
            Thread.currentThread().interrupt();
        } finally {
            runFlag.set(false);
        }
    }

    void enterReconnectMode() {
        reconnectOutgoing.clear();
        mode.set(Mode.Reconnect);
    }

    void enterWaitingForEndReconnectMode() {
        mode.set(Mode.WaitingForEndReconnect);
        reconnectOutgoing.queueMarkerMessage(END_RECONNECT);
    }

    boolean canQueueDuringReconnect(NatsMessage msg) {
        // don't over fill the "send" buffer while waiting to reconnect
        return (reconnectBufferSize < 0 || (normalOutgoing.sizeInBytes() + msg.getSizeInBytes()) < reconnectBufferSize);
    }

    boolean queue(NatsMessage msg) {
        return this.normalOutgoing.push(msg);
    }

    void queueInternalMessage(NatsMessage msg) {
        if (mode.get() == Mode.Reconnect) {
            reconnectOutgoing.push(msg);
        }
        else {
            normalOutgoing.push(msg, true);
        }
    }

    void flushBuffer() {
        // Since there is no connection level locking, we rely on synchronization
        // of the APIs here.
        writerLock.lock();
        try {
            if (this.running.get()) {
                dataPort.flush();
            }
        } catch (Exception e) {
            // NOOP;
        }
        finally {
            writerLock.unlock();
        }
    }

    long outgoingPendingMessageCount() {
        return normalOutgoing.length();
    }

    long outgoingPendingBytes() {
        return normalOutgoing.sizeInBytes();
    }
}

1
2	// Copyright 2015-2026 The NATS Authors
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at:
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package io.synadia.client.impl;
16
17	import io.synadia.client.Options;
18	import io.synadia.client.StatisticsCollector;
19	import io.synadia.client.utils.ByteArrayBuilder;
20
21	import java.io.IOException;
22	import java.nio.BufferOverflowException;
23	import java.util.concurrent.CancellationException;
24	import java.util.concurrent.CompletableFuture;
25	import java.util.concurrent.ExecutionException;
26	import java.util.concurrent.Future;
27	import java.util.concurrent.atomic.AtomicBoolean;
28	import java.util.concurrent.atomic.AtomicInteger;
29	import java.util.concurrent.atomic.AtomicReference;
30	import java.util.concurrent.locks.ReentrantLock;
31
32	import static io.synadia.client.OptionsConstants.MAX_MESSAGES_IN_NETWORK_BUFFER;
33	import static io.synadia.client.impl.MarkerMessage.END_RECONNECT;
34	import static io.synadia.client.utils.ByteArrayBuilder.bufferAllocSize;
35	import static io.synadia.client.utils.NatsConstants.CR;
36	import static io.synadia.client.utils.NatsConstants.LF;
37
38	/**
39	* The connection's writer thread. It drains the outgoing message queues into the data port,
40	* switching between the normal queue and the reconnect queue as the connection state changes.
41	*/
42	public class NatsConnectionWriter implements Runnable {
43	enum Mode {	3 ✔
44	Normal, Reconnect, WaitingForEndReconnect	18 ✔
45	}
46	private static final int BUFFER_BLOCK_SIZE = 256;
47
48	private final NatsConnection connection;
49
50	private final ReentrantLock writerLock;
51	private Future<Boolean> stopped;
52	private Future<DataPort> dataPortFuture;
53	private DataPort dataPort;
54	// A new flag per start(). The thread started by start() checks and clears only the flag it was
55	// started with, so a late thread from a previous socket cannot stop or disturb the current one.
56	private volatile AtomicBoolean running;
57	private final AtomicReference<Mode> mode;
58	private final ReentrantLock startStopLock;
59
60	private byte[] sendBuffer;
61	private final AtomicInteger sendBufferLength;
62
63	private final WriterMessageQueue normalOutgoing;
64	private final WriterMessageQueue reconnectOutgoing;
65	private final long reconnectBufferSize;
66
67	NatsConnectionWriter(NatsConnection connection) {	2 ✔
68	this.connection = connection;	3 ✔
69	writerLock = new ReentrantLock();	5 ✔
70
71	this.running = new AtomicBoolean(false);	6 ✔
72	mode = new AtomicReference<>(Mode.Normal);	6 ✔
73	this.startStopLock = new ReentrantLock();	5 ✔
74	this.stopped = new CompletableFuture<>();	5 ✔
75	((CompletableFuture<Boolean>)this.stopped).complete(Boolean.TRUE); // we are stopped on creation	6 ✔
76
77	Options options = connection.getOptions();	3 ✔
78	int sbl = bufferAllocSize(options.getBufferSize(), BUFFER_BLOCK_SIZE);	5 ✔
79	sendBufferLength = new AtomicInteger(sbl);	6 ✔
80	sendBuffer = new byte[sbl];	4 ✔
81	reconnectBufferSize = options.getReconnectBufferSize();	4 ✔
82
83	normalOutgoing = new WriterMessageQueue(	4 ✔
84	options.getMaxMessagesInOutgoingQueue(),	2 ✔
85	options.isDiscardMessagesWhenOutgoingQueueFull(),	2 ✔
86	options.getWriteQueuePushTimeout()	3 ✔
87	);
88
89	// The "reconnect" buffer contains internal messages
90	// Using this constructor makes it unbounded and without "discard when full"
91	reconnectOutgoing = new WriterMessageQueue(options.getWriteQueuePushTimeout());	7 ✔
92	}	1 ✔
93
94	// Should only be called if the current thread has exited.
95	// Use the Future from stop() to determine if it is ok to call this.
96	// This method resets that future so mistiming can result in badness.
97	void start(Future<DataPort> dataPortFuture) {
98	this.startStopLock.lock();	3 ✔
99	try {
100	this.dataPortFuture = dataPortFuture;	3 ✔
101	AtomicBoolean runFlag = new AtomicBoolean(true);	5 ✔
102	this.running = runFlag;	3 ✔
103	this.normalOutgoing.resume();	3 ✔
104	this.reconnectOutgoing.resume();	3 ✔
105	this.stopped = connection.getWriterExecutor().submit(() -> run(runFlag), Boolean.TRUE);	14 ✔
106	} finally {
107	this.startStopLock.unlock();	3 ✔
108	}
109	}	1 ✔
110
111	// May be called several times on an error.
112	// Returns a future that is completed when the thread completes, not when this
113	// method does.
114	Future<Boolean> stop() {
115	AtomicBoolean runFlag = running;	3 ✔
116	if (runFlag.get()) {	3 ✔
117	runFlag.set(false);	3 ✔
118	startStopLock.lock();	3 ✔
119	try {
120	this.normalOutgoing.pause();	3 ✔
121	this.reconnectOutgoing.pause();	3 ✔
122	this.normalOutgoing.filter();	3 ✔
123	}
124	finally {
125	this.startStopLock.unlock();	3 ✔
126	}
127	}
128	return this.stopped;	3 ✔
129	}
130
131	boolean isRunning() {
132	return running.get();	4 ✔
133	}
134
135	void sendMessageBatch(NatsMessage msg, DataPort dataPort, StatisticsCollector stats) throws IOException {
136	writerLock.lock();	3 ✔
137	try {
138	int sendPosition = 0;	2 ✔
139	int sbl = sendBufferLength.get();	4 ✔
140
141	while (msg != null) {	2 ✔
142	if (msg == END_RECONNECT) {	3 ✔
143	mode.set(Mode.Normal);	4 ✔
144	break;	1 ✔
145	}
146	long size = msg.getSizeInBytes();	3 ✔
147
148	if (sendPosition + size > sbl) {	8 !
UNCOV 149	if (sendPosition > 0) {	×
150	dataPort.write(sendBuffer, sendPosition);	×
151	stats.registerWrite(sendPosition);	×
152	sendPosition = 0;	×
153	}
UNCOV 154	if (size > sbl) { // have to resize b/c can't fit 1 message	×
UNCOV 155	sbl = bufferAllocSize((int) size, BUFFER_BLOCK_SIZE);	×
UNCOV 156	sendBufferLength.set(sbl);	×
UNCOV 157	sendBuffer = new byte[sbl];	×
158	}
159	}
160
161	ByteArrayBuilder bab = msg.getProtocolBab();	3 ✔
162	int babLen = bab.length();	3 ✔
163	System.arraycopy(bab.internalArray(), 0, sendBuffer, sendPosition, babLen);	8 ✔
164	sendPosition += babLen;	4 ✔
165
166	sendBuffer[sendPosition++] = CR;	6 ✔
167	sendBuffer[sendPosition++] = LF;	6 ✔
168
169	if (!msg.isProtocol()) { // because a protocol message does not have headers or data	3 ✔
170	sendPosition += msg.copyNotEmptyHeaders(sendPosition, sendBuffer);	8 ✔
171
172	byte[] bytes = msg.getData(); // guaranteed to not be null	3 ✔
173	if (bytes.length > 0) {	3 !
174	System.arraycopy(bytes, 0, sendBuffer, sendPosition, bytes.length);	8 ✔
175	sendPosition += bytes.length;	5 ✔
176	}
177
178	sendBuffer[sendPosition++] = CR;	6 ✔
179	sendBuffer[sendPosition++] = LF;	6 ✔
180	}
181
182	stats.incrementOut(size);	3 ✔
183
184	if (msg.flushImmediatelyAfterPublish) {	3 !
UNCOV 185	dataPort.write(sendBuffer, sendPosition);	×
UNCOV 186	dataPort.flush();	×
UNCOV 187	stats.registerWrite(sendPosition);	×
UNCOV 188	sendPosition = 0;	×
189	}
190
191	msg = msg.next;	3 ✔
192	}	1 ✔
193
194	// no need to write if there are no bytes
195	if (sendPosition > 0) {	2 !
196	dataPort.write(sendBuffer, sendPosition);	5 ✔
197	stats.registerWrite(sendPosition);	4 ✔
198	}
199	}
200	finally {
201	writerLock.unlock();	3 ✔
202	}
203	}	1 ✔
204
205	@Override
206	public void run() {
NEW 207	run(running);	×
NEW 208	}	×
209
210	private void run(AtomicBoolean runFlag) {
211	long outgoingTimeoutMillis = 2 * 60 * 1000L; // 2 minutes; can be long since no one is sending	2 ✔
212	long reconnectTimeoutMillis = 1L; // This should be short, since we are trying to get the reconnect through	2 ✔
213
214	try {
215	dataPort = this.dataPortFuture.get(); // Will wait for the future to complete	6 ✔
216	StatisticsCollector stats = this.connection.getStatisticsCollector();	4 ✔
217
218	while (runFlag.get() && !Thread.interrupted()) {	5 !
219	NatsMessage msg;
220	if (mode.get() == Mode.Normal) {	5 ✔
221	msg = this.normalOutgoing.accumulate(sendBufferLength.get(), MAX_MESSAGES_IN_NETWORK_BUFFER, outgoingTimeoutMillis);	12 ✔
222	}
223	else {
224	msg = this.reconnectOutgoing.accumulate(sendBufferLength.get(), MAX_MESSAGES_IN_NETWORK_BUFFER, reconnectTimeoutMillis);	11 ✔
225	}
226	if (msg != null) {	2 ✔
227	sendMessageBatch(msg, dataPort, stats);	6 ✔
228	}
229	}	1 ✔
UNCOV 230	} catch (IOException \| BufferOverflowException io) {	×
231	// if already not running, an IOE is not unreasonable in a transition state
NEW 232	if (runFlag.get()) {	×
UNCOV 233	this.connection.handleCommunicationIssue(io);	×
234	}
235	} catch (CancellationException \| ExecutionException ex) {	×
236	// Exit
237	} catch (InterruptedException ex) {	×
238	// Exit
239	Thread.currentThread().interrupt();	×
240	} finally {
241	runFlag.set(false);	3 ✔
242	}
243	}	1 ✔
244
245	void enterReconnectMode() {
246	reconnectOutgoing.clear();	3 ✔
247	mode.set(Mode.Reconnect);	4 ✔
248	}	1 ✔
249
250	void enterWaitingForEndReconnectMode() {
251	mode.set(Mode.WaitingForEndReconnect);	4 ✔
252	reconnectOutgoing.queueMarkerMessage(END_RECONNECT);	4 ✔
253	}	1 ✔
254
255	boolean canQueueDuringReconnect(NatsMessage msg) {
256	// don't over fill the "send" buffer while waiting to reconnect
UNCOV 257	return (reconnectBufferSize < 0 \|\| (normalOutgoing.sizeInBytes() + msg.getSizeInBytes()) < reconnectBufferSize);	×
258	}
259
260	boolean queue(NatsMessage msg) {
261	return this.normalOutgoing.push(msg);	5 ✔
262	}
263
264	void queueInternalMessage(NatsMessage msg) {
265	if (mode.get() == Mode.Reconnect) {	5 ✔
266	reconnectOutgoing.push(msg);	6 ✔
267	}
268	else {
269	normalOutgoing.push(msg, true);	6 ✔
270	}
271	}	1 ✔
272
273	void flushBuffer() {
274	// Since there is no connection level locking, we rely on synchronization
275	// of the APIs here.
UNCOV 276	writerLock.lock();	×
277	try {
UNCOV 278	if (this.running.get()) {	×
UNCOV 279	dataPort.flush();	×
280	}
281	} catch (Exception e) {	×
282	// NOOP;
283	}
284	finally {
UNCOV 285	writerLock.unlock();	×
286	}
UNCOV 287	}	×
288
289	long outgoingPendingMessageCount() {
NEW 290	return normalOutgoing.length();	×
291	}
292
293	long outgoingPendingBytes() {
NEW 294	return normalOutgoing.sizeInBytes();	×
295	}
296	}

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