ericek111/ts3j
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

More tests for packet generations. Compression test also exists now to test compression into the future. QuickLZ first checks the level on decompression, and also throws a diagnostic error with full Base64 byte array on issue. Disconnections will actually fully reset and free the client for valid re-use. There is now a Consumer for Throwables for the abstract client socket for error handling.

Browse Source
This commit is contained in:
Manevolent
2018-08-11 15:50:22 -04:00
parent 5a440c890f
commit 04299b5c77
7 changed files with 356 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/github/manevolent/ts3j/command/response/AbstractCommandResponse.java
+2 -2
View File
@@ -39,11 +39,11 @@ public abstract class AbstractCommandResponse<T> implements CommandResponse<T> {
return map(future.get());
}
protected void completeSuccess(Iterable<MultiCommand> commands) {
public void completeSuccess(Iterable<MultiCommand> commands) {
future.complete(commands);
}
protected void completeFailure(Throwable thrown) {
public void completeFailure(Throwable thrown) {
future.completeExceptionally(thrown);
}
src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java
+101 -43
View File
@@ -30,6 +30,7 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;
/**
* Server/client common abstract client socket, used to interact with the UDP connection. Provides most of the protocol
@@ -62,14 +63,16 @@ public abstract class AbstractTeamspeakClientSocket
private CommandProcessor commandProcessor;
private Consumer<Throwable> exceptionHandler = throwable -> {
throwable.printStackTrace();
};
private Identity identity = null;
private ClientConnectionState connectionState = null;
private PacketTransformation transformation = new InitPacketTransformation();
private PacketHandler handler;
private int generationId = 0;
private int clientId = 0;
private int packetId = 0;
private long lastResponse;
private long lastPing;
@@ -98,12 +101,6 @@ public abstract class AbstractTeamspeakClientSocket
protected AbstractTeamspeakClientSocket() {
super(SocketRole.CLIENT);
try {
setState(ClientConnectionState.DISCONNECTED);
} catch (Exception e) {
throw new RuntimeException(e);
}
// Initialize reassembly queue
for (PacketBodyType bodyType : PacketBodyType.values())
if (bodyType.isSplittable())
@@ -124,6 +121,12 @@ public abstract class AbstractTeamspeakClientSocket
for (PacketKind kind : PacketKind.values()) {
packetStatistics.put(kind, new PacketStatistics());
}
try {
setState(ClientConnectionState.DISCONNECTED);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
public PacketStatistics getStatistics(PacketKind kind) {
@@ -158,6 +161,7 @@ public abstract class AbstractTeamspeakClientSocket
if (state != this.connectionState) {
Ts3Debugging.debug("State changing: " + state.name());
boolean initialState = this.connectionState == null;
boolean wasDisconnected = this.connectionState == ClientConnectionState.DISCONNECTED;
if (wasDisconnected) setReading(true);
@@ -168,8 +172,8 @@ public abstract class AbstractTeamspeakClientSocket
connectionStateLock.notifyAll();
if (state == ClientConnectionState.DISCONNECTED)
clientOptions.clear();
if (state == ClientConnectionState.DISCONNECTED && !initialState)
onDisconnect();
Ts3Debugging.debug("State changed: " + state.name());
}
@@ -189,6 +193,31 @@ public abstract class AbstractTeamspeakClientSocket
}
}
protected void onDisconnect() {
clientOptions.clear();
clientId = -1;
readQueue.clear();
for (PacketBodyType bodyType : PacketBodyType.values())
if (bodyType.isSplittable())
reassemblyQueue.get(bodyType).reset();
pingQueue.clear();
sendQueue.clear();
sendQueueLow.clear();
for (PacketBodyType bodyType : PacketBodyType.values()) {
localSendCounter.get(bodyType).reset();
remoteSendCounter.get(bodyType).reset();
}
for (PacketKind kind : PacketKind.values()) {
packetStatistics.put(kind, new PacketStatistics());
}
}
public ClientConnectionState getState() {
return connectionState;
}
@@ -517,7 +546,7 @@ public abstract class AbstractTeamspeakClientSocket
packet.setBody(new PacketBodyFragment(networkPacket.getHeader().getType(), getRole().getIn()));
} else {
if (packet.getHeader().getPacketFlag(HeaderFlag.COMPRESSED))
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());
@@ -528,7 +557,6 @@ public abstract class AbstractTeamspeakClientSocket
// Finally, handle compressed bodies
if (packet.getBody() instanceof PacketBodyCompressed) {
Ts3Debugging.debug("[PROTOCOL] DECOMPRESS " + networkPacket.getHeader().getType().name());
byte[] decompressed = QuickLZ.decompress(((PacketBodyCompressed) packet.getBody()).getCompressed());
@@ -591,16 +619,17 @@ public abstract class AbstractTeamspeakClientSocket
}
networkPacket.getHeader().setGeneration(generation);
// Ignore voice and whisper, there is a decompression issue with them
if (networkPacket.getHeader().getPacketFlag(HeaderFlag.COMPRESSED) && (
networkPacket.getHeader().getType() == PacketBodyType.VOICE ||
networkPacket.getHeader().getType() == PacketBodyType.VOICE_WHISPER
))
continue;
// Read packet (decrypt, decompress, etc)
Packet packet = readPacketIntl(networkPacket);
Packet packet;
try {
packet = readPacketIntl(networkPacket);
} catch (Exception ex) {
exceptionHandler.accept(
new Exception("Problem reading " + networkPacket.getHeader().getType().name(), ex)
);
continue;
}
// Find if the packet must be acknowledged
PacketBodyType ackType = packet
@@ -705,10 +734,13 @@ public abstract class AbstractTeamspeakClientSocket
currentRto = smoothedRtt + Math.max(0D, 4 * smoothedRttVar);
}
// 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());
if (packet.getHeader().getType().canResend() && packet.getHeader().getType() != PacketBodyType.INIT1) {
// Find if we already acknowledged this packet
if (counter != null)
handle = handle & counter.put(packet.getHeader().getPacketId());
handle = handle & placed;
}
if (response != null) {
@@ -805,22 +837,6 @@ public abstract class AbstractTeamspeakClientSocket
this.clientId = clientId;
}
public int getPacketId() {
return packetId;
}
public void setPacketId(int packetId) {
this.packetId = packetId;
}
public int getGenerationId() {
return generationId;
}
public void setGenerationId(int generationId) {
this.generationId = generationId;
}
public String getNickname( ){
return getOption("client.nickname", String.class);
}
@@ -837,6 +853,14 @@ public abstract class AbstractTeamspeakClientSocket
this.commandProcessor = commandProcessor;
}
public Consumer<Throwable> getExceptionHandler() {
return exceptionHandler;
}
public void setExceptionHandler(Consumer<Throwable> exceptionHandler) {
this.exceptionHandler = exceptionHandler;
}
private class NetworkHandler implements Runnable {
@Override
public void run() {
@@ -864,10 +888,12 @@ public abstract class AbstractTeamspeakClientSocket
try {
readQueue.put(readPacket());
} catch (Throwable e) {
if (e instanceof InterruptedException) continue;
e.printStackTrace();
if (e instanceof InterruptedException) {
Thread.yield();
continue;
}
exceptionHandler.accept(e);
break;
}
}
@@ -875,6 +901,7 @@ public abstract class AbstractTeamspeakClientSocket
try {
setState(ClientConnectionState.DISCONNECTED);
} catch (Throwable e1) {
exceptionHandler.accept(e1);
}
}
}
@@ -991,6 +1018,7 @@ public abstract class AbstractTeamspeakClientSocket
*/
boolean put(int packetId);
void reset();
}
public static class RemoteCounterZero implements RemoteCounter {
@@ -1010,6 +1038,11 @@ public abstract class AbstractTeamspeakClientSocket
return true;
}
@Override
public void reset() {
// Do nothing
}
}
public static class RemoteCounterFull implements RemoteCounter {
@@ -1141,6 +1174,12 @@ public abstract class AbstractTeamspeakClientSocket
return put(packetId);
}
@Override
public void reset() {
for (int i = 0; i < bufferSize; i ++)
buffer[i] = null;
}
private boolean putRelative(int index, int generation) {
Integer existing = buffer[index];
if (existing == null || existing != generation) {
@@ -1203,6 +1242,8 @@ public abstract class AbstractTeamspeakClientSocket
default Pair<Integer,Integer> current() {
return new Pair<>(getPacketId(), getGeneration());
}
void reset();
}
public static class LocalCounterZero implements LocalCounter {
@@ -1230,6 +1271,11 @@ public abstract class AbstractTeamspeakClientSocket
public void setGeneration(int i) {
throw new UnsupportedOperationException();
}
@Override
public void reset() {
// Do nothing
}
}
public static class LocalCounterFull implements LocalCounter {
@@ -1284,6 +1330,13 @@ public abstract class AbstractTeamspeakClientSocket
}
}
@Override
public void reset() {
synchronized (sendLock) {
packetId = generationId = 0;
}
}
@Override
public Pair<Integer, Integer> next() {
synchronized (sendLock) {
@@ -1301,7 +1354,7 @@ public abstract class AbstractTeamspeakClientSocket
private final Map<Integer, Packet> queue = new LinkedHashMap<>();
private boolean state = false;
public Packet reassemble(Packet lastFragment) {
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--) {
@@ -1342,7 +1395,7 @@ public abstract class AbstractTeamspeakClientSocket
return reassembledPacket;
}
public Packet put(Packet packet) {
public Packet put(Packet packet) throws IOException {
Packet reassembled;
if (packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) {
@@ -1376,5 +1429,10 @@ public abstract class AbstractTeamspeakClientSocket
public boolean isReassembling() {
return state;
}
public void reset() {
state = false;
queue.clear();
}
}
}
src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java
+27 -7
View File
@@ -214,6 +214,24 @@ public class LocalTeamspeakClientSocket
super.setIdentity(identity);
}
@Override
protected void onDisconnect() {
super.onDisconnect();
synchronized (commandSendLock) {
for (Integer id : new ArrayList<>(awaitingCommands.keySet())) {
ClientCommandResponse response = awaitingCommands.get(id);
response.completeFailure(new IOException("client disconnected"));
}
awaitingCommands.clear();
calculateAcceptingReturnCode();
}
serverId = null;
}
/**
* Initiates a connection to a server
* @param remote remote sever to contact
@@ -232,8 +250,7 @@ public class LocalTeamspeakClientSocket
throw new IllegalStateException(connectionState.name());
setClientId(0);
setPacketId(0);
setGenerationId(0);
serverId = null;
awaitingCommands.clear();
acceptingReturnCode = lastReturnCode = 0;
@@ -320,11 +337,15 @@ public class LocalTeamspeakClientSocket
Command command,
Function<SingleCommand, T> processor
) throws IOException, TimeoutException {
int returnCode;
final ClientCommandResponse<Iterable<T>> response;
// sync may not be necessary here
synchronized (commandSendLock) {
if (getState() != ClientConnectionState.CONNECTED)
throw new IOException("not connected");
int returnCode;
// sync may not be necessary here
returnCode = lastReturnCode++;
command.add(new CommandSingleParameter("return_code", Integer.toString(returnCode)));
@@ -400,11 +421,10 @@ public class LocalTeamspeakClientSocket
try {
writePacket(voice);
} catch (IOException e) {
} catch (Exception e) {
// All we are really concerned about here is a disconnection event, in which case the loop
// will exit.
} catch (TimeoutException e) {
// Not really possible; as we might expect, voice doesn't have ACKs.
getExceptionHandler().accept(e);
}
}
}
src/main/java/com/github/manevolent/ts3j/util/QuickLZ.java
+131 -162
View File
@@ -12,6 +12,7 @@ package com.github.manevolent.ts3j.util;// QuickLZ data compression library
//
// Version: 1.5.0 final
import java.io.IOException;
import java.util.Base64;
public final class QuickLZ
@@ -309,182 +310,150 @@ public final class QuickLZ
a[i + j] = (byte)(value >>> (j * 8));
}
static public byte[] decompress(byte[] source)
{
int size = (int)sizeDecompressed(source);
if (size > 65536) throw new IllegalArgumentException("decompression too large: " + size + " (" + Base64.getEncoder().encodeToString(source) + ")");
static public byte[] decompress(byte[] source) throws IOException {
try {
int level = (source[0] >>> 2) & 0x3;
int src = headerLen(source);
int dst = 0;
long cword_val = 1;
byte[] destination = new byte[size];
int[] hashtable = new int[4096];
byte[] hash_counter = new byte[4096];
int last_matchstart = size - UNCONDITIONAL_MATCHLEN - UNCOMPRESSED_END - 1;
int last_hashed = -1;
int hash;
int fetch = 0;
if (level != 1 && level != 3)
throw new IllegalArgumentException("unsupported QuickLZ level: " + level);
int level = (source[0] >>> 2) & 0x3;
int size = (int) sizeDecompressed(source);
if (size > 65536)
throw new IllegalArgumentException("decompression too large: " + size);
if (level != 1 && level != 3)
throw new IllegalArgumentException("unsupported QuickLZ level: " + level);
int src = headerLen(source);
int dst = 0;
long cword_val = 1;
byte[] destination = new byte[size];
int[] hashtable = new int[4096];
byte[] hash_counter = new byte[4096];
int last_matchstart = size - UNCONDITIONAL_MATCHLEN - UNCOMPRESSED_END - 1;
int last_hashed = -1;
int hash;
int fetch = 0;
if ((source[0] & 1) != 1)
{
byte[] d2 = new byte[size];
System.arraycopy(source, headerLen(source), d2, 0, size);
return d2;
}
if ((source[0] & 1) != 1) {
byte[] d2 = new byte[size];
System.arraycopy(source, headerLen(source), d2, 0, size);
return d2;
}
for (;;)
{
if (cword_val == 1)
{
cword_val = fast_read(source, src, 4);
src += 4;
if (dst <= last_matchstart)
{
if(level == 1)
fetch = (int)fast_read(source, src, 3);
else
fetch = (int)fast_read(source, src, 4);
}
}
for (; ; ) {
if (cword_val == 1) {
cword_val = fast_read(source, src, 4);
src += 4;
if (dst <= last_matchstart) {
if (level == 1)
fetch = (int) fast_read(source, src, 3);
else
fetch = (int) fast_read(source, src, 4);
}
}
if ((cword_val & 1) == 1)
{
int matchlen;
int offset2;
if ((cword_val & 1) == 1) {
int matchlen;
int offset2;
cword_val = cword_val >>> 1;
cword_val = cword_val >>> 1;
if (level == 1)
{
hash = (fetch >>> 4) & 0xfff;
offset2 = hashtable[hash];
if (level == 1) {
hash = (fetch >>> 4) & 0xfff;
offset2 = hashtable[hash];
if ((fetch & 0xf) != 0)
{
matchlen = (fetch & 0xf) + 2;
src += 2;
}
else
{
matchlen = ((int)source[src + 2]) & 0xff;
src += 3;
}
}
else
{
int offset;
if ((fetch & 0xf) != 0) {
matchlen = (fetch & 0xf) + 2;
src += 2;
} else {
matchlen = ((int) source[src + 2]) & 0xff;
src += 3;
}
} else {
int offset;
if ((fetch & 3) == 0)
{
offset = (fetch & 0xff) >>> 2;
matchlen = 3;
src++;
}
else if ((fetch & 2) == 0)
{
offset = (fetch & 0xffff) >>> 2;
matchlen = 3;
src += 2;
}
else if ((fetch & 1) == 0)
{
offset = (fetch & 0xffff) >>> 6;
matchlen = ((fetch >>> 2) & 15) + 3;
src += 2;
}
else if ((fetch & 127) != 3)
{
offset = (fetch >>> 7) & 0x1ffff;
matchlen = ((fetch >>> 2) & 0x1f) + 2;
src += 3;
}
else
{
offset = (fetch >>> 15);
matchlen = ((fetch >>> 7) & 255) + 3;
src += 4;
}
offset2 = (int)(dst - offset);
}
if ((fetch & 3) == 0) {
offset = (fetch & 0xff) >>> 2;
matchlen = 3;
src++;
} else if ((fetch & 2) == 0) {
offset = (fetch & 0xffff) >>> 2;
matchlen = 3;
src += 2;
} else if ((fetch & 1) == 0) {
offset = (fetch & 0xffff) >>> 6;
matchlen = ((fetch >>> 2) & 15) + 3;
src += 2;
} else if ((fetch & 127) != 3) {
offset = (fetch >>> 7) & 0x1ffff;
matchlen = ((fetch >>> 2) & 0x1f) + 2;
src += 3;
} else {
offset = (fetch >>> 15);
matchlen = ((fetch >>> 7) & 255) + 3;
src += 4;
}
offset2 = (int) (dst - offset);
}
destination[dst + 0] = destination[offset2 + 0];
destination[dst + 1] = destination[offset2 + 1];
destination[dst + 2] = destination[offset2 + 2];
destination[dst + 0] = destination[offset2 + 0];
destination[dst + 1] = destination[offset2 + 1];
destination[dst + 2] = destination[offset2 + 2];
for (int i = 3; i < matchlen; i += 1)
{
destination[dst + i] = destination[offset2 + i];
}
dst += matchlen;
for (int i = 3; i < matchlen; i += 1) {
destination[dst + i] = destination[offset2 + i];
}
dst += matchlen;
if (level == 1)
{
fetch = (int)fast_read(destination, last_hashed + 1, 3); // destination[last_hashed + 1] | (destination[last_hashed + 2] << 8) | (destination[last_hashed + 3] << 16);
while (last_hashed < dst - matchlen)
{
last_hashed++;
hash = ((fetch >>> 12) ^ fetch) & (HASH_VALUES - 1);
hashtable[hash] = last_hashed;
hash_counter[hash] = 1;
fetch = fetch >>> 8 & 0xffff | (((int)destination[last_hashed + 3]) & 0xff) << 16;
}
fetch = (int)fast_read(source, src, 3);
}
else
{
fetch = (int)fast_read(source, src, 4);
}
last_hashed = dst - 1;
}
else
{
if (dst <= last_matchstart)
{
destination[dst] = source[src];
dst += 1;
src += 1;
cword_val = cword_val >>> 1;
if (level == 1) {
fetch = (int) fast_read(destination, last_hashed + 1, 3); // destination[last_hashed + 1] | (destination[last_hashed + 2] << 8) | (destination[last_hashed + 3] << 16);
while (last_hashed < dst - matchlen) {
last_hashed++;
hash = ((fetch >>> 12) ^ fetch) & (HASH_VALUES - 1);
hashtable[hash] = last_hashed;
hash_counter[hash] = 1;
fetch = fetch >>> 8 & 0xffff | (((int) destination[last_hashed + 3]) & 0xff) << 16;
}
fetch = (int) fast_read(source, src, 3);
} else {
fetch = (int) fast_read(source, src, 4);
}
last_hashed = dst - 1;
} else {
if (dst <= last_matchstart) {
destination[dst] = source[src];
dst += 1;
src += 1;
cword_val = cword_val >>> 1;
if (level == 1)
{
while (last_hashed < dst - 3)
{
last_hashed++;
int fetch2 = (int)fast_read(destination, last_hashed, 3);
hash = ((fetch2 >>> 12) ^ fetch2) & (HASH_VALUES - 1);
hashtable[hash] = last_hashed;
hash_counter[hash] = 1;
}
fetch = fetch >> 8 & 0xffff | (((int)source[src + 2]) & 0xff) << 16;
}
else
{
fetch = fetch >> 8 & 0xffff | (((int)source[src + 2]) & 0xff) << 16 | (((int)source[src + 3]) & 0xff) << 24;
}
}
else
{
while (dst <= size - 1)
{
if (cword_val == 1)
{
src += CWORD_LEN;
cword_val = 0x80000000L;
}
if (level == 1) {
while (last_hashed < dst - 3) {
last_hashed++;
int fetch2 = (int) fast_read(destination, last_hashed, 3);
hash = ((fetch2 >>> 12) ^ fetch2) & (HASH_VALUES - 1);
hashtable[hash] = last_hashed;
hash_counter[hash] = 1;
}
fetch = fetch >> 8 & 0xffff | (((int) source[src + 2]) & 0xff) << 16;
} else {
fetch = fetch >> 8 & 0xffff | (((int) source[src + 2]) & 0xff) << 16 | (((int) source[src + 3]) & 0xff) << 24;
}
} else {
while (dst <= size - 1) {
if (cword_val == 1) {
src += CWORD_LEN;
cword_val = 0x80000000L;
}
destination[dst] = source[src];
dst++;
src++;
cword_val = cword_val >>> 1;
}
return destination;
}
}
}
destination[dst] = source[src];
dst++;
src++;
cword_val = cword_val >>> 1;
}
return destination;
}
}
}
} catch (Throwable ex) {
throw new IOException(ex.getMessage() + " (" + Base64.getEncoder().encodeToString(source) + ")", ex);
}
}
}