Packet reassembly would fail if the server sent packets out of order. Now, it is resilient to this.

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 compressedPacket = new Packet(packet.getRole());
-                packet.setBody(new PacketBodyCompressed(packet.getHeader().getType(), packet.getRole(), compressed));
+                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);

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<Integer, Packet> 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<Packet> reassemblyList = new ArrayList<>();
 
         try {
             // Pull out all other packets in the queue before this one which would also be fragmented
             List<Integer> packetIds = new ArrayList<>();
 
-            for (int packetId = lastFragment.getHeader().getPacketId(); ; packetId--) {
+            boolean state = false;
-                if (packetId < 0) packetId = 65536 + packetId;
 
-                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 &&
+                    if (packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) {
-                            olderPacket.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED))
+                        state = !state;
-                        break;
+
+                        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
+            if (state) return null; // cannot reassemble yet because we don't have all fragments
-            Collections.reverse(reassemblyList);
+            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()));
 
@@ -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();
     }
 }

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")) {

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(

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

@@ -95,6 +95,8 @@ public abstract class AbstractTeamspeakClientSocket
     private Map<PacketKind, PacketStatistics> packetStatistics = new LinkedHashMap<>();
     //
 
+    private final List<Packet> 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()) {
+        if (fragment && !networkPacket.getHeader().getType().isSplittable())
-            fragment = fragment || reassemblyQueue.get(networkPacket.getHeader().getType()).isReassembling();
-        } else if (fragment)
             throw new IllegalArgumentException(
                     networkPacket.getHeader().getType() +
                     "packet is fragment, but not splittable"
@@ -538,15 +540,15 @@ public abstract class AbstractTeamspeakClientSocket
         }
 
         // Fragment handling
-        if (fragment) {
+        if (networkPacket.getHeader().getType().isSplittable()) {
-            Ts3Debugging.debug("[PROTOCOL] READ " + networkPacket.getHeader().getType().name() + " fragment");
+            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) {
+            if (handle) {
-                response.acknowledge(packet);
+                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);
                 }
             }
         }

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) {

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<Packet> 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);
+            reassembly.put(piece);
-            if (out != null)
+        }
-                assertEquals(text, ((PacketBody2Command)out.getBody()).getText());
+
-        }
+        Packet reassembled;
-
+        while (((reassembled = reassembly.next()) != null)) {
+            assertEquals(
+                    ((PacketBody2Command) largePacket.getBody()).getText(),
+                    ((PacketBody2Command) reassembled.getBody()).getText()
+            );
+        }
     }
 
 }

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<Packet> 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);
+            reassembly.put(piece);
-            if (out != null) {
-                Ts3Debugging.debug(((PacketBody2Command)out.getBody()).getText());
         }
+
+        Packet reassembled;
+        while (((reassembled = reassembly.next()) != null)) {
+            Ts3Debugging.debug(reassembled.getHeader().getPacketId());
         }
     }
 

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();
         }