Login Payload Manager. Modularizing the project...

build.gradle

@@ -4,7 +4,7 @@ plugins {
 }
 
 group 'org.adde0109'
-version '1.0.5-alpha'
+version '1.0.6-alpha'
 
 repositories {
     maven {

src/main/java/org/adde0109/ambassador/Ambassador.java

@@ -22,7 +22,7 @@ import org.slf4j.Logger;
 
 import java.nio.file.Path;
 
-@Plugin(id = "ambassador", name = "Ambassador", version = "1.0.5-alpha", authors = {"adde0109"})
+@Plugin(id = "ambassador", name = "Ambassador", version = "1.0.6-alpha", authors = {"adde0109"})
 public class Ambassador {
 
   public ProxyServer server;

src/main/java/org/adde0109/ambassador/forge/FML2CRPMClientConnectionPhase.java

@@ -8,88 +8,73 @@ import com.velocitypowered.proxy.VelocityServer;
 import com.velocitypowered.proxy.config.PlayerInfoForwarding;
 import com.velocitypowered.proxy.config.VelocityConfiguration;
 import com.velocitypowered.proxy.connection.MinecraftConnection;
+import com.velocitypowered.proxy.connection.backend.VelocityServerConnection;
 import com.velocitypowered.proxy.connection.client.ConnectedPlayer;
 import com.velocitypowered.proxy.protocol.StateRegistry;
 import com.velocitypowered.proxy.protocol.packet.LoginPluginMessage;
 import com.velocitypowered.proxy.protocol.packet.LoginPluginResponse;
 import com.velocitypowered.proxy.protocol.packet.PluginMessage;
 import com.velocitypowered.proxy.protocol.packet.ServerLoginSuccess;
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import org.adde0109.ambassador.velocity.VelocityForgeClientConnectionPhase;
 import org.adde0109.ambassador.velocity.VelocityForgeHandshakeSessionHandler;
+import org.adde0109.ambassador.velocity.VelocityLoginPayloadManager;
 
 import java.util.ArrayList;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 
-public class ForgeFML2ClientConnectionPhase implements VelocityForgeClientConnectionPhase {
+public class FML2CRPMClientConnectionPhase implements VelocityForgeClientConnectionPhase {
   private boolean isResettable;
 
   //TODO: Use modData inside ConnectedPlayer instead
   public byte[] modListData;
   private RegisteredServer backupServer;
 
-  private final ArrayList<Integer> listenerList = new ArrayList();
+  private VelocityLoginPayloadManager payloadManager;
-  private Runnable whenComplete;
   public ClientPhase clientPhase = ClientPhase.HANDSHAKE;
   @Override
-  public void handleLogin(ConnectedPlayer player,ForgeHandshakeUtils.CachedServerHandshake handshake, Continuation continuation) {
+  public void handleLogin(ConnectedPlayer player, VelocityServer server, Continuation continuation) {
     final MinecraftConnection connection = player.getConnection();
+    payloadManager = new VelocityLoginPayloadManager(connection);
     VelocityForgeHandshakeSessionHandler sessionHandler = new VelocityForgeHandshakeSessionHandler(connection.getSessionHandler(), player);
-    //Without initial modlist for now
+    payloadManager.sendPayload("fml:loginwrapper",Unpooled.wrappedBuffer(ForgeHandshakeUtils.emptyModlist)).thenAccept((data) -> {
-    if (true) {
+      if (modListData == null)
-      connection.delayedWrite(new LoginPluginMessage(0, "fml:loginwrapper", Unpooled.wrappedBuffer(ForgeHandshakeUtils.emptyModlist)));
+        modListData = ByteBufUtil.getBytes(data);
-      listenerList.add(0);
-    } else {
-      connection.delayedWrite(new LoginPluginMessage(0, "fml:loginwrapper", Unpooled.wrappedBuffer(handshake.modListPacket)));
-      listenerList.add(0);
-      for (int i = 0; i < handshake.otherPackets.size(); i++) {
-        connection.delayedWrite(new LoginPluginMessage(i + 1, "fml:loginwrapper", Unpooled.wrappedBuffer(handshake.otherPackets.get(i))));
-        listenerList.add(i + 1);
-      }
-    }
-    this.whenComplete = () -> {
       this.clientPhase = ClientPhase.MODDED;
       continuation.resume();
-    };
+    });
     connection.setSessionHandler(sessionHandler);
     connection.flush();
   }
 
   @Override
   public boolean handle(ConnectedPlayer player, LoginPluginResponse packet) {
-    if (packet.getId() == 98) {
-      this.clientPhase = ClientPhase.HANDSHAKE;
-    } else if (packet.getId() == 0) {
-      this.clientPhase = ClientPhase.MODLIST;
-      if (modListData == null) {
-        modListData = ByteBufUtil.getBytes(packet.content());
-      }
-    }
-    if (!listenerList.removeIf(id -> id.equals(packet.getId()))) {
-      player.getConnectionInFlight().getConnection().write(packet.retain());
-    } else if (listenerList.isEmpty() && whenComplete != null) {
-      whenComplete.run();
-      whenComplete = null;
-    }
     return true;
   }
   public void reset(ConnectedPlayer player, Runnable whenComplete) {
-    this.whenComplete = whenComplete;
     if (player.getConnectedServer() != null) {
       backupServer = player.getConnectedServer().getServer();
       player.getConnectedServer().disconnect();
       player.setConnectedServer(null);
     }
+
     MinecraftConnection connection = player.getConnection();
     connection.setSessionHandler(new VelocityForgeHandshakeSessionHandler(connection.getSessionHandler(),player));
+
     if (connection.getState() == StateRegistry.LOGIN) {
-      connection.write(new LoginPluginMessage(98,"fml:loginwrapper", Unpooled.wrappedBuffer(ForgeHandshakeUtils.generateResetPacket())));
+      payloadManager.sendPayload("fml:loginwrapper", Unpooled.wrappedBuffer(ForgeHandshakeUtils.generateResetPacket()));
     } else {
       connection.write(new PluginMessage("fml:handshake",Unpooled.wrappedBuffer(ForgeHandshakeUtils.generatePluginResetPacket())));
       connection.setState(StateRegistry.LOGIN);
     }
-    listenerList.add(98);
+    payloadManager.listenFor(98).thenAccept((ignored) -> {
+      this.clientPhase = ClientPhase.HANDSHAKE;
+      whenComplete.run();
+    });
+
     this.clientPhase = null;
   }
   public void complete(VelocityServer server, ConnectedPlayer player, MinecraftConnection connection) {
@@ -117,6 +102,18 @@ public class ForgeFML2ClientConnectionPhase implements VelocityForgeClientConnec
     }
   }
 
+  public void forwardPayload(VelocityServerConnection serverConnection, LoginPluginMessage payload) {
+    payloadManager.sendPayload("fml:loginwrapper",payload.content()).thenAccept((responseData) -> {
+      //Move this to the backend. Backend should have its own forwarder.
+      serverConnection.getConnection().write(new LoginPluginResponse(payload.getId(),responseData.isReadable(),responseData.retain()));
+    });
+  }
+
+  @Override
+  public VelocityLoginPayloadManager getPayloadManager() {
+    return payloadManager;
+  }
+
   public enum ClientPhase {
     VANILLA,
     HANDSHAKE,

src/main/java/org/adde0109/ambassador/forge/FML2ClientConnectionPhase.java

@@ -0,0 +1,42 @@
+package org.adde0109.ambassador.forge;
+
+import com.velocitypowered.api.event.Continuation;
+import com.velocitypowered.proxy.VelocityServer;
+import com.velocitypowered.proxy.connection.MinecraftConnection;
+import com.velocitypowered.proxy.connection.MinecraftSessionHandler;
+import com.velocitypowered.proxy.connection.client.ConnectedPlayer;
+import com.velocitypowered.proxy.connection.client.LoginSessionHandler;
+import com.velocitypowered.proxy.protocol.packet.LoginPluginResponse;
+import org.adde0109.ambassador.velocity.VelocityForgeClientConnectionPhase;
+import org.adde0109.ambassador.velocity.VelocityLoginPayloadManager;
+
+import java.lang.reflect.Method;
+
+public class FML2ClientConnectionPhase implements VelocityForgeClientConnectionPhase {
+
+
+  @Override
+  public void handleLogin(ConnectedPlayer player, VelocityServer server, Continuation continuation) {
+    MinecraftSessionHandler sessionHandler = player.getConnection().getSessionHandler();
+    if (player.getConnection().getSessionHandler() == null) {
+      continuation.resumeWithException(new Exception("No current player session handler"));
+      return;
+    }
+    if (!(player.getConnection().getSessionHandler() instanceof LoginSessionHandler)) {
+      continuation.resumeWithException(new Exception("Invalid current player session handler:" + player.getConnection().getSessionHandler().getClass().getName()));
+      return;
+    }
+    try {
+      Method connectToInitalServer = LoginSessionHandler.class.getDeclaredMethod("connectToInitialServer");
+      connectToInitalServer.setAccessible(true);
+      connectToInitalServer.invoke(sessionHandler);
+    } catch (ReflectiveOperationException e) {
+      continuation.resumeWithException(e);
+    }
+  }
+
+  @Override
+  public boolean handle(ConnectedPlayer player, LoginPluginResponse packet) {
+    return VelocityForgeClientConnectionPhase.super.handle(player, packet);
+  }
+}

src/main/java/org/adde0109/ambassador/forge/ForgeFML2ConnectionType.java

@@ -3,9 +3,7 @@ package org.adde0109.ambassador.forge;
 import com.velocitypowered.api.util.GameProfile;
 import com.velocitypowered.proxy.config.PlayerInfoForwarding;
 import com.velocitypowered.proxy.connection.ConnectionType;
-import com.velocitypowered.proxy.connection.MinecraftConnection;
 import com.velocitypowered.proxy.connection.backend.BackendConnectionPhase;
-import com.velocitypowered.proxy.connection.backend.BackendConnectionPhases;
 import com.velocitypowered.proxy.connection.client.ClientConnectionPhase;
 import org.adde0109.ambassador.velocity.backend.VelocityForgeBackendConnectionPhase;
 
@@ -15,7 +13,7 @@ public class ForgeFML2ConnectionType implements ConnectionType {
 
   @Override
   public ClientConnectionPhase getInitialClientPhase() {
-    return new ForgeFML2ClientConnectionPhase();
+    return new FML2CRPMClientConnectionPhase();
   }
 
   @Override
@@ -25,8 +23,9 @@ public class ForgeFML2ConnectionType implements ConnectionType {
 
   @Override
   public GameProfile addGameProfileTokensIfRequired(GameProfile original, PlayerInfoForwarding forwardingType) {
-    if (forwardingType == PlayerInfoForwarding.LEGACY)
+    if (forwardingType == PlayerInfoForwarding.LEGACY) {
       original.addProperties(Collections.singleton(new GameProfile.Property("extraData", "\1FML2\1", "")));
+    }
     return original;
   }
 }

src/main/java/org/adde0109/ambassador/velocity/VelocityEventHandler.java

@@ -3,13 +3,14 @@ package org.adde0109.ambassador.velocity;
 import com.velocitypowered.api.event.Continuation;
 import com.velocitypowered.api.event.PostOrder;
 import com.velocitypowered.api.event.Subscribe;
+import com.velocitypowered.api.event.connection.LoginEvent;
+import com.velocitypowered.api.event.connection.PostLoginEvent;
 import com.velocitypowered.api.event.permission.PermissionsSetupEvent;
 import com.velocitypowered.api.event.player.KickedFromServerEvent;
+import com.velocitypowered.proxy.VelocityServer;
 import com.velocitypowered.proxy.connection.client.ConnectedPlayer;
 import org.adde0109.ambassador.Ambassador;
-import org.adde0109.ambassador.forge.ForgeFML2ClientConnectionPhase;
+import org.adde0109.ambassador.forge.FML2CRPMClientConnectionPhase;
-
-import java.util.Objects;
 
 public class VelocityEventHandler {
 
@@ -29,12 +30,12 @@ public class VelocityEventHandler {
       continuation.resume();
       return;
     }
-    player.getConnection().eventLoop().submit(() -> phase.handleLogin(player,null,continuation));
+    player.getConnection().eventLoop().submit(() -> phase.handleLogin(player, (VelocityServer) ambassador.server,continuation));
   }
 
   @Subscribe(order = PostOrder.LAST)
   public void onKickedFromServerEvent(KickedFromServerEvent event, Continuation continuation) {
-    if (((ConnectedPlayer) event.getPlayer()).getPhase() instanceof ForgeFML2ClientConnectionPhase phase) {
+    if (((ConnectedPlayer) event.getPlayer()).getPhase() instanceof FML2CRPMClientConnectionPhase phase) {
       phase.handleKick(event);
     }
     continuation.resume();

src/main/java/org/adde0109/ambassador/velocity/VelocityForgeClientConnectionPhase.java

@@ -1,6 +1,7 @@
 package org.adde0109.ambassador.velocity;
 
 import com.velocitypowered.api.event.Continuation;
+import com.velocitypowered.proxy.VelocityServer;
 import com.velocitypowered.proxy.connection.MinecraftConnection;
 import com.velocitypowered.proxy.connection.client.ClientConnectionPhase;
 import com.velocitypowered.proxy.connection.client.ConnectedPlayer;
@@ -8,13 +9,20 @@ import com.velocitypowered.proxy.protocol.packet.LoginPluginResponse;
 import org.adde0109.ambassador.forge.ForgeHandshakeUtils;
 
 import javax.annotation.Nullable;
+import java.util.ArrayList;
 
 public interface VelocityForgeClientConnectionPhase extends ClientConnectionPhase {
   //TODO:Make class when PCF is done
 
-  default void handleLogin(ConnectedPlayer player, @Nullable ForgeHandshakeUtils.CachedServerHandshake initialHandshake, Continuation continuation) {
+
+  default void handleLogin(ConnectedPlayer player, VelocityServer server, Continuation continuation) {
+
   }
   default boolean handle(ConnectedPlayer player,LoginPluginResponse packet) {
     return false;
   }
+
+  default VelocityLoginPayloadManager getPayloadManager() {
+    return null;
+  }
 }

src/main/java/org/adde0109/ambassador/velocity/VelocityForgeHandshakeSessionHandler.java

@@ -1,16 +1,9 @@
 package org.adde0109.ambassador.velocity;
 
-import com.velocitypowered.proxy.connection.MinecraftConnection;
 import com.velocitypowered.proxy.connection.MinecraftSessionHandler;
 import com.velocitypowered.proxy.connection.client.ConnectedPlayer;
-import com.velocitypowered.proxy.connection.client.LoginSessionHandler;
-import com.velocitypowered.proxy.protocol.MinecraftPacket;
 import com.velocitypowered.proxy.protocol.packet.LoginPluginResponse;
 import io.netty.buffer.ByteBuf;
-import org.adde0109.ambassador.forge.ForgeFML2ClientConnectionPhase;
-
-import java.util.ArrayList;
-import java.util.List;
 
 public class VelocityForgeHandshakeSessionHandler implements MinecraftSessionHandler {
   private final MinecraftSessionHandler original;
@@ -23,7 +16,7 @@ public class VelocityForgeHandshakeSessionHandler implements MinecraftSessionHan
 
   @Override
   public boolean handle(LoginPluginResponse packet) {
-    if (((VelocityForgeClientConnectionPhase) player.getPhase()).handle(player, packet)) {
+    if (((VelocityForgeClientConnectionPhase) player.getPhase()).getPayloadManager().handlePayload(packet)) {
       return true;
     } else {
       return original.handle(packet);

src/main/java/org/adde0109/ambassador/velocity/VelocityLoginPayloadManager.java

@@ -0,0 +1,65 @@
+package org.adde0109.ambassador.velocity;
+
+import com.velocitypowered.proxy.connection.MinecraftConnection;
+import com.velocitypowered.proxy.protocol.packet.LoginPluginMessage;
+import com.velocitypowered.proxy.protocol.packet.LoginPluginResponse;
+import io.netty.buffer.ByteBuf;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+
+public class VelocityLoginPayloadManager {
+  private final HashMap<Integer, CompletableFuture<ByteBuf>> listenerList = new HashMap<>();
+  private int counter = 0;
+  private final MinecraftConnection connection;
+
+  public VelocityLoginPayloadManager(MinecraftConnection connection) {
+    this.connection = connection;
+  }
+
+  public CompletableFuture<ByteBuf> sendPayload(String channel, ByteBuf data) {
+    connection.write(new LoginPluginMessage(counter,channel,data));
+    final CompletableFuture<ByteBuf> callback = new CompletableFuture<>();
+    listenerList.put(counter, callback);
+    counter++;
+    return callback;
+  }
+
+  public CompletableFuture<ByteBuf> sendPayloads(String channel, List<ByteBuf> dataList) {
+    final CompletableFuture<ByteBuf> callback = new CompletableFuture<>();
+    for (ByteBuf data : dataList) {
+      connection.delayedWrite(new LoginPluginMessage(counter, channel, data));
+      listenerList.put(counter,callback);
+      counter++;
+    }
+    connection.flush();
+    return callback;
+  }
+
+  public CompletableFuture<ByteBuf> listenFor(int id) throws RuntimeException{
+    if (!listenerList.containsValue(id)) {
+      CompletableFuture<ByteBuf> callback = new CompletableFuture<>();
+      listenerList.put(id,callback);
+      return callback;
+    } else {
+      throw new RuntimeException("Already listening for:" + id);
+    }
+  }
+
+   boolean handlePayload(LoginPluginResponse response) {
+    final CompletableFuture<ByteBuf> callback = listenerList.get(response.getId());
+      if (callback != null) {
+        listenerList.remove(response.getId());
+        if (!listenerList.containsValue(callback)) {
+          callback.complete(response.content());
+        }
+        return true;
+      } else {
+        return false;
+      }
+    }
+  }

src/main/java/org/adde0109/ambassador/velocity/backend/VelocityForgeBackendConnectionPhase.java

@@ -4,9 +4,8 @@ import com.velocitypowered.proxy.VelocityServer;
 import com.velocitypowered.proxy.connection.backend.BackendConnectionPhase;
 import com.velocitypowered.proxy.connection.backend.VelocityServerConnection;
 import com.velocitypowered.proxy.connection.client.ConnectedPlayer;
-import com.velocitypowered.proxy.protocol.StateRegistry;
 import com.velocitypowered.proxy.protocol.packet.LoginPluginMessage;
-import org.adde0109.ambassador.forge.ForgeFML2ClientConnectionPhase;
+import org.adde0109.ambassador.forge.FML2CRPMClientConnectionPhase;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -19,19 +18,19 @@ public class VelocityForgeBackendConnectionPhase implements BackendConnectionPha
   }
 
   public void handleSuccess(VelocityServerConnection serverCon, VelocityServer server) {
-    ForgeFML2ClientConnectionPhase clientPhase = ((ForgeFML2ClientConnectionPhase) serverCon.getPlayer().getPhase());
+    FML2CRPMClientConnectionPhase clientPhase = ((FML2CRPMClientConnectionPhase) serverCon.getPlayer().getPhase());
-    if (clientPhase.clientPhase == ForgeFML2ClientConnectionPhase.ClientPhase.HANDSHAKE || clientPhase.clientPhase == ForgeFML2ClientConnectionPhase.ClientPhase.MODLIST) {
+    if (clientPhase.clientPhase == FML2CRPMClientConnectionPhase.ClientPhase.HANDSHAKE || clientPhase.clientPhase == FML2CRPMClientConnectionPhase.ClientPhase.MODLIST) {
       clientPhase.complete((VelocityServer) server,serverCon.getPlayer(),serverCon.getPlayer().getConnection());
     }
   }
 
   public boolean handle(VelocityServerConnection server, ConnectedPlayer player, LoginPluginMessage message) throws Exception {
-    ForgeFML2ClientConnectionPhase clientPhase = ((ForgeFML2ClientConnectionPhase) player.getPhase());
+    FML2CRPMClientConnectionPhase clientPhase = ((FML2CRPMClientConnectionPhase) player.getPhase());
     message.retain();
-    if (clientPhase.clientPhase == ForgeFML2ClientConnectionPhase.ClientPhase.VANILLA) {
+    if (clientPhase.clientPhase == FML2CRPMClientConnectionPhase.ClientPhase.VANILLA) {
       clientPhase.reset(player, () -> {
         for (LoginPluginMessage msg: queuedHandshakePackets) {
-          player.getConnection().delayedWrite(msg);
+          clientPhase.forwardPayload(server,msg);
         }
         player.getConnection().flush();
         queuedHandshakePackets = null;
@@ -39,7 +38,7 @@ public class VelocityForgeBackendConnectionPhase implements BackendConnectionPha
       queuedHandshakePackets = new ArrayList<>();
       queuedHandshakePackets.add(message);
     } else if (clientPhase.clientPhase != null) {
-      player.getConnection().write(message);
+      clientPhase.forwardPayload(server,message);
     } else {
       queuedHandshakePackets.add(message);
     }

src/main/java/org/adde0109/ambassador/velocity/backend/VelocityForgeBackendHandshakeHandler.java

@@ -10,9 +10,7 @@ import com.velocitypowered.proxy.protocol.packet.ServerLoginSuccess;
 import io.netty.channel.*;
 import io.netty.util.ReferenceCountUtil;
 import org.adde0109.ambassador.forge.ForgeConstants;
-import org.adde0109.ambassador.forge.ForgeFML2ClientConnectionPhase;
+import org.adde0109.ambassador.forge.FML2CRPMClientConnectionPhase;
-
-import java.util.concurrent.CountDownLatch;
 
 public class VelocityForgeBackendHandshakeHandler extends ChannelDuplexHandler {
 
@@ -24,24 +22,26 @@ public class VelocityForgeBackendHandshakeHandler extends ChannelDuplexHandler {
   }
 
   @Override
-  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+  public void flush(ChannelHandlerContext ctx) throws Exception {
-    if ((msg instanceof ServerLogin)) {
+    if (serverConnection != null){
+      ctx.flush();
+      return;
+    }
       ChannelHandler handler = ctx.pipeline().get(Connections.HANDLER);
       if (handler instanceof MinecraftConnection connection) {
         if (connection.getAssociation() instanceof VelocityServerConnection serverConnection) {
-          if (serverConnection.getPlayer().getPhase() instanceof ForgeFML2ClientConnectionPhase phase) {
+          if (serverConnection.getPlayer().getPhase() instanceof FML2CRPMClientConnectionPhase phase) {
             init(connection,serverConnection);
-            if (phase.clientPhase == ForgeFML2ClientConnectionPhase.ClientPhase.MODDED) {
+            if (phase.clientPhase == FML2CRPMClientConnectionPhase.ClientPhase.MODDED) {
               phase.reset(serverConnection.getPlayer(), () -> {
-                ctx.write(msg, promise);
                 ctx.flush();
               });
             } else {
-              ctx.write(msg,promise);
+              ctx.flush();
             }
           } else {
             ctx.pipeline().remove(this);
-            ctx.write(msg,promise);
+            ctx.flush();
           }
         } else {
           throw new Exception("Connection not associated with a server connection");
@@ -49,9 +49,6 @@ public class VelocityForgeBackendHandshakeHandler extends ChannelDuplexHandler {
       } else {
         throw new Exception("Default minecraft packet handler not found");
       }
-    } else {
-      ctx.write(msg,promise);
-    }
   }
 
   @Override
@@ -61,6 +58,7 @@ public class VelocityForgeBackendHandshakeHandler extends ChannelDuplexHandler {
       ReferenceCountUtil.release(msg);
     } else if (msg instanceof ServerLoginSuccess) {
       ((VelocityForgeBackendConnectionPhase) serverConnection.getPhase()).handleSuccess(serverConnection,server);
+      ctx.pipeline().remove(this);
       ctx.fireChannelRead(msg);
     } else {
       ctx.fireChannelRead(msg);