diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java index ba95c9b..7573ce7 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java +++ b/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,23 +989,25 @@ public abstract class AbstractTeamspeakClientSocket private class NetworkHandler implements Runnable { @Override public void run() { - try { - while (isReading()) { - 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); - } catch (InterruptedException e) { - break; - } + while (isReading()) { + Packet packet; + try { + 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 handling packet", e); + // A disconnect interrupts whatever the handler was waiting on; that is not an error. + if (isReading()) getExceptionHandler().accept(e); } - } catch (Throwable e) { - Ts3Debugging.debug("Problem reading packet", e); - // A disconnect interrupts whatever the handler was waiting on; that is not an error. - if (isReading()) getExceptionHandler().accept(e); } } } diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java index f7b91e7..2b18bd3 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java @@ -66,6 +66,9 @@ public class LocalTeamspeakClientSocket *

* get */ + /** Whether losing the connection now would go unreported: connected, and neither side has said goodbye. */ + private volatile boolean dropUnannounced; + private Map 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 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)); - sent = true; - return true; - } + if (sent) return; + sent = true; + } - return false; + CompletableFuture 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: