notifyconnectioninforequest

2018-08-10 16:28:55 -04:00
parent 3d09c5ef21
commit 99961d84aa
7 changed files with 362 additions and 28 deletions
@@ -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;
+    }
 }
@@ -0,0 +1,7 @@
+package com.github.manevolent.ts3j.protocol;
+
+public enum PacketKind {
+    KEEPALIVE,
+    SPEECH,
+    CONTROL
+}
@@ -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<? extends PacketBody> packetClass;

+    private final PacketKind kind;
    private final PacketBodyType acknolwedgedBy;
    private final boolean encrypted, resend, splittable, compressible, mustEncrypt;

    PacketBodyType(int index, Class<? extends PacketBody> 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;
+    }
 }
@@ -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<DataPoint> sentPacketList = new LinkedList<>();
+    private final List<DataPoint> 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;
+        }
+    }
+}
@@ -0,0 +1,4 @@
+package com.github.manevolent.ts3j.protocol.packet.statistics;
+
+public class TimedMeasurement {
+}
@@ -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<PacketBodyType, Reassembly> reassemblyQueue = new ConcurrentHashMap<>();

+    private final Map<Integer, PacketResponse> pingQueue = new ConcurrentHashMap<>();
    private final Map<Integer, PacketResponse> sendQueue = new ConcurrentHashMap<>();
    private final Map<Integer, PacketResponse> 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<Double> pings = new LinkedList<>();
+    private Map<PacketKind, PacketStatistics> 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,9 +403,7 @@ public abstract class AbstractTeamspeakClientSocket
                        networkPacket.getHeader().getType() != PacketBodyType.PING;

        final PacketResponse response;
-
-        if (willWaitForAck) {
-            Map<Integer, PacketResponse> responsibleQueue;
+        final Map<Integer, PacketResponse> responsibleQueue;

        switch (packet.getHeader().getType()) {
            case COMMAND:
@@ -389,11 +412,14 @@ public abstract class AbstractTeamspeakClientSocket
            case COMMAND_LOW:
                responsibleQueue = sendQueueLow;
                break;
+            case PING:
+                responsibleQueue = pingQueue;
+                break;
            default:
-                    throw new IllegalArgumentException("no ack queue exists for " +
-                            packet.getHeader().getType().name());
+                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<Double, Double> 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 ++;
        }

@@ -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<Double, Double> 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;
            }
        }
    }