Rework reference counting

Author: kabir

server-common/src/main/java/io/a2a/server/events/EventConsumer.java

@@ -74,14 +74,18 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
             // Track event count to ensure we only count down when the FINAL event's write completes,
             // not when earlier events' pending callbacks fire
             tube.whenRequested(requested -> {
-                int ackCount = eventsAcknowledged.incrementAndGet();
+                // Only count acknowledgments for actual events, not initial subscription requests
+                // The initial onSubscribe() triggers 2 callbacks before any events are sent
                 int sent = eventsSent.get();
-                boolean waiting = waitingForFinalEventWrite.get();
-                LOGGER.debug("whenRequested callback: requested={}, ack={}, sent={}, waiting={}",
-                    requested, ackCount, sent, waiting);
-                if (waiting && ackCount >= sent) {
-                    LOGGER.debug("Subscriber acknowledged final event ({}/{}) - async write completed", ackCount, sent);
-                    finalEventWriteLatch.countDown();
+                if (sent > 0) {
+                    int ackCount = eventsAcknowledged.incrementAndGet();
+                    boolean waiting = waitingForFinalEventWrite.get();
+                    LOGGER.debug("whenRequested callback: requested={}, ack={}, sent={}, waiting={}",
+                        requested, ackCount, sent, waiting);
+                    if (waiting && ackCount >= sent) {
+                        LOGGER.debug("Subscriber acknowledged final event ({}/{}) - async write completed", ackCount, sent);
+                        finalEventWriteLatch.countDown();
+                    }
                 }
             });