Write Listener (Rev 4)

Author: scottf

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

@@ -20,10 +20,7 @@
 import java.io.IOException;
 import java.nio.BufferOverflowException;
 import java.time.Duration;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -177,8 +174,7 @@ void sendMessageBatch(NatsMessage msg, DataPort dataPort, StatisticsCollector st
 
                 stats.incrementOut(size);
                 if (writeListener != null) {
-                    NatsMessage finalMsg = msg;
-                    writeListener.submit(() -> writeListener.buffered(finalMsg, mode.get().name()));
+                    writeListener.buffered(msg, mode.get().name());
                 }
 
                 if (msg.flushImmediatelyAfterPublish) {
@@ -206,7 +202,12 @@ public void run() {
         try {
             dataPort = this.dataPortFuture.get(); // Will wait for the future to complete
             StatisticsCollector stats = connection.getStatisticsCollector();
-            WriteListener writeListener = connection.getOptions().getWriteListener();
+            WriteListener writeListener = null;
+            WriteListener userWriteListener = connection.getOptions().getWriteListener();
+            if (userWriteListener != null) {
+                ExecutorService executor = connection.getOptions().getExecutor();
+                writeListener = (m, o) -> executor.submit(() -> userWriteListener.buffered(m, o));
+            }
 
             while (running.get() && !Thread.interrupted()) {
                 NatsMessage msg;

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

@@ -14,27 +14,7 @@
 package io.nats.client.impl;
 
 import io.nats.client.Message;
-import org.jspecify.annotations.Nullable;
 
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-public abstract class WriteListener {
-    public final ExecutorService executorService;
-
-    public WriteListener() {
-        this(null);
-    }
-
-    public WriteListener(@Nullable ExecutorService executorService) {
-        this.executorService = executorService == null
-            ? Executors.newSingleThreadExecutor()
-            : executorService;
-    }
-
-    public final void submit(Runnable runnable) {
-        executorService.submit(runnable);
-    }
-
-    public abstract void buffered(Message msg, String mode);
+public interface WriteListener {
+    void buffered(Message msg, String mode);
 }