diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java index c1036b9..e313725 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/fragment/PacketReassembly.java @@ -2,6 +2,8 @@ package com.github.manevolent.ts3j.protocol.packet.fragment; import com.github.manevolent.ts3j.protocol.Packet; import com.github.manevolent.ts3j.protocol.header.HeaderFlag; +import com.github.manevolent.ts3j.protocol.socket.client.AbstractTeamspeakClientSocket; +import com.github.manevolent.ts3j.util.Pair; import com.github.manevolent.ts3j.util.QuickLZ; import java.io.IOException; @@ -10,20 +12,18 @@ import java.util.*; import java.util.stream.Collectors; public class PacketReassembly { + private final AbstractTeamspeakClientSocket.LocalCounter counter = + new AbstractTeamspeakClientSocket.LocalCounterFull(65536, true); + private final Map queue = new LinkedHashMap<>(); - private int currentPacketId = 0; - public void setPacketId(int currentPacketId) { - this.currentPacketId = currentPacketId; - } - - public boolean put(Packet packet) throws IOException { + public void put(Packet packet) throws IOException { queue.put(packet.getHeader().getPacketId(), packet); - - return currentPacketId == packet.getHeader().getPacketId(); } public Packet next() throws IOException { + if (queue.size() <= 0) return null; + List reassemblyList = new ArrayList<>(); try { @@ -32,33 +32,38 @@ public class PacketReassembly { boolean state = false; - for (int packetId = currentPacketId; ; packetId++) { - if (packetId >= 65536) packetId = 0; + AbstractTeamspeakClientSocket.LocalCounter temporaryCounter = + new AbstractTeamspeakClientSocket.LocalCounterFull(65536, true); - Packet packet = queue.get(packetId); + Pair currentPacketId = counter.current(); + + temporaryCounter.setPacketId(currentPacketId.getKey()); + temporaryCounter.setGeneration(currentPacketId.getValue()); + + while (true) { + Pair thisPacketId = temporaryCounter.current(); + Packet packet = queue.get(thisPacketId.getKey()); if (packet == null) break; // Chain breaks here else { - packetIds.add(packetId); + packetIds.add(thisPacketId.getKey()); reassemblyList.add(packet); if (packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED)) { state = !state; if (!state) { - currentPacketId += packetIds.size(); break; // fragmentation stopped, we found the last packet - } else - continue; // keep collecting fragments + } } else if (!state) { - currentPacketId ++; break; // don't reassemble this packet, it's not a fragment } else if (state) { // keep collecting fragments - continue; } } + + temporaryCounter.next(); } if (state) return null; // cannot reassemble yet because we don't have all fragments @@ -94,6 +99,9 @@ public class PacketReassembly { // Remove read packets packetIds.forEach(queue::remove); + // Count + counter.next(packetIds.size()); + return reassembledPacket; } catch (Exception ex) { throw new IOException("Problem reassembling " + reassemblyList.size() + " packet(s): [" + @@ -106,7 +114,7 @@ public class PacketReassembly { } public void reset() { - currentPacketId = 0; + counter.reset(); queue.clear(); } } \ No newline at end of file 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 9743234..447935e 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 @@ -68,9 +68,7 @@ public abstract class AbstractTeamspeakClientSocket private CommandProcessor commandProcessor; - private Consumer exceptionHandler = throwable -> { - throwable.printStackTrace(); - }; + private Consumer exceptionHandler = Throwable::printStackTrace; private Identity identity = null; private ClientConnectionState connectionState = null; @@ -929,7 +927,9 @@ public abstract class AbstractTeamspeakClientSocket private int tries = 1, maxTries; private boolean willResend; - public PacketResponse(NetworkPacket sentPacket, Map responsibleQueue, int maxTries) { + public PacketResponse(NetworkPacket sentPacket, + Map responsibleQueue, + int maxTries) { this.sentPacket = sentPacket; this.responsibleQueue = responsibleQueue; this.maxTries = maxTries; @@ -1248,6 +1248,12 @@ public abstract class AbstractTeamspeakClientSocket public interface LocalCounter { Pair next(); + default Pair next(int n) { + Pair last = current(); + for (int i = 0; i < n; i ++) last = next(); + return last; + } + int getPacketId(); boolean setPacketId(int packetId); @@ -1340,7 +1346,6 @@ public abstract class AbstractTeamspeakClientSocket @Override public void setGeneration(int i) { synchronized (sendLock) { - Ts3Debugging.debug("setGeneration: " + i); this.generationId = 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 c89f906..4017fa0 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 @@ -60,7 +60,6 @@ public class LocalTeamspeakClientSocket private List listeners = new LinkedList<>(); private int acceptingReturnCode = Integer.MAX_VALUE; - private int lastReturnCode = 0; private Integer serverId = null; @@ -251,7 +250,7 @@ public class LocalTeamspeakClientSocket serverId = null; awaitingCommands.clear(); - acceptingReturnCode = lastReturnCode = 0; + acceptingReturnCode = 0; setOption("client.hostname", remote.getHostString()); @@ -341,10 +340,10 @@ public class LocalTeamspeakClientSocket if (getState() != ClientConnectionState.CONNECTED) throw new IOException("not connected"); - int returnCode; - - // sync may not be necessary here - returnCode = lastReturnCode++; + int maxReturnCode = calculateAcceptingReturnCode(); + if (maxReturnCode == Integer.MAX_VALUE) maxReturnCode = 0; + else maxReturnCode = maxReturnCode + 1; + int returnCode = maxReturnCode; command.add(new CommandSingleParameter("return_code", Integer.toString(returnCode))); diff --git a/src/test/java/com/github/manevolent/ts3j/FragmentTest.java b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java index 6c1c598..d343a10 100644 --- a/src/test/java/com/github/manevolent/ts3j/FragmentTest.java +++ b/src/test/java/com/github/manevolent/ts3j/FragmentTest.java @@ -25,40 +25,47 @@ public class FragmentTest extends TestCase { public void testParser() throws Exception { Ts3Debugging.setEnabled(true); - byte[] data = new byte[5003]; - new Random(0x1234567).nextBytes(data); - - ClientPacketHeader header = new ClientPacketHeader(); - Packet largePacket = new Packet(ProtocolRole.CLIENT, header); - header.setType(PacketBodyType.COMMAND); - - String text = Base64.toBase64String(data); - largePacket.setBody(new PacketBody2Command(ProtocolRole.CLIENT, text)); - - PacketReassembly reassembly = new PacketReassembly(); AbstractTeamspeakClientSocket.LocalCounterFull counter = new AbstractTeamspeakClientSocket.LocalCounterFull(65536, true); - counter.setPacketId(65534); // Test wrapping around the generation at the same time :) - reassembly.setPacketId(65535); + PacketReassembly reassembly = new PacketReassembly(); - List pieces = Fragments.split(largePacket); - Ts3Debugging.debug("Created " + pieces.size() + " pieces (total=" + data.length + ")."); + for (int i = 0; i < 10_000; i ++) { + byte[] data = new byte[5003]; + new Random(0x1234567).nextBytes(data); - for (Packet piece : pieces) { - Pair packetId = counter.next(); - piece.getHeader().setPacketId(packetId.getKey()); - piece.getHeader().setGeneration(packetId.getValue()); + ClientPacketHeader header = new ClientPacketHeader(); + Packet largePacket = new Packet(ProtocolRole.CLIENT, header); + header.setType(PacketBodyType.COMMAND); - reassembly.put(piece); - } + String text = Base64.toBase64String(data); + largePacket.setBody(new PacketBody2Command(ProtocolRole.CLIENT, text)); - Packet reassembled; - while (((reassembled = reassembly.next()) != null)) { - assertEquals( - ((PacketBody2Command) largePacket.getBody()).getText(), - ((PacketBody2Command) reassembled.getBody()).getText() - ); + List pieces = Fragments.split(largePacket); + + for (Packet piece : pieces) { + Pair packetId = counter.current(); + + piece.getHeader().setPacketId(packetId.getKey()); + piece.getHeader().setGeneration(packetId.getValue()); + + reassembly.put(piece); + + counter.next(); + } + + Packet reassembled; + boolean read = false; + while (((reassembled = reassembly.next()) != null)) { + assertEquals( + ((PacketBody2Command) largePacket.getBody()).getText(), + ((PacketBody2Command) reassembled.getBody()).getText() + ); + + read = true; + } + + assertEquals(true, read); } } diff --git a/src/test/java/com/github/manevolent/ts3j/FragmentTest2.java b/src/test/java/com/github/manevolent/ts3j/FragmentTest2.java deleted file mode 100644 index 930066e..0000000 --- a/src/test/java/com/github/manevolent/ts3j/FragmentTest2.java +++ /dev/null @@ -1,64 +0,0 @@ -package com.github.manevolent.ts3j; - -import com.github.manevolent.ts3j.protocol.Packet; -import com.github.manevolent.ts3j.protocol.ProtocolRole; -import com.github.manevolent.ts3j.protocol.header.HeaderFlag; -import com.github.manevolent.ts3j.protocol.packet.fragment.PacketBodyFragment; -import com.github.manevolent.ts3j.protocol.packet.fragment.PacketReassembly; -import com.github.manevolent.ts3j.util.Ts3Debugging; -import junit.framework.TestCase; -import org.bouncycastle.util.encoders.Base64; - -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.List; - -public class FragmentTest2 extends TestCase { - - public static void main(String[] args) throws Exception { - new FragmentTest2().testParser(); - } - - public void testParser() throws Exception { - Ts3Debugging.setEnabled(true); - - String[] packets = new String[] { - "LvdqqRB1ozMABBJlaXY0e5iIcGFja2V0c4FQkRkjgz0zMnE7IFGYbmVjIviVXFI5Ag8P3F8kgzQ4Mzg5OSFgkIgSRjSFKjI0MjgxYJiIVOJNNDE2NTA0OXFgmYh7gl+iZYsxz/mBUFGYZIUqMTkzUlOQiB5taW4hIocqMTgBYJCIFEc0oGUSMjkRYJCIIrv4MjmhYJiIUZiCNQMHAIAxGST6MDI5MTI2QWCZiKPzX2lwPTc0LjIwOS4yLjg1", - "xX0QvWydOFsAAlJHhAQAAEoHAAAAAACgY2lkPTEgY2xpZW50X2lkbGVfdGltZT0xNjgyMjdmXnVgAACAbmlxdWWR8TGSaWZpZXI9SW9RY3NZMnYwcXpNMThOeGFsAAQCgFBQazBiTEpLQT1mXm5pY2tuYbFrbm90XHNhXHN3aG9yZQEAAIBmXnZlcnNpb249My4xLjEwXHNbQnVpbGQ6XHMxNTI4NUAAgIAzNzYxNV1mXnBsYXRmb3JtPVdpbmRvd3NmXmlucHV0X22IhgOWdXRleedvdXR4IzBmXoQib25seZWgJ2D0lmhhcmR3YXJh52xeX+BkEmRlAAACiGZhdWx0X2NoYW5uZWw9XC84Zl5tZXRhX2RhdGFmXmlzXwAMAIByZWNvcmRpbmc9J2AFJV9zaWduPStcL0JXdnVlaWtHZzQAAACAWWtPMXYydXVaQjV2dEpKZ1VaNWJMOGNSZnhBc3RmbgAAAIBDVmRybzJqYSs0YSs4ckdVekR4OFwvdnZUWk9VVkQ2AECAhFU5NWhuV2I2MzhNQ1E992BzZWN1cml0eZHkc2hmXmxvZ2kcGUiMbl+Ce2ZeImJiYXMCnz0yZl5V519ncm91cJHxPTVmXnNlcgElA0lzPTYhIlCgZl5jcmVhIzM1MzAhYTc1OGZebGE=", - "+hN1ybZO1gEAAwJzdGNvgYNjJTM0MDg5MDMnYHQwwoCQb3RhbFWY8ZhzPTV3YGF3YXnoI2IXX21lc3NhZ3dldHlwZegjZmwgAACAYWdfYXZhJ3I9ZmQwMjhjYzUzNmU3YmM3MWJiOWQ4MGGQQZOhZWFjYxEUNTNmXiF7a19wb3chGzd3YCN7cmVhMnN06CMqe19tc2dmXmSAAELgZXNjcmlwdPGYPW1hbmVcc2lzs1RnaWFuEbRuaWdnZXJmXuFpBwDggiJ7IRsnYG1vbnRoX2J5dGVzX3VwbG9hZDF7IGAVIYFupZcxNzI3Mg4PwYE0Zl4jg55cMzc4NldgKoMggRnhaXByaW9D43NwZWFrIRs3YIb1X3Bob25liAYM+nRpY2ZebmVlITJfVCVhMnJ5X3ZpZXePaiYwdG9rZW5mXmlRmJLxJ2DhadEAHORW52NvbbF4ZCEbJ2Bjb3VudHJ5PVVTZl5W5wNJX2luaGVB42Vkll2S8QHCgIA3YGJhZGdlcz1vASV3b2xm6CNCZTY0SGFzaEOj81VJRD1jY2kAAACAZWJtbGJpbmtwbmNrbW1tbmhtZGhiZ2tmZG1wamRlZ+AAAKBtbGNlazFllIjxmF9maWxldHJhbnNmZXJfYmFuZHdpZCHYc48BR/ExkuIjkIghQTQ=", - }; - - List pieces = new ArrayList<>(); - - for (int i = 0; i < packets.length; i ++) { - Packet packet = new Packet(ProtocolRole.SERVER); - ByteBuffer buffer = ByteBuffer.wrap(Base64.decode(packets[i])); - packet.readHeader(buffer); - packet.setBody(new PacketBodyFragment(packet.getHeader().getType(), ProtocolRole.SERVER)); - packet.readBody(buffer); - pieces.add(packet); - - Ts3Debugging.debug( - "Read type=" + packet.getHeader().getType() - + " id=" + packet.getHeader().getPacketId() - + " flags=" + packet.getHeader().getPacketFlags() - + " fragment=" + packet.getHeader().getPacketFlag(HeaderFlag.FRAGMENTED) - ); - } - - Ts3Debugging.debug("Loaded " + pieces.size() + " pieces."); - - PacketReassembly reassembly = new PacketReassembly(); - reassembly.setPacketId(2); - - for (Packet piece : pieces) { - reassembly.put(piece); - } - - Packet reassembled; - while (((reassembled = reassembly.next()) != null)) { - Ts3Debugging.debug(reassembled.getHeader().getPacketId()); - } - } - -} diff --git a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java index a41b9ad..3de3c5f 100644 --- a/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java +++ b/src/test/java/com/github/manevolent/ts3j/ServerConnectionTest.java @@ -17,148 +17,149 @@ import static org.junit.Assert.assertEquals; public class ServerConnectionTest { public static void main(String[] args) throws Exception { - try { - Ts3Debugging.setEnabled(true); + while (true) { + try { + Ts3Debugging.setEnabled(true); - LocalIdentity identity = LocalIdentity.load( - new BigInteger(Base64.getDecoder().decode("Tj6YXM3qyRv8n25L2pH+OEJnRUl4auQf8+znjYrOmWU=")) - ); + LocalIdentity identity = LocalIdentity.load( + new BigInteger(Base64.getDecoder().decode("Tj6YXM3qyRv8n25L2pH+OEJnRUl4auQf8+znjYrOmWU=")) + ); - identity.improveSecurity(10); + identity.improveSecurity(10); - LocalTeamspeakClientSocket client = new LocalTeamspeakClientSocket(); + LocalTeamspeakClientSocket client = new LocalTeamspeakClientSocket(); - client.setIdentity(identity); + client.setIdentity(identity); - client.setNickname("Hello from Java"); + client.setNickname("Hello from Java"); - client.setOption("client.hwid", "JAVAJAVAJAVA"); + client.setOption("client.hwid", "JAVAJAVAJAVA"); - client.connect( - new InetSocketAddress( - "ts.teamlixo.net", - 9987 - ), - null, - 10000L - ); + client.connect( + new InetSocketAddress( + "ts.teamlixo.net", + 9987 + ), + null, + 10000L + ); - client.addListener(new TS3Listener() { - @Override - public void onTextMessage(TextMessageEvent e) { + client.addListener(new TS3Listener() { + @Override + public void onTextMessage(TextMessageEvent e) { - } + } - @Override - public void onClientJoin(ClientJoinEvent e) { + @Override + public void onClientJoin(ClientJoinEvent e) { - } + } - @Override - public void onClientLeave(ClientLeaveEvent e) { + @Override + public void onClientLeave(ClientLeaveEvent e) { - } + } - @Override - public void onServerEdit(ServerEditedEvent e) { + @Override + public void onServerEdit(ServerEditedEvent e) { - } + } - @Override - public void onChannelEdit(ChannelEditedEvent e) { + @Override + public void onChannelEdit(ChannelEditedEvent e) { - } + } - @Override - public void onChannelDescriptionChanged(ChannelDescriptionEditedEvent e) { + @Override + public void onChannelDescriptionChanged(ChannelDescriptionEditedEvent e) { - } + } - @Override - public void onClientMoved(ClientMovedEvent e) { + @Override + public void onClientMoved(ClientMovedEvent e) { - } + } - @Override - public void onChannelCreate(ChannelCreateEvent e) { + @Override + public void onChannelCreate(ChannelCreateEvent e) { - } + } - @Override - public void onChannelDeleted(ChannelDeletedEvent e) { + @Override + public void onChannelDeleted(ChannelDeletedEvent e) { - } + } - @Override - public void onChannelMoved(ChannelMovedEvent e) { + @Override + public void onChannelMoved(ChannelMovedEvent e) { - } + } - @Override - public void onChannelPasswordChanged(ChannelPasswordChangedEvent e) { + @Override + public void onChannelPasswordChanged(ChannelPasswordChangedEvent e) { - } + } - @Override - public void onChannelList(ChannelListEvent e) { + @Override + public void onChannelList(ChannelListEvent e) { - } + } - @Override - public void onPrivilegeKeyUsed(PrivilegeKeyUsedEvent e) { + @Override + public void onPrivilegeKeyUsed(PrivilegeKeyUsedEvent e) { - } + } - @Override - public void onChannelGroupList(ChannelGroupListEvent e) { + @Override + public void onChannelGroupList(ChannelGroupListEvent e) { - } + } - @Override - public void onServerGroupList(ServerGroupListEvent e) { + @Override + public void onServerGroupList(ServerGroupListEvent e) { - } + } - @Override - public void onClientNeededPermissions(ClientNeededPermissionsEvent e) { + @Override + public void onClientNeededPermissions(ClientNeededPermissionsEvent e) { - } + } - @Override - public void onClientChannelGroupChanged(ClientChannelGroupChangedEvent e) { + @Override + public void onClientChannelGroupChanged(ClientChannelGroupChangedEvent e) { - } + } - @Override - public void onClientChanged(ClientUpdatedEvent e) { + @Override + public void onClientChanged(ClientUpdatedEvent e) { - } + } - @Override - public void onDisconnected(DisconnectedEvent e) { + @Override + public void onDisconnected(DisconnectedEvent e) { - } + } - @Override - public void onChannelSubscribed(ChannelSubscribedEvent e) { + @Override + public void onChannelSubscribed(ChannelSubscribedEvent e) { - } - }); + } + }); - final long[] t = {System.currentTimeMillis()}; + assertEquals(client.getState(), ClientConnectionState.CONNECTED); - assertEquals(client.getState(), ClientConnectionState.CONNECTED); + client.subscribeAll(); - client.subscribeAll(); - - while (client.isConnected()) { - for (Client c : client.listClients()) - client.getClientInfo(c.getId()); + client.listClients(); Thread.sleep(1000L); + + client.listClients(); + + client.disconnect(); + } catch (Throwable ex) { + ex.printStackTrace(); } - } catch (Throwable ex) { - ex.printStackTrace(); } } }