diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java index baebafe..bc312d2 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java @@ -185,7 +185,9 @@ public class LocalClientHandlerConnecting extends LocalClientHandler { getClient().getIdentity() ); - getClient().writePacket(new PacketBody2Command( + // Don't wait for the ACK: the server ignores a resent clientek, so a lost ACK would stall us, and + // it can't decrypt clientinit before processing clientek anyway, so the two stay in order. + getClient().writePacketAsync(new PacketBody2Command( ProtocolRole.CLIENT, new SingleCommand( "clientek", ProtocolRole.CLIENT, 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 59bb55e..db8b4c6 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 @@ -283,6 +283,41 @@ public abstract class AbstractTeamspeakClientSocket @Override public void writePacket(PacketBody body) throws IOException, TimeoutException { + writePacket(newOutgoingPacket(body)); + } + + /** + * Sends a packet body without waiting for its acknowledgement. It is numbered and encrypted before this + * returns, so packets written afterwards follow it; resends until it is acknowledged continue in the background. + * + * @return completes once the packet is acknowledged, or exceptionally if it never is + */ + public CompletableFuture writePacketAsync(PacketBody body) throws IOException, TimeoutException { + Packet packet = newOutgoingPacket(body); + List responses = new ArrayList<>(); + + synchronized (sendLocks.get(packet.getHeader().getType())) { + for (Packet piece : split(packet)) + responses.add(sendPacketIntl(piece)); + } + + CompletableFuture acknowledged = new CompletableFuture<>(); + Thread thread = new Thread(() -> { + try { + for (PacketResponse response : responses) + awaitAcknowledgement(response); + acknowledged.complete(null); + } catch (Throwable e) { + acknowledged.completeExceptionally(e); + } + }, "TS3J/Resend"); + thread.setDaemon(true); + thread.start(); + + return acknowledged; + } + + private Packet newOutgoingPacket(PacketBody body) { // Construct a packet object Packet packet = new Packet(body.getRole()); @@ -318,12 +353,32 @@ public abstract class AbstractTeamspeakClientSocket // Assign header pointer packet.setHeader(header); - // Lower-level write; helper is done - writePacket(packet); + return packet; } @Override public void writePacket(Packet packet) throws IOException, TimeoutException { + List pieces = split(packet); + + if (packet.getHeader().getType().isSplittable()) { + // Note, it's very important we send these packets synchronously. + // we allow the split to happen multi-thread, but we block until all + // chunks are sent synchronously. So, all chunks of packets are synchronous. + // this happens because we care that the packet IDs are contiguous. + synchronized (sendLocks.get(packet.getHeader().getType())) { + for (Packet piece : pieces) + awaitAcknowledgement(sendPacketIntl(piece)); // Write the individual piece, incrementing its packet + } + } else { + // Don't lock here, for the slight performance benefit. + awaitAcknowledgement(sendPacketIntl(packet)); + } + } + + /** + * Checks an outgoing packet and splits it into the pieces to send. + */ + private List split(Packet packet) throws IOException { // Sanity checks. People are insane! if (packet.getHeader() == null) throw new NullPointerException("header"); @@ -339,28 +394,23 @@ public abstract class AbstractTeamspeakClientSocket // it's better to do this all at once instead of sending them as they are split, // because we might get halfway through splitting them then throw an exception and // we can't write the rest. - List packets = Fragments.split(packet); - - // Note, it's very important we send these packets synchronously. - // we allow the split to happen multi-thread, but we block until all - // chunks are sent synchronously. So, all chunks of packets are synchronous. - // this happens because we care that the packet IDs are contiguous. - synchronized (sendLocks.get(packet.getHeader().getType())) { - for (Packet piece : packets) - writePacketIntl(piece); // Write the individual piece, incrementing its packet - } - } else { - // Ensure the outbound packet's maximum size isn't too large if it can't be split (which it can't here) - if (packet.getSize() > Fragments.MAXIMUM_PACKET_SIZE) - throw new IllegalArgumentException("packet too large: " + packet.getSize() + "; " + - packet.getHeader().getType().name() + " cannot split"); - - // Don't lock here, for the slight performance benefit. - writePacketIntl(packet); + return Fragments.split(packet); } + + // Ensure the outbound packet's maximum size isn't too large if it can't be split (which it can't here) + if (packet.getSize() > Fragments.MAXIMUM_PACKET_SIZE) + throw new IllegalArgumentException("packet too large: " + packet.getSize() + "; " + + packet.getHeader().getType().name() + " cannot split"); + + return Collections.singletonList(packet); } - private void writePacketIntl(Packet packet) throws IOException, TimeoutException { + /** + * Numbers, encrypts and sends a single packet. + * + * @return the pending acknowledgement to wait for, or null if none is expected + */ + private PacketResponse sendPacketIntl(Packet packet) throws IOException { PacketHeader header = packet.getHeader(); // Count packet @@ -486,8 +536,14 @@ public abstract class AbstractTeamspeakClientSocket .get(kind) .processOutgoing(packet); - // If necessary, fulfill promise for the packet acknowledgement or throw TimeoutException - if (response != null && willWaitForAck) { + return willWaitForAck ? response : null; + } + + /** + * Waits for a packet's acknowledgement, resending it until it arrives. + */ + private void awaitAcknowledgement(PacketResponse response) throws IOException, TimeoutException { + if (response != null) { while (response.getRetries() < response.getMaxTries() && getState() != ClientConnectionState.DISCONNECTED) { try {