From 1236221f6b9e3be1261fd941fc847c7800794306 Mon Sep 17 00:00:00 2001 From: Manevolent Date: Thu, 2 Aug 2018 19:52:44 -0400 Subject: [PATCH] Commands now spool sequentially. Command ACK queues no longer collide with pings or ack_low. Getter more stable. --- .../response/AbstractCommandResponse.java | 2 - .../command/response/CommandResponse.java | 4 + .../protocol/packet/PacketBody0Voice.java | 2 + .../client/AbstractTeamspeakClientSocket.java | 64 ++++--- .../client/LocalTeamspeakClientSocket.java | 172 +++++++++--------- 5 files changed, 132 insertions(+), 112 deletions(-) diff --git a/src/main/java/com/github/manevolent/ts3j/command/response/AbstractCommandResponse.java b/src/main/java/com/github/manevolent/ts3j/command/response/AbstractCommandResponse.java index 1067054..c2f1f67 100644 --- a/src/main/java/com/github/manevolent/ts3j/command/response/AbstractCommandResponse.java +++ b/src/main/java/com/github/manevolent/ts3j/command/response/AbstractCommandResponse.java @@ -36,12 +36,10 @@ public abstract class AbstractCommandResponse implements CommandResponse { @Override public T get() throws ExecutionException, InterruptedException { - Ts3Debugging.debug("WAITING ON " + future); return map(future.get()); } protected void completeSuccess(Iterable commands) { - Ts3Debugging.debug("SUCCEEDING " + future); future.complete(commands); } diff --git a/src/main/java/com/github/manevolent/ts3j/command/response/CommandResponse.java b/src/main/java/com/github/manevolent/ts3j/command/response/CommandResponse.java index 8712dcb..834f202 100644 --- a/src/main/java/com/github/manevolent/ts3j/command/response/CommandResponse.java +++ b/src/main/java/com/github/manevolent/ts3j/command/response/CommandResponse.java @@ -1,10 +1,14 @@ package com.github.manevolent.ts3j.command.response; +import com.github.manevolent.ts3j.command.Command; + import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; public interface CommandResponse { + Command getCommand(); + default void complete() throws ExecutionException, InterruptedException { get(); diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java index 173241f..c620692 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java @@ -53,6 +53,8 @@ public class PacketBody0Voice extends PacketBody { @Override public void read(ByteBuffer buffer) { + Ts3Debugging.debug(buffer.array()); + packetId = buffer.getShort() & 0x0000FFFF; if (getRole() == ProtocolRole.SERVER) 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 8b95601..ef04b09 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 @@ -41,10 +41,14 @@ public abstract class AbstractTeamspeakClientSocket 0x54, 0x53, 0x33, 0x49, 0x4E, 0x49, 0x54, 0x31 }; - private final Object connectionStateLock = new Object(); - private final LinkedBlockingDeque readQueue = new LinkedBlockingDeque<>(65536); - private final Map reassemblyQueue = new ConcurrentHashMap<>(); private final Map clientOptions = new ConcurrentHashMap<>(); + + private final Object connectionStateLock = new Object(); + + private final LinkedBlockingDeque readQueue = new LinkedBlockingDeque<>(65536); + + private final Map reassemblyQueue = new ConcurrentHashMap<>(); + private final Map sendQueue = new ConcurrentHashMap<>(); private final Map sendQueueLow = new ConcurrentHashMap<>(); @@ -65,8 +69,9 @@ public abstract class AbstractTeamspeakClientSocket private long lastPing; private final NetworkReader networkReader = new NetworkReader(); - private final NetworkHandler networkHandler = new NetworkHandler(); private Thread networkThread = null; + + private final NetworkHandler networkHandler = new NetworkHandler(); private Thread handlerThread = null; private boolean reading = false; @@ -365,11 +370,28 @@ public abstract class AbstractTeamspeakClientSocket ); // Find if the packet must be acknowledged and create a promise for that action - boolean willAcknowledge = packet.getHeader().getType().getAcknolwedgedBy() != null; + boolean willWaitForAck = + packet.getHeader().getType().getAcknolwedgedBy() != null && + networkPacket.getHeader().getType() != PacketBodyType.PING; + final PacketResponse response; - if (willAcknowledge) - sendQueue.put( + 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()); + } + + responsibleQueue.put( packet.getHeader().getPacketId(), response = new PacketResponse( networkPacket, @@ -377,7 +399,7 @@ public abstract class AbstractTeamspeakClientSocket networkPacket.getHeader().getType().canResend() ? 10 : 0 ) ); - else + } else response = null; Ts3Debugging.debug("[PROTOCOL] WRITE " + networkPacket.getHeader().getType().name()); @@ -385,7 +407,7 @@ public abstract class AbstractTeamspeakClientSocket writeNetworkPacket(networkPacket); // If necessary, fulfill promise for the packet acknowledgement or throw TimeoutException - if (response != null && networkPacket.getHeader().getType() != PacketBodyType.PING) { + if (response != null) { while (response.getRetries() < response.getMaxTries()) { try { if (getState() == ClientConnectionState.DISCONNECTED) @@ -535,8 +557,6 @@ public abstract class AbstractTeamspeakClientSocket if (ackType != null) { switch (ackType) { case ACK: - Ts3Debugging.debug("[PROTOCOL] Acknowledge " + packet + " " + packet.getHeader().getType().name() - + " id=" + packet.getHeader().getPacketId()); writePacket(new PacketBody6Ack(getRole().getOut(), packet.getHeader().getPacketId())); break; case ACK_LOW: @@ -567,20 +587,20 @@ public abstract class AbstractTeamspeakClientSocket if (packet == null) continue; } - // Find if the packet is itself an acknowledgement + // Find if the packet is itself an acknowledgement (ACK or ACK_LOW) boolean handle; final PacketResponse response; switch (packet.getHeader().getType()) { case ACK: response = sendQueue.get(((PacketBody6Ack)packet.getBody()).getPacketId()); + if (response == null) + Ts3Debugging.debug("Unrecognized ACK: " + ((PacketBody6Ack)packet.getBody()).getPacketId()); handle = false; break; case ACK_LOW: response = sendQueueLow.get(((PacketBody7AckLow)packet.getBody()).getPacketId()); - handle = false; - break; - case PONG: - response = null; + if (response == null) + Ts3Debugging.debug("Unrecognized ACK_LOW: " + ((PacketBody6Ack)packet.getBody()).getPacketId()); handle = false; break; default: @@ -590,14 +610,14 @@ public abstract class AbstractTeamspeakClientSocket if (packet.getHeader().getType().canResend() && packet.getHeader().getType() != PacketBodyType.INIT1) { // Find if we already acknowledged this packet - handle = handle & counter.put(packet.getHeader().getPacketId()); + if (counter != null) + handle = handle & counter.put(packet.getHeader().getPacketId()); } if (response != null) { response.acknowledge(packet); - } else if (handle) { - return packet; } + if (handle) return packet; } throw new IllegalStateException("no longer reading"); @@ -947,7 +967,6 @@ public abstract class AbstractTeamspeakClientSocket if (amountMoved > 0) { // should always be true int toMove = Math.max(0, bufferSize - amountMoved); - Ts3Debugging.debug(Arrays.toString(buffer)); if (toMove > 0 && toMove < bufferSize) System.arraycopy(buffer, amountMoved, buffer, 0, toMove); @@ -955,7 +974,6 @@ public abstract class AbstractTeamspeakClientSocket for (int i = bufferSize - toNullify; i < bufferSize; i ++) buffer[i] = null; - Ts3Debugging.debug(Arrays.toString(buffer)); int bufferStartGeneration = bufferStart.getKey(); int bufferStartPosition = bufferStart.getValue(); @@ -992,8 +1010,6 @@ public abstract class AbstractTeamspeakClientSocket } } - - public static class RemoteCounterFullModulated extends RemoteCounterFull { private final int modulationSize; private final int modulationsPerGeneration; @@ -1035,8 +1051,10 @@ public abstract class AbstractTeamspeakClientSocket public interface LocalCounter { Pair next(); + int getPacketId(); boolean setPacketId(int packetId); + int getGeneration(); void setGeneration(int i); 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 976cbaa..40758aa 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 @@ -59,7 +59,7 @@ public class LocalTeamspeakClientSocket private int acceptingReturnCode = 0; private int lastReturnCode = 0; - private Map awaitingCommands = new ConcurrentHashMap<>(); + private Map awaitingCommands = new ConcurrentHashMap<>(); public LocalTeamspeakClientSocket() { super(); @@ -204,7 +204,7 @@ public class LocalTeamspeakClientSocket processor.process(client, singleCommand); } else if (singleCommand.getName().startsWith("notify")) { TS3Event event = TS3Event.createEvent(singleCommand); - for (TS3Listener listener : listeners) event.fire(listener); + listeners.forEach(event::fire); } else { CommandProcessor awaitingCommandProcessor; @@ -218,49 +218,18 @@ public class LocalTeamspeakClientSocket } public CommandResponse> getChannels() throws IOException, TimeoutException { - return executeMappedCommand( + return executeCommand( new SingleCommand("channellist", ProtocolRole.CLIENT), command -> new Channel(command.toMap()) ); } - private CommandResponse> executeCommand(Command command) + private CommandResponse> executeCommand(Command command) throws IOException, TimeoutException { return executeCommand(command, Function.identity()); } - private CommandResponse executeCommand( - Command command, - Function, T> processor - ) throws IOException, TimeoutException { - int returnCode; - - final ClientCommandResponse response; - - // sync may not be necessary here - synchronized (commandSendLock) { - returnCode = lastReturnCode++; - awaitingCommands.put(returnCode, response = new ClientCommandResponse<>(returnCode, processor)); - acceptingReturnCode = calculateAcceptingReturnCode(); - } - - try { - command.add(new CommandSingleParameter("return_code", Integer.toString(returnCode))); - - writePacket(new PacketBody2Command(getRole().getOut(), command)); - } catch (Throwable e) { - synchronized (commandSendLock) { - awaitingCommands.remove(returnCode); - acceptingReturnCode = calculateAcceptingReturnCode(); - } - - throw e; - } - - return response; - } - - private CommandResponse> executeMappedCommand( + private CommandResponse> executeCommand( Command command, Function processor ) throws IOException, TimeoutException { @@ -277,26 +246,15 @@ public class LocalTeamspeakClientSocket List commands = new LinkedList<>(); for (MultiCommand multiCommand : multiCommands) commands.addAll(multiCommand.simplify()); - List results = new LinkedList<>(); - for (SingleCommand command1 : commands) - results.add(processor.apply(command1)); - - return results; - })); + return commands.stream() + .map(processor::apply) + .collect(Collectors.toCollection(LinkedList::new)); + }, command)); acceptingReturnCode = calculateAcceptingReturnCode(); - } - try { - writePacket(new PacketBody2Command(getRole().getOut(), command)); - } catch (Throwable e) { - synchronized (commandSendLock) { - awaitingCommands.remove(returnCode); - - acceptingReturnCode = calculateAcceptingReturnCode(); - } - - throw e; + if (acceptingReturnCode == returnCode) + sendNextCommand(); } return response; @@ -324,6 +282,14 @@ public class LocalTeamspeakClientSocket return lowest; } + private void sendNextCommand() + throws IOException, TimeoutException { + synchronized (commandSendLock) { + ClientCommandResponse currentResponse = awaitingCommands.get(acceptingReturnCode); + if (currentResponse != null) currentResponse.ensureSent(); + } + } + private class MicrophoneTask implements Runnable { @Override public void run() { @@ -359,12 +325,18 @@ public class LocalTeamspeakClientSocket extends AbstractCommandResponse implements CommandProcessor { private final int returnCode; + private final Command command; private final List commands = new LinkedList<>(); + private Object sendLock = new Object(); + private boolean sent = false; + public ClientCommandResponse(int returnCode, - Function, T> processor) { + Function, T> processor, + Command command) { super(processor); this.returnCode = returnCode; + this.command = command; } @Override @@ -372,7 +344,11 @@ public class LocalTeamspeakClientSocket MultiCommand multiCommand) throws CommandProcessException { if (multiCommand.getName().equalsIgnoreCase("error")) - handleError(multiCommand.simplifyOne()); + try { + handleError(multiCommand.simplifyOne()); + } catch (IOException | TimeoutException e) { + throw new CommandProcessException(e); + } else commands.add(multiCommand); } @@ -382,32 +358,58 @@ public class LocalTeamspeakClientSocket SingleCommand command) throws CommandProcessException { if (command.getName().equalsIgnoreCase("error")) - handleError(command); + try { + handleError(command); + } catch (IOException | TimeoutException e) { + throw new CommandProcessException(e); + } else commands.add(new MultiCommand(command.getName(), ProtocolRole.CLIENT, command)); } - private void handleError(SingleCommand command) { - int errorId = Integer.parseInt(command.get("id").getValue()); + private void handleError(SingleCommand command) throws IOException, TimeoutException { + try { + int errorId = Integer.parseInt(command.get("id").getValue()); - synchronized (awaitingCommands) { - awaitingCommands.remove(returnCode); - acceptingReturnCode = calculateAcceptingReturnCode(); - } + synchronized (commandSendLock) { + awaitingCommands.remove(returnCode); + acceptingReturnCode = calculateAcceptingReturnCode(); + } - switch (errorId) { - case 0: - completeSuccess(commands); - break; - default: - completeFailure(new CommandException(command.get("msg").getValue(), errorId)); - break; + switch (errorId) { + case 0: + completeSuccess(commands); + break; + default: + completeFailure(new CommandException(command.get("msg").getValue(), errorId)); + break; + } + } finally { + sendNextCommand(); } } public int getReturnCode() { return returnCode; } + + @Override + public Command getCommand() { + return command; + } + + public void ensureSent() + throws IOException, TimeoutException { + synchronized (sendLock) { + if (!sent) { + LocalTeamspeakClientSocket.this.writePacket( + new PacketBody2Command(ProtocolRole.CLIENT, command) + ); + + sent = true; + } + } + } } private class ChannelListFinishedHandler implements CommandProcessor { @@ -461,7 +463,7 @@ public class LocalTeamspeakClientSocket if (timeInSeconds != null) banCommand.add(new CommandSingleParameter("time", timeInSeconds.toString())); if (reason != null) banCommand.add(new CommandSingleParameter("banreason", reason)); - return executeMappedCommand(banCommand, command -> new Ban(command.toMap())).get(); + return executeCommand(banCommand, command -> new Ban(command.toMap())).get(); } public Iterable banDatabaseClient(int dbid, Integer timeInSeconds, String reason) @@ -472,7 +474,7 @@ public class LocalTeamspeakClientSocket if (timeInSeconds != null) banCommand.add(new CommandSingleParameter("time", timeInSeconds.toString())); if (reason != null) banCommand.add(new CommandSingleParameter("banreason", reason)); - return executeMappedCommand(banCommand, command -> new Ban(command.toMap())).get(); + return executeCommand(banCommand, command -> new Ban(command.toMap())).get(); } public Iterable banClient(Uid uid, Integer timeInSeconds, String reason) @@ -483,7 +485,7 @@ public class LocalTeamspeakClientSocket if (timeInSeconds != null) banCommand.add(new CommandSingleParameter("time", timeInSeconds.toString())); if (reason != null) banCommand.add(new CommandSingleParameter("banreason", reason)); - return executeMappedCommand(banCommand, command -> new Ban(command.toMap())).get(); + return executeCommand(banCommand, command -> new Ban(command.toMap())).get(); } public Iterable banClient(Client client, Integer timeInSeconds, String reason) @@ -509,7 +511,7 @@ public class LocalTeamspeakClientSocket public Iterable listBans() throws InterruptedException, ExecutionException, TimeoutException, CommandException, IOException { Command banCommand = new SingleCommand("banlist", ProtocolRole.CLIENT); - return executeMappedCommand(banCommand, command -> new Ban(command.toMap())).get(); + return executeCommand(banCommand, command -> new Ban(command.toMap())).get(); } public void deleteBan(Ban ban) @@ -517,8 +519,8 @@ public class LocalTeamspeakClientSocket deleteBan(ban.getId()); } - public CommandResponse> channelAddPermission(int channelId, Permission... permission) - throws IOException, TimeoutException { + public void channelAddPermission(int channelId, Permission... permission) + throws IOException, TimeoutException, ExecutionException, InterruptedException { MultiCommand command = new MultiCommand( "channeladdperm", ProtocolRole.CLIENT, @@ -531,13 +533,11 @@ public class LocalTeamspeakClientSocket command.add(new CommandSingleParameter("cid", Integer.toString(channelId))); - return executeCommand(command); + executeCommand(command).complete(); } - public CommandResponse> channelClientAddPermission(int channelId, - int clientDatabaseId, - Permission... permission) - throws IOException, TimeoutException { + public void channelClientAddPermission(int channelId, int clientDatabaseId, Permission... permission) + throws IOException, TimeoutException, ExecutionException, InterruptedException { MultiCommand command = new MultiCommand( "channelclientaddperm", ProtocolRole.CLIENT, @@ -551,12 +551,10 @@ public class LocalTeamspeakClientSocket command.add(new CommandSingleParameter("cid", Integer.toString(channelId))); command.add(new CommandSingleParameter("cldbid", Integer.toString(channelId))); - return executeCommand(command); + executeCommand(command).complete(); } - public void channelClientDeletePermission(int channelId, - int clientDatabaseId, - Permission... permission) + public void channelClientDeletePermission(int channelId, int clientDatabaseId, Permission... permission) throws IOException, TimeoutException, ExecutionException, InterruptedException { MultiCommand command = new MultiCommand( "channeldelperm", @@ -596,7 +594,7 @@ public class LocalTeamspeakClientSocket command.add(new CommandSingleParameter("cid", Integer.toString(channelId))); - return executeMappedCommand(command, x -> new Client(x.toMap())).get(); + return executeCommand(command, x -> new Client(x.toMap())).get(); } /** @@ -629,7 +627,7 @@ public class LocalTeamspeakClientSocket throws IOException, TimeoutException, ExecutionException, InterruptedException { Command command = new SingleCommand("channellist", ProtocolRole.CLIENT); - return executeMappedCommand(command, x -> new Channel(x.toMap())).get(); + return executeCommand(command, x -> new Channel(x.toMap())).get(); } /** @@ -641,7 +639,7 @@ public class LocalTeamspeakClientSocket throws IOException, TimeoutException, ExecutionException, InterruptedException { Command command = new SingleCommand("clientlist", ProtocolRole.CLIENT); - return executeMappedCommand(command, x -> new Client(x.toMap())).get(); + return executeCommand(command, x -> new Client(x.toMap())).get(); } /**