#2101

Committed 12 Aug 2025 11:26AM UTC coverage: 95.457% (+0.02%) from 95.433%

Build # #2101

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1387 from nats-io/info-nullability

Ensuring nullability contracts

Run Details

92 of 92 new or added lines in 10 files covered. (100.0%)

108 existing lines in 12 files now uncovered.

11913 of 12480 relevant lines covered (95.46%)

0.95 hits per line

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

98.51

/src/main/java/io/nats/client/impl/NatsStatistics.java

// Copyright 2015-2018 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.nats.client.impl;

import io.nats.client.StatisticsCollector;

import java.text.NumberFormat;
import java.util.LongSummaryStatistics;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

class NatsStatistics implements StatisticsCollector {
    private final ReentrantLock readStatsLock;
    private final ReentrantLock writeStatsLock;

    private final LongSummaryStatistics readStats;
    private final LongSummaryStatistics writeStats;

    private final AtomicLong flushCounter;
    private final AtomicLong outstandingRequests;
    private final AtomicLong requestsSent;
    private final AtomicLong repliesReceived;
    private final AtomicLong duplicateRepliesReceived;
    private final AtomicLong orphanRepliesReceived;
    private final AtomicLong reconnects;
    private final AtomicLong inMsgs;
    private final AtomicLong outMsgs;
    private final AtomicLong inBytes;
    private final AtomicLong outBytes;
    private final AtomicLong pingCount;
    private final AtomicLong okCount;
    private final AtomicLong errCount;
    private final AtomicLong exceptionCount;
    private final AtomicLong droppedCount;

    private boolean trackAdvanced;

    public NatsStatistics() {
        this.readStatsLock = new ReentrantLock();
        this.writeStatsLock = new ReentrantLock();

        this.readStats = new LongSummaryStatistics();
        this.writeStats = new LongSummaryStatistics();

        this.flushCounter = new AtomicLong();
        this.outstandingRequests = new AtomicLong();
        this.requestsSent = new AtomicLong();
        this.repliesReceived = new AtomicLong();
        this.duplicateRepliesReceived = new AtomicLong();
        this.orphanRepliesReceived = new AtomicLong();
        this.reconnects = new AtomicLong();
        this.inMsgs = new AtomicLong();
        this.outMsgs = new AtomicLong();
        this.inBytes = new AtomicLong();
        this.outBytes = new AtomicLong();
        this.pingCount = new AtomicLong();
        this.okCount = new AtomicLong();
        this.errCount = new AtomicLong();
        this.exceptionCount = new AtomicLong();
        this.droppedCount = new AtomicLong();
    }

    @Override
    public void setAdvancedTracking(boolean trackAdvanced) {
        this.trackAdvanced = trackAdvanced;
    }

    @Override
    public void incrementPingCount() {
        this.pingCount.incrementAndGet();
    }

    @Override
    public void incrementDroppedCount() {
        this.droppedCount.incrementAndGet();
    }

    @Override
    public void incrementOkCount() {
        this.okCount.incrementAndGet();
    }

    @Override
    public void incrementErrCount() {
        this.errCount.incrementAndGet();
    }

    @Override
    public void incrementExceptionCount() {
        this.exceptionCount.incrementAndGet();
    }

    @Override
    public void incrementRequestsSent() {
        this.requestsSent.incrementAndGet();
    }

    @Override
    public void incrementRepliesReceived() {
        this.repliesReceived.incrementAndGet();
    }

    @Override
    public void incrementDuplicateRepliesReceived() {
        this.duplicateRepliesReceived.incrementAndGet();
    }

    @Override
    public void incrementOrphanRepliesReceived() {
        this.orphanRepliesReceived.incrementAndGet();
    }

    @Override
    public void incrementReconnects() {
        this.reconnects.incrementAndGet();
    }

    @Override
    public void incrementInMsgs() {
        this.inMsgs.incrementAndGet();
    }

    @Override
    public void incrementOutMsgs() {
        this.outMsgs.incrementAndGet();
    }

    @Override
    public void incrementInBytes(long bytes) {
        this.inBytes.addAndGet(bytes);
    }

    @Override
    public void incrementOutBytes(long bytes) {
        this.outBytes.addAndGet(bytes);
    }

    @Override
    public void incrementFlushCounter() {
        this.flushCounter.incrementAndGet();
    }

    @Override
    public void incrementOutstandingRequests() {
        this.outstandingRequests.incrementAndGet();
    }

    @Override
    public void decrementOutstandingRequests() {
        this.outstandingRequests.decrementAndGet();
    }

    @Override
    public void registerRead(long bytes) {
        if (!trackAdvanced) {
            return;
        }

        readStatsLock.lock();
        try {
            readStats.accept(bytes);
        } finally {
            readStatsLock.unlock();
        }
    }

    @Override
    public void registerWrite(long bytes) {
        if (!trackAdvanced) {
            return;
        }

        writeStatsLock.lock();
        try {
            writeStats.accept(bytes);
        } finally {
            writeStatsLock.unlock();
        }
    }

    @Override
    public long getPings() {
        return this.pingCount.get();
    }

    @Override
    public long getDroppedCount() {
        return this.droppedCount.get();
    }

    @Override
    public long getOKs() {
        return this.okCount.get();
    }

    @Override
    public long getErrs() {
        return this.errCount.get();
    }

    @Override
    public long getExceptions() {
        return this.exceptionCount.get();
    }

    @Override
    public long getRequestsSent() {
        return this.requestsSent.get();
    }

    @Override
    public long getReconnects() {
        return this.reconnects.get();
    }

    @Override
    public long getInMsgs() {
        return this.inMsgs.get();
    }

    @Override
    public long getOutMsgs() {
        return this.outMsgs.get();
    }

    @Override
    public long getInBytes() {
        return this.inBytes.get();
    }

    @Override
    public long getOutBytes() {
        return this.outBytes.get();
    }

    @Override
    public long getFlushCounter() {
        return flushCounter.get();
    }

    @Override
    public long getOutstandingRequests() {
        return outstandingRequests.get();
    }

    @Override
    public long getRepliesReceived() { return repliesReceived.get(); }

    @Override
    public long getDuplicateRepliesReceived() {
        return duplicateRepliesReceived.get();
    }

    @Override
    public long getOrphanRepliesReceived() { return orphanRepliesReceived.get(); }

    void appendNumberStat(StringBuilder builder, String name, long value) {
        builder.append(name);
        builder.append(NumberFormat.getNumberInstance().format(value));
        builder.append("\n");
    }

    void appendNumberStat(StringBuilder builder, String name, double value) {
        builder.append(name);
        builder.append(NumberFormat.getNumberInstance().format(value));
        builder.append("\n");
    }

    public String toString() {
        StringBuilder builder = new StringBuilder();

        builder.append("### Connection ###\n");
        appendNumberStat(builder, "Reconnects:                      ", this.reconnects.get());
        appendNumberStat(builder, "Requests Sent:                   ", this.requestsSent.get());
        appendNumberStat(builder, "Replies Received:                ", this.repliesReceived.get());
        if (this.trackAdvanced) {
            appendNumberStat(builder, "Duplicate Replies Received:      ", this.duplicateRepliesReceived.get());
            appendNumberStat(builder, "Orphan Replies Received:         ", this.orphanRepliesReceived.get());
        }
        appendNumberStat(builder, "Pings Sent:                      ", this.pingCount.get());
        appendNumberStat(builder, "+OKs Received:                   ", this.okCount.get());
        appendNumberStat(builder, "-Errs Received:                  ", this.errCount.get());
        appendNumberStat(builder, "Handled Exceptions:              ", this.exceptionCount.get());
        appendNumberStat(builder, "Successful Flush Calls:          ", this.flushCounter.get());
        appendNumberStat(builder, "Outstanding Request Futures:     ", this.outstandingRequests.get());
        appendNumberStat(builder, "Dropped Messages:                ", this.droppedCount.get());
        builder.append("\n");
        builder.append("### Reader ###\n");
        appendNumberStat(builder, "Messages in:                     ", this.inMsgs.get());
        appendNumberStat(builder, "Bytes in:                        ", this.inBytes.get());
        builder.append("\n");
        if (this.trackAdvanced) {
            readStatsLock.lock();
            try {
                appendNumberStat(builder, "Socket Reads:                    ", readStats.getCount());
                appendNumberStat(builder, "Average Bytes Per Read:          ", readStats.getAverage());
                appendNumberStat(builder, "Min Bytes Per Read:              ", readStats.getMin());
                appendNumberStat(builder, "Max Bytes Per Read:              ", readStats.getMax());
            } finally {
                readStatsLock.unlock();
            }
        }
        builder.append("\n");
        builder.append("### Writer ###\n");
        appendNumberStat(builder, "Messages out:                    ", this.outMsgs.get());
        appendNumberStat(builder, "Bytes out:                       ", this.outBytes.get());
        builder.append("\n");
        if (this.trackAdvanced) {
            writeStatsLock.lock();
            try {
                appendNumberStat(builder, "Socket Writes:                   ", writeStats.getCount());
                appendNumberStat(builder, "Average Bytes Per Write:         ", writeStats.getAverage());
                appendNumberStat(builder, "Min Bytes Per Write:             ", writeStats.getMin());
                appendNumberStat(builder, "Max Bytes Per Write:             ", writeStats.getMax());
            } finally {
                writeStatsLock.unlock();
            }
        }

        return builder.toString();
    }
}

1	// Copyright 2015-2018 The NATS Authors
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the License.
4	// You may obtain a copy of the License at:
5	//
6	// http://www.apache.org/licenses/LICENSE-2.0
7	//
8	// Unless required by applicable law or agreed to in writing, software
9	// distributed under the License is distributed on an "AS IS" BASIS,
10	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11	// See the License for the specific language governing permissions and
12	// limitations under the License.
13
14	package io.nats.client.impl;
15
16	import io.nats.client.StatisticsCollector;
17
18	import java.text.NumberFormat;
19	import java.util.LongSummaryStatistics;
20	import java.util.concurrent.atomic.AtomicLong;
21	import java.util.concurrent.locks.ReentrantLock;
22
23	class NatsStatistics implements StatisticsCollector {
24	private final ReentrantLock readStatsLock;
25	private final ReentrantLock writeStatsLock;
26
27	private final LongSummaryStatistics readStats;
28	private final LongSummaryStatistics writeStats;
29
30	private final AtomicLong flushCounter;
31	private final AtomicLong outstandingRequests;
32	private final AtomicLong requestsSent;
33	private final AtomicLong repliesReceived;
34	private final AtomicLong duplicateRepliesReceived;
35	private final AtomicLong orphanRepliesReceived;
36	private final AtomicLong reconnects;
37	private final AtomicLong inMsgs;
38	private final AtomicLong outMsgs;
39	private final AtomicLong inBytes;
40	private final AtomicLong outBytes;
41	private final AtomicLong pingCount;
42	private final AtomicLong okCount;
43	private final AtomicLong errCount;
44	private final AtomicLong exceptionCount;
45	private final AtomicLong droppedCount;
46
47	private boolean trackAdvanced;
48
49	public NatsStatistics() {	1✔
50	this.readStatsLock = new ReentrantLock();	1✔
51	this.writeStatsLock = new ReentrantLock();	1✔
52
53	this.readStats = new LongSummaryStatistics();	1✔
54	this.writeStats = new LongSummaryStatistics();	1✔
55
56	this.flushCounter = new AtomicLong();	1✔
57	this.outstandingRequests = new AtomicLong();	1✔
58	this.requestsSent = new AtomicLong();	1✔
59	this.repliesReceived = new AtomicLong();	1✔
60	this.duplicateRepliesReceived = new AtomicLong();	1✔
61	this.orphanRepliesReceived = new AtomicLong();	1✔
62	this.reconnects = new AtomicLong();	1✔
63	this.inMsgs = new AtomicLong();	1✔
64	this.outMsgs = new AtomicLong();	1✔
65	this.inBytes = new AtomicLong();	1✔
66	this.outBytes = new AtomicLong();	1✔
67	this.pingCount = new AtomicLong();	1✔
68	this.okCount = new AtomicLong();	1✔
69	this.errCount = new AtomicLong();	1✔
70	this.exceptionCount = new AtomicLong();	1✔
71	this.droppedCount = new AtomicLong();	1✔
72	}	1✔
73
74	@Override
75	public void setAdvancedTracking(boolean trackAdvanced) {
76	this.trackAdvanced = trackAdvanced;	1✔
77	}	1✔
78
79	@Override
80	public void incrementPingCount() {
81	this.pingCount.incrementAndGet();	1✔
82	}	1✔
83
84	@Override
85	public void incrementDroppedCount() {
86	this.droppedCount.incrementAndGet();	1✔
87	}	1✔
88
89	@Override
90	public void incrementOkCount() {
91	this.okCount.incrementAndGet();	1✔
92	}	1✔
93
94	@Override
95	public void incrementErrCount() {
96	this.errCount.incrementAndGet();	1✔
97	}	1✔
98
99	@Override
100	public void incrementExceptionCount() {
101	this.exceptionCount.incrementAndGet();	1✔
102	}	1✔
103
104	@Override
105	public void incrementRequestsSent() {
106	this.requestsSent.incrementAndGet();	1✔
107	}	1✔
108
109	@Override
110	public void incrementRepliesReceived() {
111	this.repliesReceived.incrementAndGet();	1✔
112	}	1✔
113
114	@Override
115	public void incrementDuplicateRepliesReceived() {
116	this.duplicateRepliesReceived.incrementAndGet();	1✔
117	}	1✔
118
119	@Override
120	public void incrementOrphanRepliesReceived() {
121	this.orphanRepliesReceived.incrementAndGet();	1✔
122	}	1✔
123
124	@Override
125	public void incrementReconnects() {
126	this.reconnects.incrementAndGet();	1✔
127	}	1✔
128
129	@Override
130	public void incrementInMsgs() {
131	this.inMsgs.incrementAndGet();	1✔
132	}	1✔
133
134	@Override
135	public void incrementOutMsgs() {
136	this.outMsgs.incrementAndGet();	1✔
137	}	1✔
138
139	@Override
140	public void incrementInBytes(long bytes) {
141	this.inBytes.addAndGet(bytes);	1✔
142	}	1✔
143
144	@Override
145	public void incrementOutBytes(long bytes) {
146	this.outBytes.addAndGet(bytes);	1✔
147	}	1✔
148
149	@Override
150	public void incrementFlushCounter() {
151	this.flushCounter.incrementAndGet();	1✔
152	}	1✔
153
154	@Override
155	public void incrementOutstandingRequests() {
156	this.outstandingRequests.incrementAndGet();	1✔
157	}	1✔
158
159	@Override
160	public void decrementOutstandingRequests() {
161	this.outstandingRequests.decrementAndGet();	1✔
162	}	1✔
163
164	@Override
165	public void registerRead(long bytes) {
166	if (!trackAdvanced) {	1✔
167	return;	1✔
168	}
169
170	readStatsLock.lock();	1✔
171	try {
172	readStats.accept(bytes);	1✔
173	} finally {
174	readStatsLock.unlock();	1✔
175	}
176	}	1✔
177
178	@Override
179	public void registerWrite(long bytes) {
180	if (!trackAdvanced) {	1✔
181	return;	1✔
182	}
183
184	writeStatsLock.lock();	1✔
185	try {
186	writeStats.accept(bytes);	1✔
187	} finally {
188	writeStatsLock.unlock();	1✔
189	}
190	}	1✔
191
192	@Override
193	public long getPings() {
194	return this.pingCount.get();	1✔
195	}
196
197	@Override
198	public long getDroppedCount() {
199	return this.droppedCount.get();	1✔
200	}
201
202	@Override
203	public long getOKs() {
204	return this.okCount.get();	1✔
205	}
206
207	@Override
208	public long getErrs() {
UNCOV 209	return this.errCount.get();	×
210	}
211
212	@Override
213	public long getExceptions() {
214	return this.exceptionCount.get();	1✔
215	}
216
217	@Override
218	public long getRequestsSent() {
UNCOV 219	return this.requestsSent.get();	×
220	}
221
222	@Override
223	public long getReconnects() {
224	return this.reconnects.get();	1✔
225	}
226
227	@Override
228	public long getInMsgs() {
229	return this.inMsgs.get();	1✔
230	}
231
232	@Override
233	public long getOutMsgs() {
234	return this.outMsgs.get();	1✔
235	}
236
237	@Override
238	public long getInBytes() {
239	return this.inBytes.get();	1✔
240	}
241
242	@Override
243	public long getOutBytes() {
244	return this.outBytes.get();	1✔
245	}
246
247	@Override
248	public long getFlushCounter() {
249	return flushCounter.get();	1✔
250	}
251
252	@Override
253	public long getOutstandingRequests() {
254	return outstandingRequests.get();	1✔
255	}
256
257	@Override
258	public long getRepliesReceived() { return repliesReceived.get(); }	1✔
259
260	@Override
261	public long getDuplicateRepliesReceived() {
262	return duplicateRepliesReceived.get();	1✔
263	}
264
265	@Override
266	public long getOrphanRepliesReceived() { return orphanRepliesReceived.get(); }	1✔
267
268	void appendNumberStat(StringBuilder builder, String name, long value) {
269	builder.append(name);	1✔
270	builder.append(NumberFormat.getNumberInstance().format(value));	1✔
271	builder.append("\n");	1✔
272	}	1✔
273
274	void appendNumberStat(StringBuilder builder, String name, double value) {
275	builder.append(name);	1✔
276	builder.append(NumberFormat.getNumberInstance().format(value));	1✔
277	builder.append("\n");	1✔
278	}	1✔
279
280	public String toString() {
281	StringBuilder builder = new StringBuilder();	1✔
282
283	builder.append("### Connection ###\n");	1✔
284	appendNumberStat(builder, "Reconnects: ", this.reconnects.get());	1✔
285	appendNumberStat(builder, "Requests Sent: ", this.requestsSent.get());	1✔
286	appendNumberStat(builder, "Replies Received: ", this.repliesReceived.get());	1✔
287	if (this.trackAdvanced) {	1✔
288	appendNumberStat(builder, "Duplicate Replies Received: ", this.duplicateRepliesReceived.get());	1✔
289	appendNumberStat(builder, "Orphan Replies Received: ", this.orphanRepliesReceived.get());	1✔
290	}
291	appendNumberStat(builder, "Pings Sent: ", this.pingCount.get());	1✔
292	appendNumberStat(builder, "+OKs Received: ", this.okCount.get());	1✔
293	appendNumberStat(builder, "-Errs Received: ", this.errCount.get());	1✔
294	appendNumberStat(builder, "Handled Exceptions: ", this.exceptionCount.get());	1✔
295	appendNumberStat(builder, "Successful Flush Calls: ", this.flushCounter.get());	1✔
296	appendNumberStat(builder, "Outstanding Request Futures: ", this.outstandingRequests.get());	1✔
297	appendNumberStat(builder, "Dropped Messages: ", this.droppedCount.get());	1✔
298	builder.append("\n");	1✔
299	builder.append("### Reader ###\n");	1✔
300	appendNumberStat(builder, "Messages in: ", this.inMsgs.get());	1✔
301	appendNumberStat(builder, "Bytes in: ", this.inBytes.get());	1✔
302	builder.append("\n");	1✔
303	if (this.trackAdvanced) {	1✔
304	readStatsLock.lock();	1✔
305	try {
306	appendNumberStat(builder, "Socket Reads: ", readStats.getCount());	1✔
307	appendNumberStat(builder, "Average Bytes Per Read: ", readStats.getAverage());	1✔
308	appendNumberStat(builder, "Min Bytes Per Read: ", readStats.getMin());	1✔
309	appendNumberStat(builder, "Max Bytes Per Read: ", readStats.getMax());	1✔
310	} finally {
311	readStatsLock.unlock();	1✔
312	}
313	}
314	builder.append("\n");	1✔
315	builder.append("### Writer ###\n");	1✔
316	appendNumberStat(builder, "Messages out: ", this.outMsgs.get());	1✔
317	appendNumberStat(builder, "Bytes out: ", this.outBytes.get());	1✔
318	builder.append("\n");	1✔
319	if (this.trackAdvanced) {	1✔
320	writeStatsLock.lock();	1✔
321	try {
322	appendNumberStat(builder, "Socket Writes: ", writeStats.getCount());	1✔
323	appendNumberStat(builder, "Average Bytes Per Write: ", writeStats.getAverage());	1✔
324	appendNumberStat(builder, "Min Bytes Per Write: ", writeStats.getMin());	1✔
325	appendNumberStat(builder, "Max Bytes Per Write: ", writeStats.getMax());	1✔
326	} finally {
327	writeStatsLock.unlock();	1✔
328	}
329	}
330
331	return builder.toString();	1✔
332	}
333	}

nats-io / nats.java / #2101

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