Add support for sending plugin messages over the wire

api/src/main/java/com/velocitypowered/api/proxy/Player.java

@@ -1,6 +1,7 @@
 package com.velocitypowered.api.proxy;
 
 import com.velocitypowered.api.command.CommandSource;
+import com.velocitypowered.api.proxy.messages.ChannelMessageSink;
 import com.velocitypowered.api.proxy.messages.ChannelMessageSource;
 import com.velocitypowered.api.server.ServerInfo;
 import com.velocitypowered.api.util.MessagePosition;
@@ -13,7 +14,7 @@ import java.util.UUID;
 /**
  * Represents a player who is connected to the proxy.
  */
-public interface Player extends CommandSource, InboundConnection, ChannelMessageSource {
+public interface Player extends CommandSource, InboundConnection, ChannelMessageSource, ChannelMessageSink {
     /**
      * Returns the player's current username.
      * @return the username

api/src/main/java/com/velocitypowered/api/proxy/ServerConnection.java

@@ -1,12 +1,13 @@
 package com.velocitypowered.api.proxy;
 
+import com.velocitypowered.api.proxy.messages.ChannelMessageSink;
 import com.velocitypowered.api.proxy.messages.ChannelMessageSource;
 import com.velocitypowered.api.server.ServerInfo;
 
 /**
  * Represents a connection to a backend server from the proxy for a client.
  */
-public interface ServerConnection extends ChannelMessageSource {
+public interface ServerConnection extends ChannelMessageSource, ChannelMessageSink {
     ServerInfo getServerInfo();
 
     Player getPlayer();

api/src/main/java/com/velocitypowered/api/proxy/messages/ChannelMessageSink.java

@@ -0,0 +1,5 @@
+package com.velocitypowered.api.proxy.messages;
+
+public interface ChannelMessageSink {
+    void sendPluginMessage(ChannelIdentifier identifier, byte[] data);
+}

api/src/main/java/com/velocitypowered/api/proxy/messages/MessageHandler.java

@@ -1,7 +1,7 @@
 package com.velocitypowered.api.proxy.messages;
 
 public interface MessageHandler {
-    ForwardStatus handle(ChannelMessageSource source, ChannelSide side, byte[] data);
+    ForwardStatus handle(ChannelMessageSource source, ChannelSide side, ChannelIdentifier identifier, byte[] data);
 
     enum ForwardStatus {
         FORWARD,

proxy/src/main/java/com/velocitypowered/proxy/connection/backend/VelocityServerConnection.java

@@ -1,8 +1,10 @@
 package com.velocitypowered.proxy.connection.backend;
 
+import com.google.common.base.Preconditions;
 import com.velocitypowered.api.proxy.ConnectionRequestBuilder;
 import com.velocitypowered.api.proxy.Player;
 import com.velocitypowered.api.proxy.ServerConnection;
+import com.velocitypowered.api.proxy.messages.ChannelIdentifier;
 import com.velocitypowered.proxy.config.PlayerInfoForwarding;
 import com.velocitypowered.proxy.connection.MinecraftConnectionAssociation;
 import com.velocitypowered.proxy.protocol.ProtocolConstants;
@@ -11,6 +13,7 @@ import com.velocitypowered.proxy.protocol.netty.MinecraftEncoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftVarintFrameDecoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftVarintLengthEncoder;
 import com.velocitypowered.proxy.protocol.packet.Handshake;
+import com.velocitypowered.proxy.protocol.packet.PluginMessage;
 import com.velocitypowered.proxy.protocol.packet.ServerLogin;
 import com.velocitypowered.proxy.connection.MinecraftConnection;
 import com.velocitypowered.proxy.protocol.StateRegistry;
@@ -141,4 +144,14 @@ public class VelocityServerConnection implements MinecraftConnectionAssociation,
     public String toString() {
         return "[server connection] " + proxyPlayer.getProfile().getName() + " -> " + serverInfo.getName();
     }
+
+    @Override
+    public void sendPluginMessage(ChannelIdentifier identifier, byte[] data) {
+        Preconditions.checkNotNull(identifier, "identifier");
+        Preconditions.checkNotNull(data, "data");
+        PluginMessage message = new PluginMessage();
+        message.setChannel(identifier.getId());
+        message.setData(data);
+        minecraftConnection.write(message);
+    }
 }

proxy/src/main/java/com/velocitypowered/proxy/connection/client/ConnectedPlayer.java

@@ -7,6 +7,7 @@ import com.velocitypowered.api.permission.PermissionFunction;
 import com.velocitypowered.api.permission.PermissionProvider;
 import com.velocitypowered.api.proxy.ConnectionRequestBuilder;
 import com.velocitypowered.api.proxy.ServerConnection;
+import com.velocitypowered.api.proxy.messages.ChannelIdentifier;
 import com.velocitypowered.api.util.MessagePosition;
 import com.velocitypowered.api.proxy.Player;
 import com.velocitypowered.proxy.VelocityServer;
@@ -18,6 +19,7 @@ import com.velocitypowered.proxy.protocol.packet.Chat;
 import com.velocitypowered.proxy.connection.MinecraftConnection;
 import com.velocitypowered.proxy.connection.backend.VelocityServerConnection;
 import com.velocitypowered.proxy.protocol.packet.ClientSettings;
+import com.velocitypowered.proxy.protocol.packet.PluginMessage;
 import com.velocitypowered.proxy.util.ThrowableUtils;
 import com.velocitypowered.api.server.ServerInfo;
 import com.velocitypowered.proxy.protocol.packet.Disconnect;
@@ -263,6 +265,16 @@ public class ConnectedPlayer implements MinecraftConnectionAssociation, Player {
         return permissionFunction.getPermissionSetting(permission).asBoolean();
     }
 
+    @Override
+    public void sendPluginMessage(ChannelIdentifier identifier, byte[] data) {
+        Preconditions.checkNotNull(identifier, "identifier");
+        Preconditions.checkNotNull(data, "data");
+        PluginMessage message = new PluginMessage();
+        message.setChannel(identifier.getId());
+        message.setData(data);
+        connection.write(message);
+    }
+
     private class ConnectionRequestBuilderImpl implements ConnectionRequestBuilder {
         private final ServerInfo info;
 

proxy/src/main/java/com/velocitypowered/proxy/messages/VelocityChannelRegistrar.java

@@ -12,6 +12,7 @@ import java.util.concurrent.ConcurrentHashMap;
 public class VelocityChannelRegistrar implements ChannelRegistrar {
     private static final Logger logger = LogManager.getLogger(VelocityChannelRegistrar.class);
     private final Map<String, MessageHandler> handlers = new ConcurrentHashMap<>();
+    private final Map<String, ChannelIdentifier> identifierMap = new ConcurrentHashMap<>();
 
     @Override
     public void register(MessageHandler handler, ChannelIdentifier... identifiers) {
@@ -22,18 +23,19 @@ public class VelocityChannelRegistrar implements ChannelRegistrar {
 
         for (ChannelIdentifier identifier : identifiers) {
             handlers.put(identifier.getId(), handler);
+            identifierMap.put(identifier.getId(), identifier);
         }
     }
 
     public MessageHandler.ForwardStatus handlePluginMessage(ChannelMessageSource source, ChannelSide side, PluginMessage message) {
         MessageHandler handler = handlers.get(message.getChannel());
-        if (handler == null) {
-            // Nothing we can do.
+        ChannelIdentifier identifier = identifierMap.get(message.getChannel());
+        if (handler == null || identifier == null) {
             return MessageHandler.ForwardStatus.FORWARD;
         }
 
         try {
-            return handler.handle(source, side, message.getData());
+            return handler.handle(source, side, identifier, message.getData());
         } catch (Exception e) {
             logger.info("Unable to handle plugin message on channel {} for {}", message.getChannel(), source);
             // In case of doubt, do not forward the message on.
@@ -43,6 +45,14 @@ public class VelocityChannelRegistrar implements ChannelRegistrar {
 
     @Override
     public void unregister(ChannelIdentifier... identifiers) {
+        for (ChannelIdentifier identifier : identifiers) {
+            Preconditions.checkArgument(identifier instanceof LegacyChannelIdentifier || identifier instanceof MinecraftChannelIdentifier,
+                    "identifier is unknown");
+        }
 
+        for (ChannelIdentifier identifier : identifiers) {
+            handlers.remove(identifier.getId());
+            identifierMap.remove(identifier.getId());
+        }
     }
 }