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

55.56

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

// Copyright (c) 2026 Synadia Communications Inc. All Rights Reserved.

package io.synadia.client.impl;

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

import java.io.IOException;

/**
 * A {@link DataPort} that serves some already-read bytes before reading from the real port.
 * {@link NatsConnectionV3} reads the connect handshake itself; anything the server sent after the
 * handshake's PONG in the same read is handed to the reader through this.
 */
final class PrefixedDataPort implements DataPort {
    private final DataPort port;
    private byte[] prefix;
    private int prefixPos;

    PrefixedDataPort(DataPort port, byte[] prefix) {
        this.port = port;
        this.prefix = prefix;
    }

    @Override
    public void connect(@NonNull NatsConnection conn, @NonNull NatsUri uri, long timeoutNanos) throws IOException {
        port.connect(conn, uri, timeoutNanos);
    }

    @Override
    public void afterConstruct(@NonNull Options options) {
        port.afterConstruct(options);
    }

    @Override
    public void upgradeToSecure() throws IOException {
        port.upgradeToSecure();
    }

    @Override
    public int read(byte[] dst, int off, int len) throws IOException {
        byte[] p = prefix;
        if (p != null) {
            int available = p.length - prefixPos;
            int n = available < len ? available : len;
            System.arraycopy(p, prefixPos, dst, off, n);
            prefixPos += n;
            if (prefixPos == p.length) {
                prefix = null;
            }
            return n;
        }
        return port.read(dst, off, len);
    }

    @Override
    public void write(byte[] src, int toWrite) throws IOException {
        port.write(src, toWrite);
    }

    @Override
    public void shutdownInput() throws IOException {
        port.shutdownInput();
    }

    @Override
    public void close() throws IOException {
        port.close();
    }

    @Override
    public void forceClose() throws IOException {
        port.forceClose();
    }

    @Override
    public void flush() throws IOException {
        port.flush();
    }
}

1	// Copyright (c) 2026 Synadia Communications Inc. All Rights Reserved.
2
3	package io.synadia.client.impl;
4
5	import io.synadia.client.Options;
6	import io.synadia.client.utils.NatsUri;
7	import org.jspecify.annotations.NonNull;
8
9	import java.io.IOException;
10
11	/**
12	* A {@link DataPort} that serves some already-read bytes before reading from the real port.
13	* {@link NatsConnectionV3} reads the connect handshake itself; anything the server sent after the
14	* handshake's PONG in the same read is handed to the reader through this.
15	*/
16	final class PrefixedDataPort implements DataPort {
17	private final DataPort port;
18	private byte[] prefix;
19	private int prefixPos;
20
21	PrefixedDataPort(DataPort port, byte[] prefix) {	2 ✔
22	this.port = port;	3 ✔
23	this.prefix = prefix;	3 ✔
24	}	1 ✔
25
26	@Override
27	public void connect(@NonNull NatsConnection conn, @NonNull NatsUri uri, long timeoutNanos) throws IOException {
NEW 28	port.connect(conn, uri, timeoutNanos);	×
NEW 29	}	×
30
31	@Override
32	public void afterConstruct(@NonNull Options options) {
NEW 33	port.afterConstruct(options);	×
NEW 34	}	×
35
36	@Override
37	public void upgradeToSecure() throws IOException {
NEW 38	port.upgradeToSecure();	×
NEW 39	}	×
40
41	@Override
42	public int read(byte[] dst, int off, int len) throws IOException {
43	byte[] p = prefix;	3 ✔
44	if (p != null) {	2 ✔
45	int available = p.length - prefixPos;	6 ✔
46	int n = available < len ? available : len;	7 ✔
47	System.arraycopy(p, prefixPos, dst, off, n);	7 ✔
48	prefixPos += n;	6 ✔
49	if (prefixPos == p.length) {	5 ✔
50	prefix = null;	3 ✔
51	}
52	return n;	2 ✔
53	}
54	return port.read(dst, off, len);	7 ✔
55	}
56
57	@Override
58	public void write(byte[] src, int toWrite) throws IOException {
NEW 59	port.write(src, toWrite);	×
NEW 60	}	×
61
62	@Override
63	public void shutdownInput() throws IOException {
NEW 64	port.shutdownInput();	×
NEW 65	}	×
66
67	@Override
68	public void close() throws IOException {
NEW 69	port.close();	×
NEW 70	}	×
71
72	@Override
73	public void forceClose() throws IOException {
NEW 74	port.forceClose();	×
NEW 75	}	×
76
77	@Override
78	public void flush() throws IOException {
NEW 79	port.flush();	×
NEW 80	}	×
81	}

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