From 7c33cfa97a1f9c2078697c1799edb0b45c64ded8 Mon Sep 17 00:00:00 2001 From: Manevolent Date: Tue, 22 Jan 2019 21:07:51 -0500 Subject: [PATCH 1/2] - There was a bug that could cause TS3J to stop firing certain events to all event listeners if the command execution codes were generated quickly, sequentially by those event listener threads. This is now fixed. - There is now more detailed output for command futures when reconnecting: "client reconnecting" instead of "client disconnected" - The MusicPlayer example now returns exit code -1 when it is kicked/disconnected prematurely. --- .../ts3j/examples/audio/MusicPlayer.java | 10 +++ .../client/AbstractTeamspeakClientSocket.java | 24 +++--- .../client/LocalTeamspeakClientSocket.java | 80 ++++++++++++++----- 3 files changed, 82 insertions(+), 32 deletions(-) diff --git a/examples/audio/src/main/java/com/github/manevolent/ts3j/examples/audio/MusicPlayer.java b/examples/audio/src/main/java/com/github/manevolent/ts3j/examples/audio/MusicPlayer.java index 2c21829..daba276 100644 --- a/examples/audio/src/main/java/com/github/manevolent/ts3j/examples/audio/MusicPlayer.java +++ b/examples/audio/src/main/java/com/github/manevolent/ts3j/examples/audio/MusicPlayer.java @@ -4,6 +4,9 @@ import com.github.manevolent.ffmpeg4j.*; import com.github.manevolent.ffmpeg4j.filter.audio.FFmpegAudioResampleFilter; import com.github.manevolent.ffmpeg4j.source.FFmpegAudioSourceSubstream; import com.github.manevolent.ffmpeg4j.stream.source.FFmpegSourceStream; +import com.github.manevolent.ts3j.event.ClientUpdatedEvent; +import com.github.manevolent.ts3j.event.DisconnectedEvent; +import com.github.manevolent.ts3j.event.TS3Listener; import com.github.manevolent.ts3j.identity.LocalIdentity; import com.github.manevolent.ts3j.protocol.socket.client.LocalTeamspeakClientSocket; import com.github.manevolent.ts3j.util.Ts3Debugging; @@ -51,6 +54,13 @@ public class MusicPlayer { client.setNickname(MusicPlayer.class.getSimpleName()); client.setHWID("TestTestTest"); + client.addListener(new TS3Listener() { + @Override + public void onDisconnected(DisconnectedEvent e) { + System.exit(-1); + } + }); + // Create a sink TeamspeakFastMixerSink sink = new TeamspeakFastMixerSink( AUDIO_FORMAT, 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 b11112a..e5070c7 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 @@ -149,10 +149,12 @@ public abstract class AbstractTeamspeakClientSocket protected void start() { Thread networkThread = new Thread(networkReader); + networkThread.setName("TS3J/NetworkReader/" + networkThread.getId()); networkThread.setDaemon(true); networkThread.start(); Thread handlerThread = new Thread(networkHandler); + handlerThread.setName("TS3J/NetworkHandler/" + handlerThread.getId()); handlerThread.setDaemon(true); handlerThread.start(); } @@ -922,19 +924,21 @@ public abstract class AbstractTeamspeakClientSocket private class NetworkHandler implements Runnable { @Override public void run() { - while (isReading()) { - try { - Packet packet = readQueue.take(); + try { + while (isReading()) { + try { + Packet packet = readQueue.take(); - // separate these so we getHandler at proper runtime instant rather than get it and let it change - // on us + // separate these so we getHandler at proper runtime instant rather than get it and let it change + // on us - getHandler().handlePacket(packet); - } catch (Throwable e) { - if (e instanceof InterruptedException) continue; - - getExceptionHandler().accept(e); + getHandler().handlePacket(packet); + } catch (InterruptedException e) { + break; + } } + } catch (Throwable e) { + getExceptionHandler().accept(e); } } } 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 a69ce0d..0607929 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 @@ -55,10 +55,15 @@ public class LocalTeamspeakClientSocket private List listeners = new LinkedList<>(); - private int acceptingReturnCode = Integer.MAX_VALUE; - private Integer serverId = null; + /** + * The internal command buffer + * + * This buffer ascends through specific command codes, which are calculated using: + * + * get + */ private Map awaitingCommands = new ConcurrentHashMap<>(); public LocalTeamspeakClientSocket() { @@ -292,8 +297,6 @@ public class LocalTeamspeakClientSocket } awaitingCommands.clear(); - - calculateAcceptingReturnCode(); } serverId = null; @@ -372,10 +375,10 @@ public class LocalTeamspeakClientSocket synchronized (commandSendLock) { for (Integer id : new ArrayList<>(awaitingCommands.keySet())) { ClientCommandResponse response = awaitingCommands.get(id); - response.completeFailure(new IOException("client disconnected")); + response.completeFailure(new IOException("client reconnecting")); } + awaitingCommands.clear(); - calculateAcceptingReturnCode(); } setOption("client.hostname", remote.getHostString()); @@ -436,7 +439,7 @@ public class LocalTeamspeakClientSocket CommandProcessor awaitingCommandProcessor; synchronized (commandSendLock) { - awaitingCommandProcessor = awaitingCommands.get(acceptingReturnCode); + awaitingCommandProcessor = awaitingCommands.get(getAwaitingCommandCode()); } if (awaitingCommandProcessor != null) @@ -468,14 +471,14 @@ public class LocalTeamspeakClientSocket else if (getState() == ClientConnectionState.CONNECTING) throw new IOException("connecting"); - int maxReturnCode = calculateAcceptingReturnCode(); - if (maxReturnCode == Integer.MAX_VALUE) maxReturnCode = 0; - else maxReturnCode = maxReturnCode + 1; - int returnCode = maxReturnCode; + // Calculate the next command send code we should use + // This is so we can sequentially send commands and expect responses in order from the server + int sendCode = getNextCommandCode(); + if (sendCode >= Integer.MAX_VALUE) throw new ArrayIndexOutOfBoundsException("sendCode"); - command.add(new CommandSingleParameter("return_code", Integer.toString(returnCode))); + command.add(new CommandSingleParameter("return_code", Integer.toString(sendCode))); - awaitingCommands.put(returnCode, response = new ClientCommandResponse<>(returnCode, multiCommands -> { + awaitingCommands.put(sendCode, response = new ClientCommandResponse<>(sendCode, multiCommands -> { List commands = new LinkedList<>(); for (MultiCommand multiCommand : multiCommands) commands.addAll(multiCommand.simplify()); @@ -484,10 +487,14 @@ public class LocalTeamspeakClientSocket .collect(Collectors.toCollection(LinkedList::new)); }, command)); - acceptingReturnCode = calculateAcceptingReturnCode(); - - if (acceptingReturnCode == returnCode) - sendNextCommand(); + // Ensure that the next command we expect to receive a response for is sent to the network + // but, only send that command if the accepting receive code matches the code we picked above + // this ensures we don't double-send a command + int nextCommandCode = getAwaitingCommandCode(); + if (nextCommandCode < 0) // This should never happen + throw new IllegalStateException("nextCommandCode <= 0: " + nextCommandCode); + else if (nextCommandCode == sendCode) + sendCommand(nextCommandCode); } return response; @@ -508,17 +515,42 @@ public class LocalTeamspeakClientSocket } } - private int calculateAcceptingReturnCode() { + /** + * Gets the next command return_code we will create and buffer for sending. + * + * @return Next command send code + */ + private int getNextCommandCode() { + int highest = 0; + for (Integer integer : awaitingCommands.keySet()) highest = Math.max(highest, integer); + return highest + 1; + } + + /** + * Gets the next return_code we expect to receive from the server. + * + * @return command receive code + */ + private int getAwaitingCommandCode() { + if (awaitingCommands.size() <= 0) return -1; + int lowest = Integer.MAX_VALUE; for (Integer integer : awaitingCommands.keySet()) lowest = Math.min(lowest, integer); - return lowest; } + private void sendCommand(int code) + throws IOException, TimeoutException { + synchronized (commandSendLock) { + ClientCommandResponse currentResponse = awaitingCommands.get(code); + if (currentResponse != null) currentResponse.ensureSent(); + } + } + private void sendNextCommand() throws IOException, TimeoutException { synchronized (commandSendLock) { - ClientCommandResponse currentResponse = awaitingCommands.get(acceptingReturnCode); + ClientCommandResponse currentResponse = awaitingCommands.get(getAwaitingCommandCode()); if (currentResponse != null) currentResponse.ensureSent(); } } @@ -649,9 +681,13 @@ public class LocalTeamspeakClientSocket } catch (Throwable e) { completeFailure(e); } finally { - synchronized (commandSendLock) { + synchronized (commandSendLock) { // (this is likely already locked) + // Remove the current "awaiting" command. This is removing this instance, effectively. awaitingCommands.remove(getReturnCode()); - acceptingReturnCode = calculateAcceptingReturnCode(); + + // Ensure we send the next command + // This is essential to keep the command chain flowing when there are commands buffered in the + // command send (awaitingCommands) queue sendNextCommand(); } } From e84542c6bbbd9e1a586554634b87a6e7ecac5155 Mon Sep 17 00:00:00 2001 From: sump pump Date: Mon, 28 Jan 2019 15:23:22 -0500 Subject: [PATCH 2/2] Update README.md --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index 809bbc2..2f836ce 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,8 @@ Discord: https://discord.gg/4q9jpCa TS3J is an open-source implementation of the reverse-engineered Teamspeak3 full server/client protocol, as an adaptation of Splamy's C# TS3Client source code. You can find that here: https://github.com/Splamy/TS3AudioBot/. +A standalone proof-of-concept was created to wrap ts3j: https://github.com/Manevolent/ts3j-musicbot + (Shameless plug) If you want to deliver a great music bot, check out ffmpeg4j, a wrapper around the native C library FFmpeg, to get insanely low CPU usage on your YouTube video downloads (or streams if you don't want to hit disk), and audio playback: https://github.com/Manevolent/ffmpeg4j