From abc2a60ab04223357b8f22f904f1643ae410925d Mon Sep 17 00:00:00 2001 From: Daniel McCoy Stephenson Date: Tue, 27 Oct 2026 22:54:43 -0600 Subject: [PATCH] Coalesce repeated events so no program can flood trace (0.7.0) One server whose /power command was dispatched every ~100 ms by another plugin sent ~425k command events in five days, one per run. The client now coalesces: the first report of an event (name, value and tags) is sent at once and opens a 60 s window; identical reports inside it are counted and sent as one event when it closes, tagged coalesced= and with value= when the event had none. A window that sent a count opens the next, so a steady flood is one event a minute; close() sends every open count. startup is never coalesced. At most 256 events are counted at once, so memory stays bounded. The sender becomes a single-thread ScheduledThreadPoolExecutor so the same daemon thread closes windows; the 256-report queue bound is kept by counting queued sends. The consumer API is unchanged; new public constants make this a minor release. Co-Authored-By: Claude Opus 5.5 --- README.md | 40 ++- pom.xml | 2 +- .../stephenson/trace/TraceClient.java | 239 +++++++++++++++++- .../stephenson/trace/TraceClientTest.java | 211 +++++++++++++++- 4 files changed, 477 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 5a12971..d3943a4 100644 --- a/README.md +++ b/README.md @@ -138,8 +138,46 @@ generates, reads or writes an ID. | **Returns immediately** | The HTTP call runs on one daemon thread the client owns. A Spigot plugin can report from the server thread and no tick waits on the network. | | **Never throws** | A server that is down, slow, or rejecting the key is a dropped report, not an exception in your program. Drops are logged at `FINE` if you gave a logger, otherwise not at all. | | **Bounded** | At most 256 reports wait to be sent; past that, new ones are dropped without a log line, even at `FINE`. A trace server that is unreachable for a week costs a few kilobytes, not your heap. | +| **Throttled** | Repeats of one event within a minute are counted, not sent: the first goes out at once, the rest as one event when the minute is up. See [below](#repeated-events-are-coalesced). | | **`close()` drains** | Reports already queued get up to the client timeout (5 s total) to be sent before the thread stops, so a CLI that reports and exits at once does not lose its event. Still bounded: an unreachable server delays exit by at most the timeout. | +## Repeated events are coalesced + +Since 0.7.0, a program cannot flood the trace server, however often it +reports. Before, one server whose `/power` command was being run every +100 ms — another plugin dispatching it — sent hundreds of thousands of +`command` events in a few days, one per run. + +- An event is its name, value and tags (once `version`, `install` and any + server-wide tags are added; tag order does not matter). The first report of + an event is sent at once and opens a **60-second window**. +- Identical reports inside the window are only counted. When the window + closes, a count above zero is sent as **one more event**: the same name + and tags, plus the tag `coalesced` with the count — and the count as its + `value` when the event had no value. An event that has a value keeps it, + since it stands for that many more of the same value. +- A window that sent a count opens the next one, so an event repeated + without pause costs one report a minute. A window that counted nothing + just ends. +- `close()` sends every count still open, inside the same 5-second bound. +- `startup` is never coalesced or held back. +- At most 256 distinct events are counted at once. A new event while 256 are + being counted is dropped without a log line, like a report that finds the + queue full, so memory stays bounded even for a program that reports a new + tag every call. + +So the number of times something happened is the sum, over stored events, +of `value` for events tagged `coalesced` and 1 for the rest — for a valueless +event like `command`, simply `SUM(COALESCE(value, 1))`. Counting rows still +works for "was it used, by how many servers"; it now undercounts "how many +times", by design. + +A minute is long enough that a command dispatched every tick is one event a +minute instead of 1,200, and short enough that a person trying a command +sees it on the trace server at once (the first report is never delayed) and +the count soon after. The server buckets by day, so nothing it shows is +finer than that anyway. + ## Opting out Reporting is **opt-out**, and the person running the program always has the @@ -213,7 +251,7 @@ plugins already vendor bStats' `Metrics.java`. com.github.Stephenson-Software trace-client-java - 0.6.1 + 0.7.0 ``` diff --git a/pom.xml b/pom.xml index 8f2542d..cb292dd 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ software.stephenson trace-client - 0.6.1 + 0.7.0 jar trace-client diff --git a/src/main/java/software/stephenson/trace/TraceClient.java b/src/main/java/software/stephenson/trace/TraceClient.java index 6a56831..b894591 100644 --- a/src/main/java/software/stephenson/trace/TraceClient.java +++ b/src/main/java/software/stephenson/trace/TraceClient.java @@ -1,5 +1,5 @@ /* - * trace-client 0.6.1 -- https://github.com/Stephenson-Software/trace-client-java + * trace-client 0.7.0 -- https://github.com/Stephenson-Software/trace-client-java * * One call to report that a program was used. Copy this file into a project as * is, or depend on the artifact; either way there is nothing else to add. @@ -20,11 +20,14 @@ import java.nio.file.Path; import java.nio.file.StandardOpenOption; import java.util.Collections; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.TreeMap; import java.util.UUID; -import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -52,6 +55,13 @@ * without a log line, even at {@link Level#FINE}. A trace server * that is unreachable for a week costs a few kilobytes, not the * host's heap. + *
  • It is throttled. Repeats of one event -- same name, value and + * tags -- within {@value #COALESCE_WINDOW_MS} ms are counted, not sent: + * the first goes out at once, the rest as one event when the window + * closes, carrying how many there were (see {@link #report(String, + * Double, Map)}). A program, or another plugin dispatching its + * command in a loop, cannot flood the trace server. {@code startup} + * is never held back.
    • * * *

    Reporting is opt-out, and the person running the program always has the @@ -125,11 +135,35 @@ public final class TraceClient { /** This client's version, as sent in the User-Agent. */ - public static final String VERSION = "0.6.1"; + public static final String VERSION = "0.7.0"; /** How many reports may wait to be sent before new ones are dropped. */ public static final int QUEUE_CAPACITY = 256; + /** + * How long repeats of one event are counted rather than sent. A minute: + * long enough that a command dispatched every tick, or every 100 ms by + * another plugin, is one event a minute instead of hundreds; short + * enough that a person using a command a few times sees it on the trace + * server within a minute, and that close() seldom has much to flush. + * The server buckets by day, so nothing it shows is finer than this. + */ + public static final long COALESCE_WINDOW_MS = 60_000; + + /** + * How many distinct events may be counted at once. An event that is not + * already being counted while this many are is dropped, without a log + * line -- like a report that finds the queue full -- so a program + * reporting a new name or tag every call costs bounded memory too. + */ + public static final int MAX_COALESCED_EVENTS = 256; + + /** The tag a coalesced event carries how many repeats it stands for in. */ + public static final String COALESCED_TAG = "coalesced"; + + /** The event that is never coalesced: a start is never held back. */ + static final String STARTUP = "startup"; + private static final int CONNECT_TIMEOUT_MS = 5_000; private static final int READ_TIMEOUT_MS = 5_000; @@ -206,7 +240,17 @@ public final class TraceClient { private final String disabledReason; // null when enabled private final Map serverWideTags; // never null; read once, at build() private final String installId; // null when disabled or when there is none - private final ThreadPoolExecutor executor; // null when disabled + private final long coalesceWindowNanos; + private final ScheduledThreadPoolExecutor executor; // null when disabled + + // The rest is guarded by lock: report() runs on the host's threads, + // flushes run on the sender thread. + private final Object lock = new Object(); + // Events being counted, by coalesceKey(), oldest first. + private final Map windows = new LinkedHashMap<>(); + private int queued; // sends waiting for the sender thread, not yet started + private ScheduledFuture flushTimer; // the next flushDue(), or null + private boolean closed; private TraceClient(Builder builder) { this.endpoint = builder.baseUrl.replaceAll("/+$", "") + "/api/metrics"; @@ -222,17 +266,21 @@ private TraceClient(Builder builder) { // After the opt-outs, never before: a disabled client neither makes // up an ID nor writes one to disk. this.installId = disabledReason == null ? resolveInstallId(builder, serverWide) : null; + this.coalesceWindowNanos = TimeUnit.MILLISECONDS.toNanos(builder.coalesceWindowMs); if (disabledReason == null) { - this.executor = new ThreadPoolExecutor( - 1, 1, 30, TimeUnit.SECONDS, - new ArrayBlockingQueue(QUEUE_CAPACITY), + // Scheduled, so the one thread that sends also closes windows: + // no second thread. Its queue is unbounded; enqueue() is the + // bound, and at most one flush is ever scheduled beside the sends. + this.executor = new ScheduledThreadPoolExecutor(1, runnable -> { Thread thread = new Thread(runnable, "trace-client/" + application); thread.setDaemon(true); return thread; }, new ThreadPoolExecutor.DiscardPolicy()); + this.executor.setKeepAliveTime(30, TimeUnit.SECONDS); this.executor.allowCoreThreadTimeOut(true); + this.executor.setRemoveOnCancelPolicy(true); } else { this.executor = null; } @@ -710,19 +758,169 @@ public void report(String name) { * Reports that {@code name} happened, with an optional numeric value and * optional string tags. Returns immediately; see the class comment. A * {@code null} or blank name sends nothing. + * + *

    Repeats are coalesced. The first report of an event -- its name, + * value and tags, once {@code version}, {@code install} and the + * server-wide tags are added -- is sent at once and opens a window of + * {@value #COALESCE_WINDOW_MS} ms. Identical reports inside the window are + * only counted. When it closes, a count above zero is sent as one more + * event: the same name and tags plus the tag {@value #COALESCED_TAG} with + * the count, and the count as its {@code value} when the event had none + * (an event with a value keeps it: it stands for that many more of the + * same value). A window that sent a count opens the next one, so an + * event repeated without pause costs one report a minute; one that + * counted nothing ends. {@link #close()} sends every count still open. + * + *

    So the number of times an event happened is the sum, over the + * events the server stored, of {@code value} where the + * {@value #COALESCED_TAG} tag is present and 1 where it is not -- for + * a valueless event, simply {@code COALESCE(value, 1)}. + * + *

    {@code startup} is never coalesced, nor held back: every one is + * sent at once. */ public void report(String name, Double value, Map tags) { if (executor == null || name == null || name.trim().isEmpty()) { return; } - final String body = json(application, name, value, - withServerWideTags(withInstall(withVersion(tags, version), installId), serverWideTags)); - executor.execute(() -> send(body)); + // json() leaves out a value that is NaN or infinite, so they are the + // same event as no value at all. + Double finite = value == null || value.isNaN() || value.isInfinite() ? null : value; + Map merged = + withServerWideTags(withInstall(withVersion(tags, version), installId), serverWideTags); + String body = json(application, name, finite, merged); + synchronized (lock) { + if (closed) { + return; + } + if (!STARTUP.equals(name)) { + String key = json(application, name, finite, new TreeMap<>(merged)); // tag order is not identity + Window window = windows.get(key); + if (window != null) { + window.count++; + return; + } + if (windows.size() >= MAX_COALESCED_EVENTS) { + return; // bounded, like a full queue: dropped without a log line + } + windows.put(key, new Window(name, finite, merged, System.nanoTime() + coalesceWindowNanos)); + scheduleFlush(); + } + enqueue(body); + } + } + + /** One event being counted: what to send if it repeats, and until when. */ + private static final class Window { + final String name; + final Double value; // null, or finite + final Map tags; + long deadline; // System.nanoTime() at which the window closes + long count; // repeats since the window opened + + Window(String name, Double value, Map tags, long deadline) { + this.name = name; + this.value = value; + this.tags = tags; + this.deadline = deadline; + } + } + + /** + * Queues {@code body} for the sender thread, unless + * {@value #QUEUE_CAPACITY} sends are already waiting. Caller holds + * {@code lock}. + */ + private void enqueue(final String body) { + if (queued >= QUEUE_CAPACITY) { + return; + } + queued++; + executor.execute(() -> { + synchronized (lock) { + queued--; + } + send(body); + }); + } + + /** Schedules flushDue() for the earliest window, if none is scheduled. Caller holds {@code lock}. */ + private void scheduleFlush() { + if (flushTimer != null || windows.isEmpty()) { + return; + } + // Windows are opened and reopened in deadline order, but a scan is + // cheap and does not depend on it. + long now = System.nanoTime(); + long wait = Long.MAX_VALUE; + for (Window window : windows.values()) { + wait = Math.min(wait, window.deadline - now); + } + flushTimer = executor.schedule(this::flushDue, Math.max(0, wait), TimeUnit.NANOSECONDS); + } + + /** Runs on the sender thread when the earliest window closes. */ + private void flushDue() { + synchronized (lock) { + flushTimer = null; + if (closed) { + return; + } + flush(false); + scheduleFlush(); + } + } + + /** + * Sends a count for every window that has closed -- every window at all + * when {@code everything} -- and ends or reopens it. Caller holds + * {@code lock}. + */ + private void flush(boolean everything) { + long now = System.nanoTime(); + for (Iterator it = windows.values().iterator(); it.hasNext(); ) { + Window window = it.next(); + if (!everything && now - window.deadline < 0) { + continue; + } + if (window.count > 0) { + enqueue(json(application, window.name, + window.value != null ? window.value : Double.valueOf(window.count), + withCount(window.tags, window.count))); + } + if (everything || window.count == 0) { + it.remove(); + } else { + window.count = 0; + window.deadline = now + coalesceWindowNanos; + } + } } /** - * Stops the sending thread, giving reports already queued up to - * {@value #READ_TIMEOUT_MS} ms in total to be sent first. A program that + * The tags plus {@value #COALESCED_TAG} with the count, unless they + * already carry one or are at {@link #MAX_TAGS}. A copy. + */ + static Map withCount(Map tags, long count) { + Map merged = new LinkedHashMap<>(tags); + if (!merged.containsKey(COALESCED_TAG) && merged.size() < MAX_TAGS) { + merged.put(COALESCED_TAG, String.valueOf(count)); + } + return merged; + } + + /** How many distinct events are being counted. For tests. */ + int coalescing() { + synchronized (lock) { + return windows.size(); + } + } + + /** + * Sends every count still being coalesced (see {@link #report(String, + * Double, Map)}), then stops the sending thread, giving reports already + * queued up to {@value #READ_TIMEOUT_MS} ms in total to be sent first. + * A program that * reports and then exits within milliseconds -- a CLI -- would otherwise * lose its one event to the race between queueing it and the thread * picking it up. The bound still holds: an unreachable server delays exit @@ -733,6 +931,16 @@ public void close() { if (executor == null) { return; } + synchronized (lock) { + if (!closed) { + closed = true; + flush(true); + if (flushTimer != null) { + flushTimer.cancel(false); // or shutdown() would wait for it + flushTimer = null; + } + } + } executor.shutdown(); // no new work; queued reports still run try { if (!executor.awaitTermination(READ_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { @@ -860,6 +1068,7 @@ public static final class Builder { private String installId; private File installIdFile; private Logger logger; + private long coalesceWindowMs = COALESCE_WINDOW_MS; private Builder(String baseUrl, String application, String version) { if (baseUrl == null || baseUrl.trim().isEmpty()) { @@ -975,6 +1184,12 @@ public Builder logger(Logger logger) { return this; } + /** A shorter coalescing window, so tests need not wait a minute. Not public: the window is the client's. */ + Builder coalesceWindowMillis(long millis) { + this.coalesceWindowMs = millis; + return this; + } + /** * Builds the client. The environment variables * {@value TraceClient#ENV_USAGE_REPORTING} and diff --git a/src/test/java/software/stephenson/trace/TraceClientTest.java b/src/test/java/software/stephenson/trace/TraceClientTest.java index 536d510..43d8f33 100644 --- a/src/test/java/software/stephenson/trace/TraceClientTest.java +++ b/src/test/java/software/stephenson/trace/TraceClientTest.java @@ -524,8 +524,10 @@ void queue_isBoundedAndDropsRatherThanGrows() throws Exception { int flood = TraceClient.QUEUE_CAPACITY * 3; // Act + // startup, because it is never coalesced: every one of these goes + // to the queue, so the queue is all that bounds them. for (int i = 0; i < flood; i++) { - client.report("flood"); + client.report("startup"); } release.countDown(); long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20); @@ -1890,6 +1892,213 @@ private static Map tagsOf(String content) { return TraceClient.parseServerWideConfig(Arrays.asList(content.split("\n", -1))).tags; } + /** Waits until the shared server has received {@code count} reports, or 5 s pass. */ + private void awaitReceived(int count) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (received.size() < count && System.nanoTime() < deadline) { + Thread.sleep(20); + } + } + + @Test + void coalesce_sendsTheFirstAtOnceAndTheRepeatsAsOneCountedEventWhenTheWindowCloses() throws Exception { + // Arrange + // The flood that prompted this: another plugin dispatching /power + // every 100 ms, each one a command event. + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k") + .coalesceWindowMillis(1_000).build(); + Map power = Collections.singletonMap("name", "power"); + + // Act + for (int i = 0; i < 50; i++) { + client.report("command", null, power); + } + + // Assert + assertTrue(arrived.await(5, TimeUnit.SECONDS), "the first report goes out at once"); + Thread.sleep(200); + assertEquals(1, received.size(), "the repeats wait for the window to close"); + awaitReceived(2); + assertEquals(2, received.size(), "one event for all 49 repeats"); + assertEquals("{\"application\":\"MyPlugin\",\"name\":\"command\",\"tags\":{\"name\":\"power\",\"version\":\"1.2.3\"}}", + received.get(0).body); + assertEquals("{\"application\":\"MyPlugin\",\"name\":\"command\",\"value\":49.0," + + "\"tags\":{\"name\":\"power\",\"version\":\"1.2.3\",\"coalesced\":\"49\"}}", + received.get(1).body); + client.close(); + assertEquals(2, received.size(), "nothing was left to flush"); + } + + @Test + void coalesce_aSteadyFloodIsOneEventPerWindowAndAQuietWindowEnds() throws Exception { + // Arrange + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k") + .coalesceWindowMillis(300).build(); + + // Act + // About four windows of a report every 10 ms, then quiet. + long until = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(1_200); + int reports = 0; + while (System.nanoTime() < until) { + client.report("command", null, Collections.singletonMap("name", "power")); + reports++; + Thread.sleep(10); + } + long quiet = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (client.coalescing() > 0 && System.nanoTime() < quiet) { + Thread.sleep(20); + } + + // Assert + assertEquals(0, client.coalescing(), "a window that counted nothing ends"); + int sent = received.size(); + assertTrue(sent >= 3 && sent <= 7, "about one event per 300 ms window, but sent " + sent + " for " + reports); + long total = 1; + for (Received request : received.subList(1, sent)) { + java.util.regex.Matcher count = java.util.regex.Pattern.compile("\"coalesced\":\"(\\d+)\"").matcher(request.body); + assertTrue(count.find(), "every event after the first is a count: " + request.body); + total += Long.parseLong(count.group(1)); + } + assertEquals(reports, total, "the counts add up to every report"); + client.close(); + assertEquals(sent, received.size(), "nothing was left to flush"); + } + + @Test + void close_sendsEveryCountStillOpen() throws Exception { + // Arrange + // The real window, a minute: only close() can send these counts. + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k").build(); + for (int i = 0; i < 10; i++) { + client.report("command", null, Collections.singletonMap("name", "home")); + } + for (int i = 0; i < 3; i++) { + client.report("command", null, Collections.singletonMap("name", "spawn")); + } + client.report("command", null, Collections.singletonMap("name", "once")); + + // Act + long before = System.nanoTime(); + client.close(); + long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - before); + + // Assert + assertTrue(elapsedMs < 5_000, "close() must not wait for the window: took " + elapsedMs + " ms"); + List bodies = received.stream().map(r -> r.body).collect(Collectors.toList()); + assertEquals(5, bodies.size(), "three first reports and two counts: " + bodies); + assertTrue(bodies.contains("{\"application\":\"MyPlugin\",\"name\":\"command\",\"value\":9.0," + + "\"tags\":{\"name\":\"home\",\"version\":\"1.2.3\",\"coalesced\":\"9\"}}"), bodies.toString()); + assertTrue(bodies.contains("{\"application\":\"MyPlugin\",\"name\":\"command\",\"value\":2.0," + + "\"tags\":{\"name\":\"spawn\",\"version\":\"1.2.3\",\"coalesced\":\"2\"}}"), bodies.toString()); + assertEquals(1, bodies.stream().filter(b -> b.contains("\"once\"")).count(), "nothing to count, nothing more sent"); + } + + @Test + void coalesce_keysOnNameValueAndTagsButNotTagOrder() throws Exception { + // Arrange + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k").build(); + Map ab = new LinkedHashMap<>(); + ab.put("a", "1"); + ab.put("b", "2"); + Map ba = new LinkedHashMap<>(); + ba.put("b", "2"); + ba.put("a", "1"); + + // Act + client.report("command", null, ab); + client.report("command", null, ba); // the same event + client.report("command", 2.5, ab); // a different value: a different event + client.report("command", 2.5, ab); + client.report("command", 2.5, ab); + client.report("other", null, ab); // a different name + client.report("command", null, Collections.singletonMap("a", "1")); // different tags + client.report("command", Double.NaN, ab); // NaN is sent as no value: the first event again + awaitReceived(4); + Thread.sleep(200); + int sentAtOnce = received.size(); + client.close(); + + // Assert + assertEquals(4, sentAtOnce, "one first report per distinct event"); + List bodies = received.stream().map(r -> r.body).collect(Collectors.toList()); + assertEquals(6, bodies.size(), bodies.toString()); + assertTrue(bodies.contains("{\"application\":\"MyPlugin\",\"name\":\"command\",\"value\":2.0," + + "\"tags\":{\"a\":\"1\",\"b\":\"2\",\"version\":\"1.2.3\",\"coalesced\":\"2\"}}"), bodies.toString()); + assertTrue(bodies.contains("{\"application\":\"MyPlugin\",\"name\":\"command\",\"value\":2.5," + + "\"tags\":{\"a\":\"1\",\"b\":\"2\",\"version\":\"1.2.3\",\"coalesced\":\"2\"}}"), + "an event with a value keeps it; the count is the tag: " + bodies); + } + + @Test + void coalesce_neverHoldsBackStartup() throws Exception { + // Arrange + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k").build(); + + // Act + for (int i = 0; i < 5; i++) { + client.report("startup"); + } + awaitReceived(5); + + // Assert + assertEquals(5, received.size(), "every startup goes out at once"); + assertEquals(0, client.coalescing(), "startup is never counted"); + client.close(); + assertEquals(5, received.size()); + } + + @Test + void coalesce_countsAtMostMaxCoalescedEventsAndDropsNewOnesBeyond() throws Exception { + // Arrange + // Nothing listening, so the sends fail fast and the test is about + // memory alone: a new name every call must not grow without bound. + TraceClient client = TraceClient.builder("http://127.0.0.1:1", "MyPlugin", "1.2.3").key("k").build(); + + // Act + for (int i = 0; i < TraceClient.MAX_COALESCED_EVENTS * 4; i++) { + client.report("command", null, Collections.singletonMap("name", "c" + i)); + client.report("command", null, Collections.singletonMap("name", "c" + i)); + } + + // Assert + assertEquals(TraceClient.MAX_COALESCED_EVENTS, client.coalescing()); + assertDoesNotThrow(() -> client.report("startup"), "startup still goes to the queue"); + client.close(); + assertEquals(0, client.coalescing(), "close() lets go of every count"); + } + + @Test + void coalesce_afterCloseNothingIsCountedOrSent() throws Exception { + // Arrange + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k").build(); + client.close(); + + // Act + client.report("command"); + client.report("command"); + + // Assert + assertEquals(0, client.coalescing()); + assertFalse(arrived.await(300, TimeUnit.MILLISECONDS), "nothing should be sent after close()"); + } + + @Test + void withCount_addsTheCountOnlyWhenThereIsRoomAndNoneAlready() { + Map tags = new LinkedHashMap<>(); + tags.put("version", "1"); + assertEquals("7", TraceClient.withCount(tags, 7).get(TraceClient.COALESCED_TAG)); + assertFalse(tags.containsKey(TraceClient.COALESCED_TAG), "the caller's map is never modified"); + + Map own = Collections.singletonMap(TraceClient.COALESCED_TAG, "mine"); + assertEquals("mine", TraceClient.withCount(own, 7).get(TraceClient.COALESCED_TAG)); + + Map full = new LinkedHashMap<>(); + for (int i = 0; i < TraceClient.MAX_TAGS; i++) { + full.put("t" + i, "v"); + } + assertEquals(TraceClient.MAX_TAGS, TraceClient.withCount(full, 7).size(), "never past the server's tag limit"); + } + private static byte[] readAll(InputStream in) throws java.io.IOException { ByteArrayOutputStream out = new ByteArrayOutputStream(); byte[] buffer = new byte[1024];