Refactor telemetry client

doc/changes/changes_0.2.0.md

@@ -0,0 +1,10 @@
+# Telemetry Java 0.2.0, released 2026-??-??
+
+Code name:
+
+## Summary
+
+## Features
+
+* ISSUE_NUMBER: description
+

pom.xml

@@ -1,11 +1,9 @@
 <?xml version="1.0" encoding="UTF-8" standalone="no"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.exasol</groupId>
     <artifactId>telemetry-java</artifactId>
-    <version>0.1.0</version>
+    <version>0.2.0</version>
     <name>telemetry-java</name>
     <description>Minimal, zero-dependency telemetry library for Java applications.</description>
     <url>https://github.com/exasol/telemetry-java/</url>
@@ -114,7 +112,7 @@
     <parent>
         <artifactId>telemetry-java-generated-parent</artifactId>
         <groupId>com.exasol</groupId>
-        <version>0.1.0</version>
+        <version>0.2.0</version>
         <relativePath>pk_generated_parent.pom</relativePath>
     </parent>
 </project>

src/main/java/com/exasol/telemetry/AsyncTelemetryClient.java

@@ -0,0 +1,171 @@
+package com.exasol.telemetry;
+
+import static java.util.Objects.requireNonNull;
+
+import java.time.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.logging.Logger;
+
+final class AsyncTelemetryClient implements TelemetryClient {
+    private static final Logger LOGGER = Logger.getLogger(AsyncTelemetryClient.class.getName());
+
+    private final TelemetryConfig config;
+    private final BlockingQueue<TelemetryEvent> queue;
+    private final HttpTransport transport;
+    private final Thread senderThread;
+    private final Clock clock;
+    private final CountDownLatch terminated = new CountDownLatch(1);
+    private volatile boolean closed;
+
+    AsyncTelemetryClient(final TelemetryConfig config) {
+        this(config, Clock.systemUTC());
+    }
+
+    AsyncTelemetryClient(final TelemetryConfig config, final Clock clock) {
+        this.config = requireNonNull(config, "config");
+        this.clock = requireNonNull(clock, "clock");
+        this.queue = new ArrayBlockingQueue<>(config.getQueueCapacity());
+        this.transport = new HttpTransport(config);
+        this.senderThread = new Thread(this::runSender, "telemetry-java-sender");
+        this.senderThread.setDaemon(true);
+        this.senderThread.start();
+    }
+
+    @Override
+    public void track(final String feature) {
+        if (closed || feature == null) {
+            return;
+        }
+        final TelemetryEvent event = new TelemetryEvent(feature, clock.instant());
+        enqueue(event);
+    }
+
+    @SuppressWarnings("java:S899") // Intentionally ignore return value of offer() to avoid blocking the caller.
+    private void enqueue(final TelemetryEvent event) {
+        queue.offer(event);
+    }
+
+    private void runSender() {
+        try {
+            while (!closed || !queue.isEmpty()) {
+                final TelemetryEvent event = queue.poll(100, TimeUnit.MILLISECONDS);
+                if (event != null) {
+                    sendWithRetry(drainBatch(event));
+                }
+            }
+        } catch (final InterruptedException ignored) {
+            Thread.currentThread().interrupt();
+        } finally {
+            terminated.countDown();
+        }
+    }
+
+    private List<TelemetryEvent> drainBatch(final TelemetryEvent firstEvent) {
+        final List<TelemetryEvent> batch = new ArrayList<>();
+        batch.add(firstEvent);
+        queue.drainTo(batch);
+        return batch;
+    }
+
+    // [impl~telemetry-client-send-with-retry~1->req~async-delivery~1]
+    private void sendWithRetry(final List<TelemetryEvent> events) {
+        final Instant start = clock.instant();
+        final Message message = Message.fromEvents(config.getProjectTag(), config.getProductVersion(), start, events);
+        final Instant deadline = start.plus(config.getRetryTimeout());
+        Duration delay = config.getInitialRetryDelay();
+
+        while (true) {
+            if (Thread.currentThread().isInterrupted()) {
+                return;
+            }
+            try {
+                transport.send(message);
+                LOGGER.fine(() -> "Telemetry sent to the server with " + events.size() + " event(s).");
+                return;
+            } catch (final Exception exception) {
+                LOGGER.fine(() -> "Telemetry sending failed for " + events.size() + " event(s): "
+                        + rootCauseMessage(exception));
+                if (Thread.currentThread().isInterrupted()) {
+                    return;
+                }
+                final Instant now = clock.instant();
+                if (!now.isBefore(deadline)) {
+                    return;
+                }
+                final Duration remaining = Duration.between(now, deadline);
+                sleep(min(delay, remaining));
+                delay = min(delay.multipliedBy(2), config.getMaxRetryDelay());
+            }
+        }
+    }
+
+    private static Duration min(final Duration left, final Duration right) {
+        return left.compareTo(right) <= 0 ? left : right;
+    }
+
+    private static String rootCauseMessage(final Throwable throwable) {
+        Throwable cause = throwable;
+        while (cause != null) {
+            if (cause instanceof HttpException) {
+                final HttpException httpException = (HttpException) cause;
+                return "server status " + httpException.getStatusCode() + " (" + httpException.getServerStatus() + ")";
+            }
+            if (cause.getCause() == null) {
+                final String message = cause.getMessage();
+                if (message == null || message.isBlank()) {
+                    return cause.getClass().getSimpleName();
+                }
+            }
+            cause = cause.getCause();
+        }
+        return "";
+    }
+
+    private void sleep(final Duration duration) {
+        try {
+            Thread.sleep(Math.max(1, duration.toMillis()));
+        } catch (final InterruptedException ignored) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    // Visible for testing
+    boolean awaitStopped(final Duration timeout) throws InterruptedException {
+        return terminated.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    // Visible for testing
+    boolean isRunning() {
+        return senderThread.isAlive();
+    }
+
+    @Override
+    public void close() {
+        if (closed) {
+            return;
+        }
+        closed = true;
+        awaitSenderStop();
+        LOGGER.fine("Telemetry is stopped.");
+    }
+
+    private void awaitSenderStop() {
+        final long timeoutNanos = config.getRetryTimeout().toNanos();
+        final long deadlineNanos = System.nanoTime() + timeoutNanos;
+        try {
+            while (senderThread.isAlive()) {
+                final long remainingNanos = deadlineNanos - System.nanoTime();
+                if (remainingNanos <= 0) {
+                    senderThread.interrupt();
+                    senderThread.join();
+                    return;
+                }
+                TimeUnit.NANOSECONDS.timedJoin(senderThread, remainingNanos);
+            }
+        } catch (final InterruptedException ignored) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}

src/main/java/com/exasol/telemetry/NoOpTelemetryClient.java

@@ -0,0 +1,17 @@
+package com.exasol.telemetry;
+
+final class NoOpTelemetryClient implements TelemetryClient {
+    NoOpTelemetryClient() {
+        // Intentionally empty.
+    }
+
+    @Override
+    public void track(final String feature) {
+        // Intentionally does nothing.
+    }
+
+    @Override
+    public void close() {
+        // Intentionally does nothing.
+    }
+}