Mic packet ID was wrongly generated, changed some tests and increased verbosity of the reassembly exceptions

This commit is contained in:
Manevolent
2018-08-12 12:14:29 -04:00
parent bc37491328
commit 555f6a56d6
5 changed files with 94 additions and 48 deletions
@@ -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();
}
}
@@ -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<Integer, Packet> 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<Packet> 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 {
@@ -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());
@@ -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();
@@ -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) {