diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java index 22d2249..0745c4a 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/Fragments.java @@ -4,7 +4,6 @@ import com.github.manevolent.ts3j.protocol.Packet; import com.github.manevolent.ts3j.protocol.header.HeaderFlag; import com.github.manevolent.ts3j.protocol.packet.PacketBodyCompressed; import com.github.manevolent.ts3j.util.QuickLZ; -import com.github.manevolent.ts3j.util.Ts3Debugging; import java.nio.ByteBuffer; import java.util.Collections; @@ -39,8 +38,17 @@ public final class Fragments { if (packet.getHeader().getType().isCompressible()) { byte[] compressed = QuickLZ.compress(outputBuffer.array(), 1); - packet.getHeader().setPacketFlag(HeaderFlag.COMPRESSED, true); - packet.setBody(new PacketBodyCompressed(packet.getHeader().getType(), packet.getRole(), compressed)); + Packet compressedPacket = new Packet(packet.getRole()); + compressedPacket.setHeader(packet.getHeader().clone()); + compressedPacket.getHeader().setPacketFlag(HeaderFlag.COMPRESSED, true); + compressedPacket.setBody( + new PacketBodyCompressed( + packet.getHeader().getType(), packet.getRole(), + compressed + ) + ); + + packet = compressedPacket; if (packet.getSize() <= MAXIMUM_PACKET_SIZE) return Collections.singletonList(packet); diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java index 39089eb..c1036b9 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java @@ -11,36 +11,58 @@ import java.util.stream.Collectors; public class PacketReassembly { private final Map queue = new LinkedHashMap<>(); - private boolean state = false; + private int currentPacketId = 0; - public Packet reassemble(Packet lastFragment) throws IOException { + public void setPacketId(int currentPacketId) { + this.currentPacketId = currentPacketId; + } + + public boolean put(Packet packet) throws IOException { + queue.put(packet.getHeader().getPacketId(), packet); + + return currentPacketId == packet.getHeader().getPacketId(); + } + + public Packet next() throws IOException { List reassemblyList = new ArrayList<>(); try { // Pull out all other packets in the queue before this one which would also be fragmented List packetIds = new ArrayList<>(); - for (int packetId = lastFragment.getHeader().getPacketId(); ; packetId--) { - if (packetId < 0) packetId = 65536 + packetId; + boolean state = false; - Packet olderPacket = queue.get(packetId); + for (int packetId = currentPacketId; ; packetId++) { + if (packetId >= 65536) packetId = 0; - if (olderPacket == null) + Packet packet = queue.get(packetId); + + if (packet == null) break; // Chain breaks here - else if (olderPacket.getHeader().getType() != lastFragment.getHeader().getType()) - continue; // skip??? else { packetIds.add(packetId); - reassemblyList.add(olderPacket); + reassemblyList.add(packet); - if (lastFragment.getHeader().getPacketId() != packetId && - olderPacket.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) - break; + if (packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) { + state = !state; + + if (!state) { + currentPacketId += packetIds.size(); + break; // fragmentation stopped, we found the last packet + } else + continue; // keep collecting fragments + } else if (!state) { + currentPacketId ++; + break; // don't reassemble this packet, it's not a fragment + } else if (state) { + // keep collecting fragments + continue; + } } } - // Reverse the collection - Collections.reverse(reassemblyList); + if (state) return null; // cannot reassemble yet because we don't have all fragments + else if (reassemblyList.size() <= 0) return null; // cannot reassemble because we miss the first packet // Rebuild a master packet from the contents of all previous packet fragments int totalLength = @@ -53,7 +75,7 @@ public class PacketReassembly { Packet firstPacket = reassemblyList.get(0); Packet reassembledPacket = new Packet(firstPacket.getRole()); - reassembledPacket.setHeader(firstPacket.getHeader()); + reassembledPacket.setHeader(firstPacket.getHeader().clone()); ByteBuffer reassemblyBuffer = ByteBuffer.allocate(totalLength); @@ -62,6 +84,8 @@ public class PacketReassembly { for (Packet old : reassemblyList) old.writeBody(reassemblyBuffer); + reassemblyBuffer.position(0); + if (firstPacket.getHeader().getPacketFlag(HeaderFlag.COMPRESSED)) reassemblyBuffer = ByteBuffer.wrap(QuickLZ.decompress(reassemblyBuffer.array())); @@ -70,7 +94,7 @@ public class PacketReassembly { // Remove read packets packetIds.forEach(queue::remove); - return reassembledPacket; + return reassembledPacket; } catch (Exception ex) { throw new IOException("Problem reassembling " + reassemblyList.size() + " packet(s): [" + String.join(",", @@ -81,43 +105,8 @@ public class PacketReassembly { } } - public Packet put(Packet packet) throws IOException { - Packet reassembled; - - if (packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) { - boolean oldState = state; - - state = !state; - - if (!(packet.getBody() instanceof PacketBodyFragment)) - throw new IllegalArgumentException("packet fragment object is not representative of a fragment"); - - queue.put(packet.getHeader().getPacketId(), packet); - - if (oldState) { - reassembled = reassemble(packet); - } else { - reassembled = null; - } - } else { - if (state) { - queue.put(packet.getHeader().getPacketId(), packet); - - reassembled = null; - } else { - reassembled = packet; - } - } - - return reassembled; - } - - public boolean isReassembling() { - return state; - } - public void reset() { - state = false; + currentPacketId = 0; queue.clear(); } } \ No newline at end of file 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 f4f5215..09f4b88 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 @@ -141,8 +141,6 @@ public class LocalClientHandlerConnecting extends LocalClientHandler { } } else if (packet.getBody() instanceof PacketBody2Command) { - Ts3Debugging.debug(((PacketBody2Command) packet.getBody()).getText()); - SingleCommand command = ((PacketBody2Command) packet.getBody()).parse().simplifyOne(); if (command.getName().equalsIgnoreCase("initivexpand2")) { diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java index 0b8abcb..74940fa 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerFull.java @@ -26,7 +26,7 @@ public abstract class LocalClientHandlerFull extends LocalClientHandler { public void handlePacket(Packet packet) throws IOException, TimeoutException { switch (packet.getBody().getType()) { case COMMAND: - Ts3Debugging.debug(((PacketBody2Command) packet.getBody()).getText()); + Ts3Debugging.debug("[COMMAND] " + ((PacketBody2Command) packet.getBody()).getText()); try { handleCommand(MultiCommand.parse( @@ -39,7 +39,7 @@ public abstract class LocalClientHandlerFull extends LocalClientHandler { break; case COMMAND_LOW: - Ts3Debugging.debug("[LOW] " + ((PacketBody2Command) packet.getBody()).getText()); + Ts3Debugging.debug("[COMMAND] [LOW] " + ((PacketBody2Command) packet.getBody()).getText()); try { handleCommand(MultiCommand.parse( 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 d8cc227..9743234 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 @@ -95,6 +95,8 @@ public abstract class AbstractTeamspeakClientSocket private Map packetStatistics = new LinkedHashMap<>(); // + private final List handlerBacklog = new LinkedList<>(); + private final NetworkReader networkReader = new NetworkReader(); private Thread networkThread = null; @@ -207,6 +209,8 @@ public abstract class AbstractTeamspeakClientSocket readQueue.clear(); + handlerBacklog.clear(); + for (PacketBodyType bodyType : PacketBodyType.values()) if (bodyType.isSplittable()) reassemblyQueue.get(bodyType).reset(); @@ -500,9 +504,7 @@ public abstract class AbstractTeamspeakClientSocket boolean fragment = networkPacket.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED); - if (networkPacket.getHeader().getType().isSplittable()) { - fragment = fragment || reassemblyQueue.get(networkPacket.getHeader().getType()).isReassembling(); - } else if (fragment) + if (fragment && !networkPacket.getHeader().getType().isSplittable()) throw new IllegalArgumentException( networkPacket.getHeader().getType() + "packet is fragment, but not splittable" @@ -538,15 +540,15 @@ public abstract class AbstractTeamspeakClientSocket } // Fragment handling - if (fragment) { - Ts3Debugging.debug("[PROTOCOL] READ " + networkPacket.getHeader().getType().name() + " fragment"); + if (networkPacket.getHeader().getType().isSplittable()) { + Ts3Debugging.debug("[PROTOCOL] READ " + networkPacket.getHeader().getType().name()); packet.setBody(new PacketBodyFragment(networkPacket.getHeader().getType(), getRole().getIn())); } else { + Ts3Debugging.debug("[PROTOCOL] READ " + networkPacket.getHeader().getType().name()); + if (packet.getHeader().getPacketFlag(HeaderFlag.COMPRESSED) && packet.getHeader().getType().isCompressible()) packet.setBody(new PacketBodyCompressed(networkPacket.getHeader().getType(), getRole().getIn())); - - Ts3Debugging.debug("[PROTOCOL] READ " + networkPacket.getHeader().getType().name()); } // Read raw packet body @@ -586,6 +588,12 @@ public abstract class AbstractTeamspeakClientSocket long timeout; while (isReading()) { + // De-spool spooled-up packets + if (handlerBacklog.size() > 0) { + Packet packet = handlerBacklog.remove(0); + return packet; + } + try { timeout = getState() == ClientConnectionState.CONNECTED ? Math.min(1000L, 1000L - (System.currentTimeMillis() - lastPing)) : @@ -663,12 +671,6 @@ public abstract class AbstractTeamspeakClientSocket } } - // Find if the packet must be reassembled - if (packet.getHeader().getType().isSplittable()) { - packet = reassemblyQueue.get(packet.getHeader().getType()).put(packet); - if (packet == null) continue; - } - // Find if the packet is itself an acknowledgement (ACK or ACK_LOW) boolean handle; final PacketResponse response; @@ -684,8 +686,6 @@ public abstract class AbstractTeamspeakClientSocket handle = false; break; case ACK_LOW: - lastResponse = System.currentTimeMillis(); - response = sendQueueLow.get(((PacketBody7AckLow)packet.getBody()).getPacketId()); if (response == null) { Ts3Debugging.debug("Unrecognized ACK_LOW: " + ((PacketBody6Ack) packet.getBody()).getPacketId()); @@ -696,7 +696,6 @@ public abstract class AbstractTeamspeakClientSocket handle = false; break; case PONG: - lastResponse = System.currentTimeMillis(); response = pingQueue.get(((PacketBody5Pong)packet.getBody()).getPacketId()); if (response == null) { Ts3Debugging.debug("Unrecognized PONG: " + ((PacketBody5Pong) packet.getBody()).getPacketId()); @@ -733,6 +732,9 @@ public abstract class AbstractTeamspeakClientSocket currentRto = smoothedRtt + Math.max(0D, 4 * smoothedRttVar); } + if (response != null) + response.acknowledge(packet); + // This is pulled out of the following IF block so that packet generation can increase correctly boolean placed = counter != null && counter.put(packet.getHeader().getPacketId()); @@ -742,10 +744,24 @@ public abstract class AbstractTeamspeakClientSocket handle = handle & placed; } - if (response != null) { - response.acknowledge(packet); + if (handle) { + if (packet.getHeader().getType().isSplittable()) { + if (!placed) continue; // reset + + // Place in the reassembly queue + PacketReassembly reassembly = reassemblyQueue.get(packet.getHeader().getType()); + reassembly.put(packet); + + Packet reassembled; + while ((reassembled = reassembly.next()) != null) { + Ts3Debugging.debug("[PROTOCOL] REASSEMBLE " + reassembled.getHeader().getType().name() + + " id=" + reassembled.getHeader().getPacketId() + " len=" + reassembled.getSize()); + handlerBacklog.add(reassembled); + } + } else { + handlerBacklog.add(packet); + } } - if (handle) return packet; } throw new IllegalStateException("no longer reading"); @@ -874,7 +890,7 @@ public abstract class AbstractTeamspeakClientSocket } catch (Throwable e) { if (e instanceof InterruptedException) continue; - Ts3Debugging.debug("Problem handling packet", e); + 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 87eeda1..c89f906 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 @@ -141,8 +141,6 @@ public class LocalTeamspeakClientSocket response.add(new CommandSingleParameter("connection_server2client_packetloss_total", Integer.toString(0))); - Ts3Debugging.debug(response.build()); - try { writePacket(new PacketBody2Command(getRole().getOut(), response)); } catch (Exception e) { diff --git a/src/test/java/com/github/manevolent/ts3j/FragmentTest.java b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java index 7cab3f3..6c1c598 100644 --- a/src/test/java/com/github/manevolent/ts3j/FragmentTest.java +++ b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java @@ -25,7 +25,7 @@ public class FragmentTest extends TestCase { public void testParser() throws Exception { Ts3Debugging.setEnabled(true); - byte[] data = new byte[1503]; + byte[] data = new byte[5003]; new Random(0x1234567).nextBytes(data); ClientPacketHeader header = new ClientPacketHeader(); @@ -40,6 +40,7 @@ public class FragmentTest extends TestCase { new AbstractTeamspeakClientSocket.LocalCounterFull(65536, true); counter.setPacketId(65534); // Test wrapping around the generation at the same time :) + reassembly.setPacketId(65535); List pieces = Fragments.split(largePacket); Ts3Debugging.debug("Created " + pieces.size() + " pieces (total=" + data.length + ")."); @@ -49,11 +50,16 @@ public class FragmentTest extends TestCase { piece.getHeader().setPacketId(packetId.getKey()); piece.getHeader().setGeneration(packetId.getValue()); - Packet out = reassembly.put(piece); - if (out != null) - assertEquals(text, ((PacketBody2Command)out.getBody()).getText()); + reassembly.put(piece); } + Packet reassembled; + while (((reassembled = reassembly.next()) != null)) { + assertEquals( + ((PacketBody2Command) largePacket.getBody()).getText(), + ((PacketBody2Command) reassembled.getBody()).getText() + ); + } } } diff --git a/src/test/java/com/github/manevolent/ts3j/FragmentTest2.java b/src/test/java/com/github/manevolent/ts3j/FragmentTest2.java index 4b94114..930066e 100644 --- a/src/test/java/com/github/manevolent/ts3j/FragmentTest2.java +++ b/src/test/java/com/github/manevolent/ts3j/FragmentTest2.java @@ -2,24 +2,16 @@ package com.github.manevolent.ts3j; import com.github.manevolent.ts3j.protocol.Packet; import com.github.manevolent.ts3j.protocol.ProtocolRole; -import com.github.manevolent.ts3j.protocol.header.ClientPacketHeader; import com.github.manevolent.ts3j.protocol.header.HeaderFlag; -import com.github.manevolent.ts3j.protocol.packet.PacketBody2Command; -import com.github.manevolent.ts3j.protocol.packet.PacketBodyType; -import com.github.manevolent.ts3j.protocol.packet.fragment.Fragments; import com.github.manevolent.ts3j.protocol.packet.fragment.PacketBodyFragment; import com.github.manevolent.ts3j.protocol.packet.fragment.PacketReassembly; -import com.github.manevolent.ts3j.protocol.socket.client.AbstractTeamspeakClientSocket; -import com.github.manevolent.ts3j.util.Pair; import com.github.manevolent.ts3j.util.Ts3Debugging; import junit.framework.TestCase; import org.bouncycastle.util.encoders.Base64; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.Random; public class FragmentTest2 extends TestCase { @@ -31,7 +23,9 @@ public class FragmentTest2 extends TestCase { Ts3Debugging.setEnabled(true); String[] packets = new String[] { - // Put diagnostic here + "LvdqqRB1ozMABBJlaXY0e5iIcGFja2V0c4FQkRkjgz0zMnE7IFGYbmVjIviVXFI5Ag8P3F8kgzQ4Mzg5OSFgkIgSRjSFKjI0MjgxYJiIVOJNNDE2NTA0OXFgmYh7gl+iZYsxz/mBUFGYZIUqMTkzUlOQiB5taW4hIocqMTgBYJCIFEc0oGUSMjkRYJCIIrv4MjmhYJiIUZiCNQMHAIAxGST6MDI5MTI2QWCZiKPzX2lwPTc0LjIwOS4yLjg1", + "xX0QvWydOFsAAlJHhAQAAEoHAAAAAACgY2lkPTEgY2xpZW50X2lkbGVfdGltZT0xNjgyMjdmXnVgAACAbmlxdWWR8TGSaWZpZXI9SW9RY3NZMnYwcXpNMThOeGFsAAQCgFBQazBiTEpLQT1mXm5pY2tuYbFrbm90XHNhXHN3aG9yZQEAAIBmXnZlcnNpb249My4xLjEwXHNbQnVpbGQ6XHMxNTI4NUAAgIAzNzYxNV1mXnBsYXRmb3JtPVdpbmRvd3NmXmlucHV0X22IhgOWdXRleedvdXR4IzBmXoQib25seZWgJ2D0lmhhcmR3YXJh52xeX+BkEmRlAAACiGZhdWx0X2NoYW5uZWw9XC84Zl5tZXRhX2RhdGFmXmlzXwAMAIByZWNvcmRpbmc9J2AFJV9zaWduPStcL0JXdnVlaWtHZzQAAACAWWtPMXYydXVaQjV2dEpKZ1VaNWJMOGNSZnhBc3RmbgAAAIBDVmRybzJqYSs0YSs4ckdVekR4OFwvdnZUWk9VVkQ2AECAhFU5NWhuV2I2MzhNQ1E992BzZWN1cml0eZHkc2hmXmxvZ2kcGUiMbl+Ce2ZeImJiYXMCnz0yZl5V519ncm91cJHxPTVmXnNlcgElA0lzPTYhIlCgZl5jcmVhIzM1MzAhYTc1OGZebGE=", + "+hN1ybZO1gEAAwJzdGNvgYNjJTM0MDg5MDMnYHQwwoCQb3RhbFWY8ZhzPTV3YGF3YXnoI2IXX21lc3NhZ3dldHlwZegjZmwgAACAYWdfYXZhJ3I9ZmQwMjhjYzUzNmU3YmM3MWJiOWQ4MGGQQZOhZWFjYxEUNTNmXiF7a19wb3chGzd3YCN7cmVhMnN06CMqe19tc2dmXmSAAELgZXNjcmlwdPGYPW1hbmVcc2lzs1RnaWFuEbRuaWdnZXJmXuFpBwDggiJ7IRsnYG1vbnRoX2J5dGVzX3VwbG9hZDF7IGAVIYFupZcxNzI3Mg4PwYE0Zl4jg55cMzc4NldgKoMggRnhaXByaW9D43NwZWFrIRs3YIb1X3Bob25liAYM+nRpY2ZebmVlITJfVCVhMnJ5X3ZpZXePaiYwdG9rZW5mXmlRmJLxJ2DhadEAHORW52NvbbF4ZCEbJ2Bjb3VudHJ5PVVTZl5W5wNJX2luaGVB42Vkll2S8QHCgIA3YGJhZGdlcz1vASV3b2xm6CNCZTY0SGFzaEOj81VJRD1jY2kAAACAZWJtbGJpbmtwbmNrbW1tbmhtZGhiZ2tmZG1wamRlZ+AAAKBtbGNlazFllIjxmF9maWxldHJhbnNmZXJfYmFuZHdpZCHYc48BR/ExkuIjkIghQTQ=", }; List pieces = new ArrayList<>(); @@ -55,11 +49,15 @@ public class FragmentTest2 extends TestCase { Ts3Debugging.debug("Loaded " + pieces.size() + " pieces."); PacketReassembly reassembly = new PacketReassembly(); + reassembly.setPacketId(2); + for (Packet piece : pieces) { - Packet out = reassembly.put(piece); - if (out != null) { - Ts3Debugging.debug(((PacketBody2Command)out.getBody()).getText()); - } + reassembly.put(piece); + } + + Packet reassembled; + while (((reassembled = reassembly.next()) != null)) { + Ts3Debugging.debug(reassembled.getHeader().getPacketId()); } } diff --git a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java index d1fb124..a41b9ad 100644 --- a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java +++ b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java @@ -147,37 +147,16 @@ public class ServerConnectionTest { final long[] t = {System.currentTimeMillis()}; - client.setMicrophone(new Microphone() { - @Override - public boolean isReady() { - return true; - } - - @Override - public CodecType getCodec() { - return CodecType.OPUS_MUSIC; - } - - @Override - public byte[] provide() { - System.err.println(System.currentTimeMillis() - t[0]); - t[0] = System.currentTimeMillis(); - return new byte[10]; - } - }); - assertEquals(client.getState(), ClientConnectionState.CONNECTED); client.subscribeAll(); - for (int i = 0; i < 100000; i ++) { + while (client.isConnected()) { for (Client c : client.listClients()) client.getClientInfo(c.getId()); Thread.sleep(1000L); } - - Thread.sleep(100000000L); } catch (Throwable ex) { ex.printStackTrace(); }