diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/Packet.java b/src/main/java/com/github/manevolent/ts3j/protocol/Packet.java index 1cd7176..e192fd6 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/Packet.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/Packet.java @@ -136,4 +136,10 @@ public final class Packet { public long getCreatedNanotime() { return creationNanotime; } + + public byte[] toByteArray() { + ByteBuffer buffer = ByteBuffer.allocate(getSize()); + write(buffer); + return buffer.array(); + } } 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 b45f135..ed3189c 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 @@ -8,57 +8,68 @@ import com.github.manevolent.ts3j.util.Ts3Debugging; import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; +import java.util.stream.Collectors; public class PacketReassembly { private final Map queue = new LinkedHashMap<>(); private boolean state = false; public Packet reassemble(Packet lastFragment) throws IOException { - // Pull out all other packets in the queue before this one which would also be fragmented List reassemblyList = new ArrayList<>(); - for (int packetId = lastFragment.getHeader().getPacketId();; packetId--) { - if (packetId < 0) packetId = 65536 + packetId; - Packet olderPacket = queue.get(packetId); + try { + // Pull out all other packets in the queue before this one which would also be fragmented + for (int packetId = lastFragment.getHeader().getPacketId(); ; packetId--) { + if (packetId < 0) packetId = 65536 + packetId; - if (olderPacket == null) - break; // Chain breaks here - else if (olderPacket.getHeader().getType() != lastFragment.getHeader().getType()) - continue; // skip - else { - reassemblyList.add(olderPacket); + Packet olderPacket = queue.get(packetId); + + if (olderPacket == null) + break; // Chain breaks here + else if (olderPacket.getHeader().getType() != lastFragment.getHeader().getType()) + continue; // skip + else { + reassemblyList.add(olderPacket); + } } + + // Reverse the collection + Collections.reverse(reassemblyList); + + // Rebuild a master packet from the contents of all previous packet fragments + int totalLength = + reassemblyList + .stream() + .mapToInt(x -> x.getBody().getSize()) + .sum(); + + if (totalLength < 0) throw new IllegalArgumentException("reassembly too small: " + totalLength); + + Packet firstPacket = reassemblyList.get(0); + Packet reassembledPacket = new Packet(firstPacket.getRole()); + reassembledPacket.setHeader(firstPacket.getHeader()); + + ByteBuffer reassemblyBuffer = ByteBuffer.allocate(totalLength); + + reassembledPacket.getHeader().setPacketFlag(HeaderFlag.FRAGMENTED, false); + + for (Packet old : reassemblyList) + old.writeBody(reassemblyBuffer); + + if (firstPacket.getHeader().getPacketFlag(HeaderFlag.COMPRESSED)) + reassemblyBuffer = ByteBuffer.wrap(QuickLZ.decompress(reassemblyBuffer.array())); + + reassembledPacket.readBody(reassemblyBuffer); + + return reassembledPacket; + } catch (Exception ex) { + throw new IOException("Problem reassembling " + reassemblyList.size() + " packet(s): [" + + String.join(",", + reassemblyList.stream().map(x -> "{" + + Base64.getEncoder().encodeToString(x.toByteArray()) + + "}").collect(Collectors.toList())) + "]", + ex); } - - // Reverse the collection - Collections.reverse(reassemblyList); - - // Rebuild a master packet from the contents of all previous packet fragments - int totalLength = - reassemblyList - .stream() - .mapToInt(x -> x.getBody().getSize()) - .sum(); - - if (totalLength < 0) throw new IllegalArgumentException("reassembly too small: " + totalLength); - - Packet firstPacket = reassemblyList.get(0); - Packet reassembledPacket = new Packet(firstPacket.getRole()); - reassembledPacket.setHeader(firstPacket.getHeader()); - - ByteBuffer reassemblyBuffer = ByteBuffer.allocate(totalLength); - - reassembledPacket.getHeader().setPacketFlag(HeaderFlag.FRAGMENTED, false); - - for (Packet old : reassemblyList) - old.writeBody(reassemblyBuffer); - - if (firstPacket.getHeader().getPacketFlag(HeaderFlag.COMPRESSED)) - reassemblyBuffer = ByteBuffer.wrap(QuickLZ.decompress(reassemblyBuffer.array())); - - reassembledPacket.readBody(reassemblyBuffer); - - return reassembledPacket; } public Packet put(Packet packet) throws IOException { 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 11827b5..0ed8a01 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 @@ -291,11 +291,6 @@ public abstract class AbstractTeamspeakClientSocket case COMMAND_LOW: header.setPacketFlag(HeaderFlag.NEW_PROTOCOL, true); break; - case VOICE: - // > X is a ushort in H2N order of an own audio packet counter - // it seems it can be the same as the packet counter so we will let the packethandler do it. - ((PacketBody0Voice)body).setPacketId(header.getPacketId()); - break; } // Assign header pointer @@ -383,6 +378,12 @@ public abstract class AbstractTeamspeakClientSocket break; } + if (header.getType() == PacketBodyType.VOICE) { + // > X is a ushort in H2N order of an own audio packet counter + // it seems it can be the same as the packet counter so we will let the packethandler do it. + ((PacketBody0Voice)packet.getBody()).setPacketId(header.getPacketId()); + } + // Use the counters determined above to generate a packet id and generation id header.setPacketId(counter.getKey()); header.setGeneration(counter.getValue()); diff --git a/src/test/java/com/github/manevolent/ts3j/FragmentTest.java b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java index acdb22a..12f312c 100644 --- a/src/test/java/com/github/manevolent/ts3j/FragmentTest.java +++ b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java @@ -14,7 +14,6 @@ import com.github.manevolent.ts3j.util.Ts3Debugging; import junit.framework.TestCase; import org.bouncycastle.util.encoders.Base64; -import java.io.IOException; import java.util.List; import java.util.Random; @@ -27,7 +26,7 @@ public class FragmentTest extends TestCase { public void testParser() throws Exception { Ts3Debugging.setEnabled(true); - byte[] data = new byte[2000]; + byte[] data = new byte[1503]; new Random(0x1234567).nextBytes(data); ClientPacketHeader header = new ClientPacketHeader(); diff --git a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java index 63a40ff..d1fb124 100644 --- a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java +++ b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java @@ -1,5 +1,8 @@ package com.github.manevolent.ts3j; ; +import com.github.manevolent.ts3j.api.Client; +import com.github.manevolent.ts3j.audio.Microphone; +import com.github.manevolent.ts3j.enums.CodecType; import com.github.manevolent.ts3j.event.*; import com.github.manevolent.ts3j.identity.LocalIdentity; import com.github.manevolent.ts3j.protocol.client.ClientConnectionState; @@ -142,11 +145,37 @@ 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 < 1000; i ++) client.getClientInfo(client.getClientId()); + for (int i = 0; i < 100000; i ++) { + for (Client c : client.listClients()) + client.getClientInfo(c.getId()); + + Thread.sleep(1000L); + } Thread.sleep(100000000L); } catch (Throwable ex) {