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 6a5565d..1cd7176 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/Packet.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/Packet.java @@ -9,6 +9,7 @@ import java.nio.ByteBuffer; * Represents a fully contained network packet, traveling in or out. */ public final class Packet { + private final long creationNanotime = System.nanoTime(); private final ProtocolRole role; private PacketHeader header; @@ -131,4 +132,8 @@ public final class Packet { public int getSize() { return (getHeader() == null ? 0 : getHeader().getSize()) + (getBody() == null ? 0 : getBody().getSize()); } + + public long getCreatedNanotime() { + return creationNanotime; + } } diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/PacketKind.java b/src/main/java/com/github/manevolent/ts3j/protocol/PacketKind.java new file mode 100644 index 0000000..af1550b --- /dev/null +++ b/src/main/java/com/github/manevolent/ts3j/protocol/PacketKind.java @@ -0,0 +1,7 @@ +package com.github.manevolent.ts3j.protocol; + +public enum PacketKind { + KEEPALIVE, + SPEECH, + CONTROL +} diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBodyType.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBodyType.java index dec3c68..8e144b8 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBodyType.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBodyType.java @@ -1,24 +1,27 @@ package com.github.manevolent.ts3j.protocol.packet; +import com.github.manevolent.ts3j.protocol.PacketKind; + public enum PacketBodyType { - VOICE(0x0, PacketBody0Voice.class, null, true, false, false, false, false), - VOICE_WHISPER(0x1, PacketBody1VoiceWhisper.class, null, false, false, false, false, false), - PONG(0x5, PacketBody5Pong.class, null, false, false, false, false, false), - PING(0x4, PacketBody4Ping.class, PONG, false, false, false, false, false), - ACK(0x6, PacketBody6Ack.class, null, true, true, false, false, false), - ACK_LOW(0x7, PacketBody7AckLow.class, null, true, true, false, false, false), - INIT1(0x8, PacketBody8Init1.class, null, false, true, false, false, false), - COMMAND_LOW(0x3, PacketBody3CommandLow.class, ACK_LOW, true, true, true, true, true), - COMMAND(0x2, PacketBody2Command.class, ACK, true, true, true, true, true); + VOICE(0x0, PacketBody0Voice.class, PacketKind.SPEECH, null, true, false, false, false, false), + VOICE_WHISPER(0x1, PacketBody1VoiceWhisper.class, PacketKind.SPEECH, null, false, false, false, false, false), + PONG(0x5, PacketBody5Pong.class, PacketKind.KEEPALIVE, null, false, false, false, false, false), + PING(0x4, PacketBody4Ping.class, PacketKind.KEEPALIVE, PONG, false, false, false, false, false), + ACK(0x6, PacketBody6Ack.class, PacketKind.CONTROL, null, true, true, false, false, false), + ACK_LOW(0x7, PacketBody7AckLow.class, PacketKind.CONTROL, null, true, true, false, false, false), + INIT1(0x8, PacketBody8Init1.class, PacketKind.CONTROL, null, false, true, false, false, false), + COMMAND_LOW(0x3, PacketBody3CommandLow.class, PacketKind.CONTROL, ACK_LOW, true, true, true, true, true), + COMMAND(0x2, PacketBody2Command.class, PacketKind.CONTROL, ACK, true, true, true, true, true); private final int index; private final Class packetClass; + private final PacketKind kind; private final PacketBodyType acknolwedgedBy; private final boolean encrypted, resend, splittable, compressible, mustEncrypt; PacketBodyType(int index, Class packetClass, - PacketBodyType acknolwedgedBy, + PacketKind kind, PacketBodyType acknolwedgedBy, boolean encrypted, boolean resend, boolean splittable, @@ -26,6 +29,7 @@ public enum PacketBodyType { boolean mustEncrypt) { this.index = index; this.packetClass = packetClass; + this.kind = kind; this.acknolwedgedBy = acknolwedgedBy; this.resend = resend; this.encrypted = encrypted; @@ -72,4 +76,8 @@ public enum PacketBodyType { public boolean mustEncrypt() { return mustEncrypt; } + + public PacketKind getKind() { + return kind; + } } diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/statistics/PacketStatistics.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/statistics/PacketStatistics.java new file mode 100644 index 0000000..72a0d89 --- /dev/null +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/statistics/PacketStatistics.java @@ -0,0 +1,149 @@ + +package com.github.manevolent.ts3j.protocol.packet.statistics; + +import com.github.manevolent.ts3j.protocol.Packet; + +import java.util.LinkedList; +import java.util.List; + +public class PacketStatistics { + private final List sentPacketList = new LinkedList<>(); + private final List receivedPacketList = new LinkedList<>(); + + private int sentBytes; + private int sentPackets; + private int receivedBytes; + private int receivedPackets; + + public void processIncoming(Packet packet) { + int size = packet.getSize(); + + receivedBytes += size; + receivedPackets ++; + + receivedPacketList.add(new DataPoint(packet.getCreatedNanotime() / 1000L, packet.getSize())); + + long t = System.currentTimeMillis() - 60_000L; + receivedPacketList.removeIf(x -> x.getTime() < t); + } + + public void processOutgoing(Packet packet) { + int size = packet.getSize(); + + sentBytes += size; + sentPackets ++; + + sentPacketList.add(new DataPoint(packet.getCreatedNanotime() / 1000L, packet.getSize())); + + // Remove oldest + long t = System.currentTimeMillis() - 60_000L; + sentPacketList.removeIf(x -> x.getTime() < t); + } + + public int getSentPackets() { + return sentPackets; + } + + public int getSentBytes() { + return sentBytes; + } + + public int getSentPacketsLastSecond() { + long now = System.currentTimeMillis() - 1_000L; + + return (int) sentPacketList.stream() + .filter(x -> x.getTime() >= now) + .count(); + } + + public int getSentPacketsLastMinute() { + long now = System.currentTimeMillis() - 60_000L; + + return (int) sentPacketList.stream() + .filter(x -> x.getTime() >= now) + .count(); + } + + public int getReceivedPackets() { + return receivedPackets; + } + + public int getReceivedPacketsLastSecond() { + long now = System.currentTimeMillis() - 1_000L; + + return (int) receivedPacketList.stream() + .filter(x -> x.getTime() >= now) + .count(); + } + + public int getReceivedPacketsLastMinute() { + long now = System.currentTimeMillis() - 60_000L; + + return (int) receivedPacketList.stream() + .filter(x -> x.getTime() >= now) + .count(); + } + + + public int getReceivedBytes() { + return receivedBytes; + } + + public int getSentBytesLastMinute() { + long now = System.currentTimeMillis() - 60_000L; + + return sentPacketList.stream() + .filter(x -> x.getTime() >= now) + .map(DataPoint::getSize) + .reduce((a, b) -> a + b) + .orElse(0); + } + + public int getSentBytesLastSecond() { + long now = System.currentTimeMillis() - 1_000L; + + return sentPacketList.stream() + .filter(x -> x.getTime() >= now) + .map(DataPoint::getSize) + .reduce((a, b) -> a + b) + .orElse(0); + } + + public int getReceivedBytesLastMinute() { + long now = System.currentTimeMillis() - 60_000L; + + return receivedPacketList.stream() + .filter(x -> x.getTime() >= now) + .map(DataPoint::getSize) + .reduce((a, b) -> a + b) + .orElse(0); + } + + public int getReceivedBytesLastSecond() { + long now = System.currentTimeMillis() - 1_000L; + + return receivedPacketList.stream() + .filter(x -> x.getTime() >= now) + .map(DataPoint::getSize) + .reduce((a, b) -> a + b) + .orElse(0); + } + + private final class DataPoint { + private final long time; + private final int size; + + private DataPoint(long time, int size) { + this.time = time; + this.size = size; + } + + public long getTime() { + return time; + } + + public int getSize() { + return size; + } + } +} \ No newline at end of file diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/statistics/TimedMeasurement.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/statistics/TimedMeasurement.java new file mode 100644 index 0000000..5610427 --- /dev/null +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/statistics/TimedMeasurement.java @@ -0,0 +1,4 @@ +package com.github.manevolent.ts3j.protocol.packet.statistics; + +public class TimedMeasurement { +} 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 36dbe39..6d3c114 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 @@ -4,6 +4,7 @@ import com.github.manevolent.ts3j.command.*; import com.github.manevolent.ts3j.identity.Identity; import com.github.manevolent.ts3j.protocol.NetworkPacket; import com.github.manevolent.ts3j.protocol.Packet; +import com.github.manevolent.ts3j.protocol.PacketKind; import com.github.manevolent.ts3j.protocol.SocketRole; import com.github.manevolent.ts3j.protocol.client.ClientConnectionState; import com.github.manevolent.ts3j.protocol.header.ClientPacketHeader; @@ -12,6 +13,7 @@ import com.github.manevolent.ts3j.protocol.header.PacketHeader; import com.github.manevolent.ts3j.protocol.packet.*; import com.github.manevolent.ts3j.protocol.packet.handler.PacketHandler; import com.github.manevolent.ts3j.protocol.packet.handler.client.LocalClientHandler; +import com.github.manevolent.ts3j.protocol.packet.statistics.PacketStatistics; import com.github.manevolent.ts3j.protocol.packet.transformation.InitPacketTransformation; import com.github.manevolent.ts3j.protocol.packet.transformation.PacketTransformation; import com.github.manevolent.ts3j.protocol.socket.AbstractTeamspeakSocket; @@ -51,6 +53,7 @@ public abstract class AbstractTeamspeakClientSocket private final Map reassemblyQueue = new ConcurrentHashMap<>(); + private final Map pingQueue = new ConcurrentHashMap<>(); private final Map sendQueue = new ConcurrentHashMap<>(); private final Map sendQueueLow = new ConcurrentHashMap<>(); @@ -71,6 +74,19 @@ public abstract class AbstractTeamspeakClientSocket private long lastResponse; private long lastPing; + // RTT + private static final double alphaSmooth = 0.125D; + private static final double betaSmooth = 0.25D; + private double smoothedRtt; + private double smoothedRttVar; + private double currentRto; + // + + // Ping + private List pings = new LinkedList<>(); + private Map packetStatistics = new LinkedHashMap<>(); + // + private final NetworkReader networkReader = new NetworkReader(); private Thread networkThread = null; @@ -103,6 +119,15 @@ public abstract class AbstractTeamspeakClientSocket remoteSendCounter.put(bodyType, new RemoteCounterFull(65536, 100)); } } + + // Initialize statistics + for (PacketKind kind : PacketKind.values()) { + packetStatistics.put(kind, new PacketStatistics()); + } + } + + public PacketStatistics getStatistics(PacketKind kind) { + return packetStatistics.get(kind); } protected void start() { @@ -378,22 +403,23 @@ public abstract class AbstractTeamspeakClientSocket networkPacket.getHeader().getType() != PacketBodyType.PING; final PacketResponse response; + final Map responsibleQueue; - if (willWaitForAck) { - Map responsibleQueue; - - switch (packet.getHeader().getType()) { - case COMMAND: - responsibleQueue = sendQueue; - break; - case COMMAND_LOW: - responsibleQueue = sendQueueLow; - break; - default: - throw new IllegalArgumentException("no ack queue exists for " + - packet.getHeader().getType().name()); - } + switch (packet.getHeader().getType()) { + case COMMAND: + responsibleQueue = sendQueue; + break; + case COMMAND_LOW: + responsibleQueue = sendQueueLow; + break; + case PING: + responsibleQueue = pingQueue; + break; + default: + responsibleQueue = null; + } + if (responsibleQueue != null) { responsibleQueue.put( packet.getHeader().getPacketId(), response = new PacketResponse( @@ -409,8 +435,14 @@ public abstract class AbstractTeamspeakClientSocket writeNetworkPacket(networkPacket); + PacketKind kind = networkPacket.getHeader().getType().getKind(); + if (kind != null) + packetStatistics + .get(kind) + .processOutgoing(packet); + // If necessary, fulfill promise for the packet acknowledgement or throw TimeoutException - if (response != null) { + if (response != null && willWaitForAck) { while (response.getRetries() < response.getMaxTries() && getState() != ClientConnectionState.DISCONNECTED) { try { @@ -513,6 +545,13 @@ public abstract class AbstractTeamspeakClientSocket packet.readBody(packetBuffer); } + + PacketKind kind = networkPacket.getHeader().getType().getKind(); + if (kind != null) + packetStatistics + .get(kind) + .processIncoming(packet); + // Return to parent handler return packet; } @@ -633,11 +672,42 @@ public abstract class AbstractTeamspeakClientSocket break; case PONG: lastResponse = System.currentTimeMillis(); + response = pingQueue.get(((PacketBody5Pong)packet.getBody()).getPacketId()); + if (response == null) { + Ts3Debugging.debug("Unrecognized PONG: " + ((PacketBody6Ack) packet.getBody()).getPacketId()); + } else { + lastResponse = System.currentTimeMillis(); + } + handle = false; + break; default: handle = true; response = null; } + // Calculate RTT + if (response != null) { + long rttNano = packet.getCreatedNanotime() - response.getLastSent(); + double sampleRtt = rttNano / 1_000_000_000D; + + if (packet.getHeader().getType() == PacketBodyType.PONG) { + synchronized (pings) { + pings.add(sampleRtt); // add rtt + + while (pings.size() > 5) + pings.remove(5 - 1); // rm oldest + } + } + + if (smoothedRtt < 0) + smoothedRtt = sampleRtt; + else + smoothedRtt = (long)((1 - alphaSmooth) * smoothedRtt + alphaSmooth * sampleRtt); + + smoothedRttVar = (long)((1 - betaSmooth) * smoothedRttVar + betaSmooth * Math.abs(sampleRtt - smoothedRtt)); + currentRto = smoothedRtt + Math.max(0D, 4 * smoothedRttVar); + } + if (packet.getHeader().getType().canResend() && packet.getHeader().getType() != PacketBodyType.INIT1) { // Find if we already acknowledged this packet if (counter != null) @@ -653,6 +723,35 @@ public abstract class AbstractTeamspeakClientSocket throw new IllegalStateException("no longer reading"); } + /** + * Gets the ping + */ + public Pair getPing() { + synchronized (pings) { + if (pings.size() == 1) { + return new Pair<>(pings.get(0), 0D); + } else if (pings.size() > 1) { + double avg = pings.stream().reduce((a, b) -> (a + b) / 2D).orElse(0D); + double sum = 0; + int n = 0; + for (double val : pings) { + n ++; + sum += (val - avg) * (val - avg); + } + + double stdDev; + if (n > 1) + stdDev = Math.sqrt(sum / (n - 1)); + else + stdDev = 0; + + return new Pair<>(pings.get(0), stdDev); + } else { + return new Pair<>((double)TIMEOUT, 0D); + } + } + } + @Override public PacketTransformation getTransformation() { return transformation; @@ -788,7 +887,7 @@ public abstract class AbstractTeamspeakClientSocket this.sentPacket = sentPacket; this.responsibleQueue = responsibleQueue; this.maxTries = maxTries; - this.lastSent = System.currentTimeMillis(); + this.lastSent = System.nanoTime(); this.willResend = maxTries > 0; } @@ -798,7 +897,16 @@ public abstract class AbstractTeamspeakClientSocket public void resend() throws IOException { AbstractTeamspeakClientSocket.this.writeNetworkPacket(sentPacket); - setLastSent(System.currentTimeMillis()); + + long t = System.nanoTime(); + long before = lastSent; + long elapsed = t - before; + + setLastSent(t); + + double elapsedSeconds = elapsed / 1_000_000_000D; + currentRto += elapsedSeconds; + tries ++; } 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 77ae932..129f756 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 @@ -13,13 +13,16 @@ import com.github.manevolent.ts3j.event.*; import com.github.manevolent.ts3j.identity.LocalIdentity; import com.github.manevolent.ts3j.identity.Uid; import com.github.manevolent.ts3j.protocol.NetworkPacket; +import com.github.manevolent.ts3j.protocol.PacketKind; import com.github.manevolent.ts3j.protocol.ProtocolRole; import com.github.manevolent.ts3j.protocol.client.ClientConnectionState; import com.github.manevolent.ts3j.protocol.header.PacketHeader; import com.github.manevolent.ts3j.protocol.packet.PacketBody0Voice; import com.github.manevolent.ts3j.protocol.packet.PacketBody2Command; +import com.github.manevolent.ts3j.protocol.packet.statistics.PacketStatistics; import com.github.manevolent.ts3j.util.HighPrecisionRecurrentTask; import com.github.manevolent.ts3j.util.Ts3Debugging; +import javafx.util.Pair; import java.io.IOException; import java.net.*; @@ -33,6 +36,7 @@ import java.util.stream.Collectors; public class LocalTeamspeakClientSocket extends AbstractTeamspeakClientSocket implements CommandProcessor { + private final DatagramSocket socket; { @@ -99,6 +103,51 @@ public class LocalTeamspeakClientSocket } } }); + + namedProcessors.put("notifyconnectioninforequest", new CommandProcessor() { + @Override + public void process(AbstractTeamspeakClientSocket client, SingleCommand singleCommand) + throws CommandProcessException { + SingleCommand response = new SingleCommand("setconnectioninfo", getRole().getOut()); + + Pair ping = getPing(); + response.add(new CommandSingleParameter("connection_ping", + Integer.toString((int)(double)ping.getKey()))); + response.add(new CommandSingleParameter("connection_ping_deviation", + Integer.toString((int)(double)ping.getValue()))); + + for (PacketKind kind : PacketKind.values()) { + PacketStatistics stats = getStatistics(kind); + String name = kind.name().toLowerCase(); + response.add(new CommandSingleParameter("connection_packets_sent_" + name, + Integer.toString(stats.getSentPackets()))); + response.add(new CommandSingleParameter("connection_packets_received_" + name, + Integer.toString(stats.getReceivedPackets()))); + response.add(new CommandSingleParameter("connection_bytes_sent_" + name, + Integer.toString(stats.getSentBytes()))); + response.add(new CommandSingleParameter("connection_bytes_received_" + name, + Integer.toString(stats.getReceivedBytes()))); + response.add(new CommandSingleParameter("connection_server2client_packetloss_" + name, + Integer.toString(0))); + response.add(new CommandSingleParameter("connection_bandwidth_sent_last_second_" + name, + Integer.toString(stats.getSentBytesLastSecond()))); + response.add(new CommandSingleParameter("connection_bandwidth_sent_last_minute_" + name, + Integer.toString(stats.getSentBytesLastMinute()))); + response.add(new CommandSingleParameter("connection_bandwidth_received_last_second_" + name, + Integer.toString(stats.getReceivedBytesLastSecond()))); + response.add(new CommandSingleParameter("connection_bandwidth_received_last_minute_" + name, + Integer.toString(stats.getReceivedBytesLastMinute()))); + } + + Ts3Debugging.debug(response.build()); + + try { + writePacket(new PacketBody2Command(getRole().getOut(), response)); + } catch (Exception e) { + Ts3Debugging.debug("Problem sending setconnectioninfo", e); + } + } + }); } public void addListener(TS3Listener listener) { @@ -436,7 +485,7 @@ public class LocalTeamspeakClientSocket return command; } - public void ensureSent() + public boolean ensureSent() throws IOException, TimeoutException { synchronized (sendLock) { if (!sent) { @@ -445,7 +494,11 @@ public class LocalTeamspeakClientSocket ); sent = true; + + return true; } + + return false; } } }