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

Commands now spool sequentially. Command ACK queues no longer collide with pings or ack_low. Getter more stable.

Browse Source
This commit is contained in:
Manevolent
2018-08-02 19:52:44 -04:00
parent 964936abdf
commit 1236221f6b
5 changed files with 132 additions and 112 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
View File
@@ -36,12 +36,10 @@ public abstract class AbstractCommandResponse<T> implements CommandResponse<T> {
@Override
public T get()
throws ExecutionException, InterruptedException {
Ts3Debugging.debug("WAITING ON " + future);
return map(future.get());
}
protected void completeSuccess(Iterable<MultiCommand> commands) {
Ts3Debugging.debug("SUCCEEDING " + future);
future.complete(commands);
}
src/main/java/com/github/manevolent/ts3j/command/response/CommandResponse.java
+4
View File
@@ -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<T> {
Command getCommand();
default void complete()
throws ExecutionException, InterruptedException {
get();
src/main/java/com/github/manevolent/ts3j/protocol/packet/PacketBody0Voice.java
+2
View File
@@ -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)
src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java
+40 -22
View File
@@ -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<Packet> readQueue = new LinkedBlockingDeque<>(65536);
private final Map<PacketBodyType, Reassembly> reassemblyQueue = new ConcurrentHashMap<>();
private final Map<String, Object> clientOptions = new ConcurrentHashMap<>();
private final Object connectionStateLock = new Object();
private final LinkedBlockingDeque<Packet> readQueue = new LinkedBlockingDeque<>(65536);
private final Map<PacketBodyType, Reassembly> reassemblyQueue = new ConcurrentHashMap<>();
private final Map<Integer, PacketResponse> sendQueue = new ConcurrentHashMap<>();
private final Map<Integer, PacketResponse> 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<Integer, PacketResponse> 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
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<Integer, Integer> next();
int getPacketId();
boolean setPacketId(int packetId);
int getGeneration();
void setGeneration(int i);
src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java
+72 -74
View File
@@ -59,7 +59,7 @@ public class LocalTeamspeakClientSocket
private int acceptingReturnCode = 0;
private int lastReturnCode = 0;
private Map<Integer, CommandProcessor> awaitingCommands = new ConcurrentHashMap<>();
private Map<Integer, ClientCommandResponse> 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<Iterable<Channel>> getChannels() throws IOException, TimeoutException {
return executeMappedCommand(
return executeCommand(
new SingleCommand("channellist", ProtocolRole.CLIENT),
command -> new Channel(command.toMap())
);
}
private CommandResponse<Iterable<MultiCommand>> executeCommand(Command command)
private CommandResponse<Iterable<SingleCommand>> executeCommand(Command command)
throws IOException, TimeoutException {
return executeCommand(command, Function.identity());
}
private <T> CommandResponse<T> executeCommand(
Command command,
Function<Iterable<MultiCommand>, T> processor
) throws IOException, TimeoutException {
int returnCode;
final ClientCommandResponse<T> 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 <T> CommandResponse<Iterable<T>> executeMappedCommand(
private <T> CommandResponse<Iterable<T>> executeCommand(
Command command,
Function<SingleCommand, T> processor
) throws IOException, TimeoutException {
@@ -277,26 +246,15 @@ public class LocalTeamspeakClientSocket
List<SingleCommand> commands = new LinkedList<>();
for (MultiCommand multiCommand : multiCommands) commands.addAll(multiCommand.simplify());
List<T> 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<T>
implements CommandProcessor {
private final int returnCode;
private final Command command;
private final List<MultiCommand> commands = new LinkedList<>();
private Object sendLock = new Object();
private boolean sent = false;
public ClientCommandResponse(int returnCode,
Function<Iterable<MultiCommand>, T> processor) {
Function<Iterable<MultiCommand>, 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"))
try {
handleError(multiCommand.simplifyOne());
} catch (IOException | TimeoutException e) {
throw new CommandProcessException(e);
}
else
commands.add(multiCommand);
}
@@ -382,15 +358,20 @@ public class LocalTeamspeakClientSocket
SingleCommand command)
throws CommandProcessException {
if (command.getName().equalsIgnoreCase("error"))
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) {
private void handleError(SingleCommand command) throws IOException, TimeoutException {
try {
int errorId = Integer.parseInt(command.get("id").getValue());
synchronized (awaitingCommands) {
synchronized (commandSendLock) {
awaitingCommands.remove(returnCode);
acceptingReturnCode = calculateAcceptingReturnCode();
}
@@ -403,11 +384,32 @@ public class LocalTeamspeakClientSocket
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<Ban> 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<Ban> 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<Ban> banClient(Client client, Integer timeInSeconds, String reason)
@@ -509,7 +511,7 @@ public class LocalTeamspeakClientSocket
public Iterable<Ban> 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<Iterable<MultiCommand>> 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<Iterable<MultiCommand>> 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();
}
/**