Merge pull request #1005 from creeper123123321/sendpackettoserver

PacketWrapper#sendToServer changes

bungee/src/main/java/us/myles/ViaVersion/bungee/providers/BungeeMovementTransmitter.java

@@ -5,6 +5,7 @@ import us.myles.ViaVersion.api.data.UserConnection;
 import us.myles.ViaVersion.api.type.Type;
 import us.myles.ViaVersion.packets.State;
 import us.myles.ViaVersion.protocols.base.ProtocolInfo;
+import us.myles.ViaVersion.protocols.protocol1_9to1_8.Protocol1_9TO1_8;
 import us.myles.ViaVersion.protocols.protocol1_9to1_8.providers.MovementTransmitterProvider;
 import us.myles.ViaVersion.protocols.protocol1_9to1_8.storage.MovementTracker;
 
@@ -24,7 +25,7 @@ public class BungeeMovementTransmitter extends MovementTransmitterProvider {
             PacketWrapper wrapper = new PacketWrapper(0x03, null, userConnection);
             wrapper.write(Type.BOOLEAN, userConnection.get(MovementTracker.class).isGround());
             try {
-                wrapper.sendToServer();
+                wrapper.sendToServer(Protocol1_9TO1_8.class);
             } catch (Exception e) {
                 e.printStackTrace();
             }

common/src/main/java/us/myles/ViaVersion/api/PacketWrapper.java

@@ -298,7 +298,7 @@ public class PacketWrapper {
      */
     public void send(Class<? extends Protocol> packetProtocol, boolean skipCurrentPipeline, boolean currentThread) throws Exception {
         if (!isCancelled()) {
-            ByteBuf output = constructPacket(packetProtocol, skipCurrentPipeline);
+            ByteBuf output = constructPacket(packetProtocol, skipCurrentPipeline, Direction.OUTGOING);
             user().sendRawPacket(output, currentThread);
         }
     }
@@ -311,11 +311,13 @@ public class PacketWrapper {
      * @return Packet buffer
      * @throws Exception if it fails to write
      */
-    private ByteBuf constructPacket(Class<? extends Protocol> packetProtocol, boolean skipCurrentPipeline) throws Exception {
+    private ByteBuf constructPacket(Class<? extends Protocol> packetProtocol, boolean skipCurrentPipeline, Direction direction) throws Exception {
         // Apply current pipeline
         List<Protocol> protocols = new ArrayList<>(user().get(ProtocolInfo.class).getPipeline().pipes());
-        // Other way if outgoing
-        Collections.reverse(protocols);
+        if (direction == Direction.OUTGOING) {
+            // Other way if outgoing
+            Collections.reverse(protocols);
+        }
         int index = 0;
         for (int i = 0; i < protocols.size(); i++) {
             if (protocols.get(i).getClass().equals(packetProtocol)) {
@@ -360,7 +362,7 @@ public class PacketWrapper {
      */
     public ChannelFuture sendFuture(Class<? extends Protocol> packetProtocol) throws Exception {
         if (!isCancelled()) {
-            ByteBuf output = constructPacket(packetProtocol, true);
+            ByteBuf output = constructPacket(packetProtocol, true, Direction.OUTGOING);
             return user().sendRawPacketFuture(output);
         }
         return user().getChannel().newFailedFuture(new Exception("Cancelled packet"));
@@ -471,15 +473,40 @@ public class PacketWrapper {
      *
      * @throws Exception If it failed to write
      */
+    @Deprecated
     public void sendToServer() throws Exception {
         if (!isCancelled()) {
             ByteBuf output = inputBuffer == null ? user().getChannel().alloc().buffer() : inputBuffer.alloc().buffer();
             writeToBuffer(output);
 
-            user().getChannel().pipeline().context(Via.getManager().getInjector().getDecoderName()).fireChannelRead(output);
+            user().sendRawPacketToServer(output, true);
         }
     }
 
+    /**
+     * Send this packet to the server.
+     *
+     * @param packetProtocol - The protocol version of the packet.
+     * @param skipCurrentPipeline - Skip the current pipeline
+     * @param currentThread - Run in the same thread
+     * @throws Exception if it fails to write
+     */
+    public void sendToServer(Class<? extends Protocol> packetProtocol, boolean skipCurrentPipeline, boolean currentThread) throws Exception {
+        if (!isCancelled()) {
+            ByteBuf output = constructPacket(packetProtocol, skipCurrentPipeline, Direction.INCOMING);
+            user().sendRawPacketToServer(output, currentThread);
+        }
+    }
+
+    public void sendToServer(Class<? extends Protocol> packetProtocol, boolean skipCurrentPipeline) throws Exception {
+        sendToServer(packetProtocol, skipCurrentPipeline, false);
+    }
+
+    public void sendToServer(Class<? extends Protocol> packetProtocol) throws Exception {
+        sendToServer(packetProtocol, true);
+    }
+
+
     @Override
     public String toString() {
         return "PacketWrapper{" +

common/src/main/java/us/myles/ViaVersion/api/data/UserConnection.java

@@ -3,12 +3,17 @@ package us.myles.ViaVersion.api.data;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.socket.SocketChannel;
 import lombok.Data;
+import lombok.NonNull;
 import net.md_5.bungee.api.ChatColor;
+import us.myles.ViaVersion.api.PacketWrapper;
 import us.myles.ViaVersion.api.Via;
 import us.myles.ViaVersion.api.ViaVersionConfig;
+import us.myles.ViaVersion.api.type.Type;
 import us.myles.ViaVersion.protocols.base.ProtocolInfo;
+import us.myles.ViaVersion.util.PipelineUtil;
 
 import java.util.Map;
 import java.util.UUID;
@@ -16,6 +21,13 @@ import java.util.concurrent.ConcurrentHashMap;
 
 @Data
 public class UserConnection {
+    /**
+     * The channel of the user.
+     * /!\ In some unofficial platforms this is a client channel
+     *
+     * TODO - Weak this field to {@link io.netty.channel.Channel}?
+     */
+    @NonNull
     private final SocketChannel channel;
     Map<Class, StoredObject> storedObjects = new ConcurrentHashMap<>();
     private boolean active = true;
@@ -128,7 +140,7 @@ public class UserConnection {
      */
     public boolean incrementReceived() {
         // handle stats
-        Long diff = System.currentTimeMillis() - startTime;
+        long diff = System.currentTimeMillis() - startTime;
         if (diff >= 1000) {
             packetsPerSecond = intervalPackets;
             startTime = System.currentTimeMillis();
@@ -147,7 +159,7 @@ public class UserConnection {
         // Max PPS Checker
         if (conf.getMaxPPS() > 0) {
             if (getPacketsPerSecond() >= conf.getMaxPPS()) {
-                disconnect(conf.getMaxPPSKickMessage().replace("%pps", ((Long) getPacketsPerSecond()).intValue() + ""));
+                disconnect(conf.getMaxPPSKickMessage().replace("%pps", Long.toString(getPacketsPerSecond())));
                 return true; // don't send current packet
             }
         }
@@ -165,7 +177,7 @@ public class UserConnection {
                 }
 
                 if (getWarnings() >= conf.getMaxWarnings()) {
-                    disconnect(conf.getMaxWarningsKickMessage().replace("%pps", ((Long) getPacketsPerSecond()).intValue() + ""));
+                    disconnect(conf.getMaxWarningsKickMessage().replace("%pps", Long.toString(getPacketsPerSecond())));
                     return true; // don't send current packet
                 }
             }
@@ -195,4 +207,48 @@ public class UserConnection {
         }
 
     }
+
+    /**
+     * Sends a raw packet to the server
+     *
+     * @param packet Raw packet to be sent
+     * @param currentThread If {@code true} executes immediately, {@code false} submits a task to EventLoop
+     */
+    public void sendRawPacketToServer(final ByteBuf packet, boolean currentThread) {
+        final ByteBuf buf = packet.alloc().buffer();
+        try {
+            Type.VAR_INT.write(buf, PacketWrapper.PASSTHROUGH_ID);
+        } catch (Exception e) {
+            // Should not happen
+            Via.getPlatform().getLogger().warning("Type.VAR_INT.write thrown an exception: " + e);
+        }
+        buf.writeBytes(packet);
+        packet.release();
+        final ChannelHandlerContext context = PipelineUtil.getPreviousContext(Via.getManager().getInjector().getDecoderName(), getChannel().pipeline());
+        if (currentThread) {
+            if (context != null) {
+                context.fireChannelRead(buf);
+            } else {
+                getChannel().pipeline().fireChannelRead(buf);
+            }
+        } else {
+            channel.eventLoop().submit(new Runnable() {
+                @Override
+                public void run() {
+                    if (context != null) {
+                        context.fireChannelRead(buf);
+                    } else {
+                        getChannel().pipeline().fireChannelRead(buf);
+                    }
+                }
+            });
+        }
+    }
+
+    /**
+     * Sends a raw packet to the server. It will submit a task to EventLoop
+     *
+     * @param packet Raw packet to be sent
+     */
+    public void sendRawPacketToServer(ByteBuf packet) { sendRawPacketToServer(packet, false); }
 }

common/src/main/java/us/myles/ViaVersion/util/PipelineUtil.java

@@ -100,4 +100,15 @@ public class PipelineUtil {
         }
         return null;
     }
+
+    public static ChannelHandlerContext getPreviousContext(String name, ChannelPipeline pipeline) {
+        String previous = null;
+        for (String entry : pipeline.toMap().keySet()) {
+            if (entry.equals(name)) {
+                return pipeline.context(previous);
+            }
+            previous = entry;
+        }
+        return null;
+    }
 }