Drop the connection when a command never gets through

The server handles commands in order and holds back everything sent
after a lost one, so giving up on a command now ends the connection, as
in the official client. A connection lost without either side saying
goodbye (a timeout, or such a command) is now reported to listeners as
a disconnect ("connection lost").

- A command's acknowledgement is awaited in the background: ensureSent
  is often called from the packet handler thread, which must keep
  handling while resends go on. A command that fails makes way for the
  next one.
- One packet that fails to handle no longer ends the handler thread,
  which left everything after it unhandled while ACKs and pings kept the
  connection looking alive.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java

@@ -558,8 +558,13 @@ public abstract class AbstractTeamspeakClientSocket
                 } catch (TimeoutException e) {
                     if (!response.willResend() || response.getResends() >= ResendTiming.MAX_RESENDS
                             || System.nanoTime() - response.getFirstSent()
-                                    >= TimeUnit.MILLISECONDS.toNanos(ResendTiming.GIVE_UP_MS))
+                                    >= TimeUnit.MILLISECONDS.toNanos(ResendTiming.GIVE_UP_MS)) {
+                        // Commands are handled in order: the server holds back everything sent after this one
+                        // until it arrives, so giving up on it loses the connection, as in the official client.
+                        if (getState() == ClientConnectionState.CONNECTED)
+                            setState(ClientConnectionState.DISCONNECTED);
                         throw new TimeoutException("send");
+                    }
                     response.resend();
                 } catch (CancellationException e) {
                     throw e;
@@ -984,26 +989,28 @@ public abstract class AbstractTeamspeakClientSocket
     private class NetworkHandler implements Runnable {
         @Override
         public void run() {
-            try {
             while (isReading()) {
+                Packet packet;
                 try {
-                        Packet packet = readQueue.take();
-
-                        // separate these so we getHandler at proper runtime instant rather than get it and let it change
-                        // on us
-
-                        getHandler().handlePacket(packet);
+                    packet = readQueue.take();
                 } catch (InterruptedException e) {
                     break;
                 }
-                }
+
+                // One bad packet must not end this thread: everything received after it would queue up unhandled
+                // while ACKs and pings, answered by the reader, keep the connection looking alive.
+                try {
+                    // separate these so we getHandler at proper runtime instant rather than get it and let it change
+                    // on us
+                    getHandler().handlePacket(packet);
                 } catch (Throwable e) {
-                Ts3Debugging.debug("Problem reading packet", e);
+                    Ts3Debugging.debug("Problem handling packet", e);
                     // A disconnect interrupts whatever the handler was waiting on; that is not an error.
                     if (isReading()) getExceptionHandler().accept(e);
                 }
             }
         }
+    }
 
     private class NetworkReader implements Runnable {
         @Override

src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java

@@ -66,6 +66,9 @@ public class LocalTeamspeakClientSocket
      * <p>
      * get
      */
+    /** Whether losing the connection now would go unreported: connected, and neither side has said goodbye. */
+    private volatile boolean dropUnannounced;
+
     private Map<Integer, ClientCommandResponse> awaitingCommands = new ConcurrentHashMap<>();
 
     public LocalTeamspeakClientSocket() {
@@ -90,6 +93,7 @@ public class LocalTeamspeakClientSocket
                     throws CommandProcessException {
                 int clientId = Integer.parseInt(command.get("clid").getValue());
                 if (clientId == getClientId()) {
+                    dropUnannounced = false; // this is the announcement
                     try {
                         setState(ClientConnectionState.DISCONNECTED);
                     } catch (Exception e) {
@@ -144,7 +148,7 @@ public class LocalTeamspeakClientSocket
                 response.add(new CommandSingleParameter("connection_server2client_packetloss_total", Integer.toString(0)));
 
                 try {
-                    writePacket(new PacketBody2Command(getRole().getOut(), response));
+                    writePacketAsync(new PacketBody2Command(getRole().getOut(), response));
                 } catch (Exception e) {
                     Ts3Debugging.debug("Problem sending setconnectioninfo", e);
                 }
@@ -283,6 +287,7 @@ public class LocalTeamspeakClientSocket
     @Override
     protected void onConnect() {
         super.onConnect();
+        dropUnannounced = true;
 
         if (microphoneThread != null && !microphoneThread.isAlive())
             microphoneThread.start();
@@ -304,6 +309,20 @@ public class LocalTeamspeakClientSocket
             awaitingCommands.clear();
         }
 
+        if (dropUnannounced) {
+            dropUnannounced = false;
+            // Timed out, or gave up on a command: nobody else tells the listeners.
+            Map<String, String> reason = new HashMap<>();
+            reason.put("reasonid", "3");
+            reason.put("reasonmsg", "connection lost");
+            DisconnectedEvent e = new DisconnectedEvent(reason);
+            try {
+                commandExecutionService.submit(() -> listeners.forEach(e::fire));
+            } catch (RejectedExecutionException ignored) {
+                // the listeners are gone already
+            }
+        }
+
         serverId = null;
         serverProperties = Collections.emptyMap();
 
@@ -740,16 +759,39 @@ public class LocalTeamspeakClientSocket
             return command;
         }
 
-        public boolean ensureSent()
+        /**
+         * Sends the command unless it already was. Its acknowledgement is awaited in the background: the reply
+         * proves delivery anyway, and this is often called from the packet handler thread, which must keep
+         * handling while resends go on. A command that never gets through fails and makes way for the next one.
+         */
+        public void ensureSent()
                 throws IOException, TimeoutException {
             synchronized (sendLock) {
-                if (!sent) {
-                    LocalTeamspeakClientSocket.this.writePacket(new PacketBody2Command(ProtocolRole.CLIENT, command));
+                if (sent) return;
                 sent = true;
-                    return true;
             }
 
-                return false;
+            CompletableFuture<Void> acknowledged;
+            try {
+                acknowledged = writePacketAsync(new PacketBody2Command(ProtocolRole.CLIENT, command));
+            } catch (IOException | TimeoutException | RuntimeException e) {
+                abandon(e);
+                throw e;
+            }
+            acknowledged.whenComplete((ignored, e) -> {
+                if (e != null) abandon(e);
+            });
+        }
+
+        private void abandon(Throwable cause) {
+            completeFailure(cause);
+            synchronized (commandSendLock) {
+                if (!awaitingCommands.remove(returnCode, this)) return; // answered after all
+                try {
+                    sendNextCommand();
+                } catch (IOException | TimeoutException e) {
+                    // that command failed as well, and made way in turn
+                }
             }
         }
     }
@@ -1304,6 +1346,7 @@ public class LocalTeamspeakClientSocket
             throws
             IOException, TimeoutException,
             ExecutionException, InterruptedException {
+        dropUnannounced = false;
         try {
             switch (getState()) {
                 case RETRIEVING_DATA: