fixed request/response

HttpMarco · Oct 11, 2024 · fbbd283 · fbbd283
1 parent ffc1df7
commit fbbd283
Show file tree

Hide file tree

Showing 9 changed files with 211 additions and 42 deletions.
diff --git a/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/CommunicationListener.java b/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/CommunicationListener.java
@@ -1,10 +1,9 @@
 package dev.httpmarco.osgan.networking;
 
 import dev.httpmarco.osgan.networking.channel.ChannelTransmit;
-import dev.httpmarco.osgan.networking.packet.BadRequestPacket;
-import dev.httpmarco.osgan.networking.packet.Packet;
-import dev.httpmarco.osgan.networking.packet.RequestPacket;
-import dev.httpmarco.osgan.networking.packet.RequestResponsePacket;
+import dev.httpmarco.osgan.networking.packet.*;
+import io.netty5.channel.Channel;
+import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.experimental.Accessors;
 import org.jetbrains.annotations.NotNull;
@@ -22,7 +21,10 @@ public abstract class CommunicationListener {
     private final Map<Class<? extends Packet>, List<BiConsumer<ChannelTransmit, Packet>>> listeners = new HashMap<>();
     private final Map<String, Function<CommunicationProperty, Packet>> responders = new HashMap<>();
 
-    @Deprecated
+    private final Map<String, List<ChannelTransmit>> servicesWithResponders = new HashMap<>();
+    private final Map<ChannelTransmit, List<String>> respondersByChannel = new HashMap<>();
+
+    private final Map<UUID, PendingRequest> pending = new HashMap<>();
     private final Map<UUID, CommunicationFuture<? extends Packet>> requests = new HashMap<>();
 
     @SuppressWarnings("unchecked")
@@ -59,40 +61,50 @@ public <P extends Packet> P request(String id, Class<P> packet) {
         return this.request(id, packet, new CommunicationProperty());
     }
 
-
-    public void callResponder(ChannelTransmit channelTransmit, @NotNull RequestPacket requestPacket) {
-        if (!responders.containsKey(requestPacket.id())) {
-            channelTransmit.sendPacket(new BadRequestPacket(requestPacket.uuid()));
-            System.out.println("Found no responder for: " + requestPacket.id());
-            return;
-        }
-
+    public void respond(ChannelTransmit channelTransmit, RequestPacket requestPacket) {
         var response = responders.get(requestPacket.id()).apply(requestPacket.property());
         channelTransmit.sendPacket(new RequestResponsePacket(requestPacket.uuid(), response));
     }
 
+    public boolean hasResponder(String id) {
+        return responders.containsKey(id);
+    }
+
     public void responder(String id, Function<CommunicationProperty, Packet> packetFunction) {
         this.responders.put(id, packetFunction);
     }
 
     public <P extends Packet> boolean call(@NotNull P packet, ChannelTransmit channelTransmit) {
         if (packet instanceof RequestPacket requestPacket) {
-            this.callResponder(channelTransmit, requestPacket);
+            requestReceive(channelTransmit, requestPacket);
             return true;
         }
 
         if (packet instanceof RequestResponsePacket requestResponsePacket) {
-            if (!this.requests.containsKey(requestResponsePacket.uuid())) {
-                return true;
-            }
-            ((CommunicationFuture<Packet>) this.requests.get(requestResponsePacket.uuid())).complete(requestResponsePacket.response());
-            this.requests.remove(requestResponsePacket.uuid());
+            responseReceive(channelTransmit, requestResponsePacket);
             return true;
         }
 
         if (packet instanceof BadRequestPacket badRequestPacket) {
-            this.requests.remove(badRequestPacket.uuid());
-            System.out.println("Invalid request from: " + badRequestPacket.uuid());
+            badRequestReceive(channelTransmit, badRequestPacket);
+            return true;
+        }
+
+        if (packet instanceof RegisterResponderPacket registerResponderPacket) {
+            var id = registerResponderPacket.id();
+
+            if (!this.servicesWithResponders.containsKey(id)) {
+                this.servicesWithResponders.put(id, new ArrayList<>());
+            }
+
+            if (!this.respondersByChannel.containsKey(channelTransmit)) {
+                this.respondersByChannel.put(channelTransmit, new ArrayList<>());
+            }
+
+            this.servicesWithResponders.get(id).add(channelTransmit);
+            this.respondersByChannel.get(channelTransmit).add(id);
+
+            System.out.println("Registered responder: " + id);
             return true;
         }
 
@@ -107,4 +119,10 @@ public <P extends Packet> boolean call(@NotNull P packet, ChannelTransmit channe
     }
 
     public abstract void sendPacket(Packet packet);
+
+    public abstract void requestReceive(ChannelTransmit channelTransmit, RequestPacket packet);
+
+    public abstract void badRequestReceive(ChannelTransmit channelTransmit, BadRequestPacket packet);
+
+    public abstract void responseReceive(ChannelTransmit channelTransmit, RequestResponsePacket packet);
 }
diff --git a/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/channel/ChannelTransmit.java b/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/channel/ChannelTransmit.java
@@ -1,15 +1,15 @@
 package dev.httpmarco.osgan.networking.channel;
 
-import dev.httpmarco.osgan.networking.CommunicationAttributes;
 import dev.httpmarco.osgan.networking.CommunicationListener;
+import dev.httpmarco.osgan.networking.packet.BadRequestPacket;
 import dev.httpmarco.osgan.networking.packet.Packet;
+import dev.httpmarco.osgan.networking.packet.RequestPacket;
+import dev.httpmarco.osgan.networking.packet.RequestResponsePacket;
 import io.netty5.channel.Channel;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.experimental.Accessors;
 
-import java.util.UUID;
-
 @Getter
 @AllArgsConstructor
 @Accessors(fluent = true)
@@ -26,6 +26,18 @@ public void sendPacket(Packet packet) {
         this.channel.writeAndFlush(packet);
     }
 
+    @Override
+    public void requestReceive(ChannelTransmit channelTransmit, RequestPacket packet) {
+    }
+
+    @Override
+    public void badRequestReceive(ChannelTransmit channelTransmit, BadRequestPacket packet) {
+    }
+
+    @Override
+    public void responseReceive(ChannelTransmit channelTransmit, RequestResponsePacket packet) {
+    }
+
     @Override
     public boolean equals(Object obj) {
         return obj instanceof ChannelTransmit transmit && transmit.channel.equals(this.channel);

diff --git a/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/client/CommunicationClient.java b/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/client/CommunicationClient.java
@@ -1,11 +1,9 @@
 package dev.httpmarco.osgan.networking.client;
 
-import dev.httpmarco.osgan.networking.CommunicationComponent;
-import dev.httpmarco.osgan.networking.CommunicationNetworkUtils;
-import dev.httpmarco.osgan.networking.CommunicationTransmitHandler;
+import dev.httpmarco.osgan.networking.*;
 import dev.httpmarco.osgan.networking.channel.ChannelInitializer;
 import dev.httpmarco.osgan.networking.channel.ChannelTransmit;
-import dev.httpmarco.osgan.networking.packet.Packet;
+import dev.httpmarco.osgan.networking.packet.*;
 import dev.httpmarco.osgan.networking.security.SecurityChannelParametrize;
 import io.netty5.bootstrap.Bootstrap;
 import io.netty5.channel.ChannelOption;
@@ -18,6 +16,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 @Accessors(fluent = true)
 public final class CommunicationClient extends CommunicationComponent<CommunicationClientAction> {
@@ -84,8 +83,40 @@ public void sendPacket(Packet packet) {
         this.channelTransmit.sendPacket(packet);
     }
 
+    @Override
+    public void requestReceive(ChannelTransmit channelTransmit, RequestPacket packet) {
+        if (!hasResponder(packet.id())) {
+            return;
+        }
+
+        respond(channelTransmit, packet);
+    }
+
+    @Override
+    public void badRequestReceive(ChannelTransmit channelTransmit, BadRequestPacket packet) {
+        requests().remove(packet.uuid());
+        System.out.println("Invalid request (" + packet.uuid() + "): " + packet.message());
+    }
+
+    @Override
+    public void responseReceive(ChannelTransmit channelTransmit, RequestResponsePacket packet) {
+        if (!requests().containsKey(packet.uuid())) {
+            return;
+        }
+
+        ((CommunicationFuture<Packet>) requests().get(packet.uuid())).complete(packet.buildPacket());
+        requests().remove(packet.uuid());
+    }
+
     @Override
     protected void callClientAction(CommunicationClientAction action) {
         this.callClientAction(action, channelTransmit);
     }
+
+    @Override
+    public void responder(String id, Function<CommunicationProperty, Packet> packetFunction) {
+        super.responder(id, packetFunction);
+
+        this.sendPacket(new RegisterResponderPacket(id));
+    }
 }
diff --git a/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/BadRequestPacket.java b/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/BadRequestPacket.java
@@ -12,14 +12,17 @@
 public class BadRequestPacket extends Packet {
 
     private UUID uuid;
+    private String message;
 
     @Override
     public void read(PacketBuffer buffer) {
         this.uuid = buffer.readUniqueId();
+        this.message = buffer.readString();
     }
 
     @Override
     public void write(PacketBuffer buffer) {
         buffer.writeUniqueId(uuid);
+        buffer.writeString(message);
     }
 }
diff --git a/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/PendingRequest.java b/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/PendingRequest.java
@@ -0,0 +1,8 @@
+package dev.httpmarco.osgan.networking.packet;
+
+import dev.httpmarco.osgan.networking.channel.ChannelTransmit;
+
+import java.util.UUID;
+
+public record PendingRequest(ChannelTransmit transmit, String id, UUID uniqueId, long timestamp) {
+}
diff --git a/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/RegisterResponderPacket.java b/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/RegisterResponderPacket.java
@@ -0,0 +1,22 @@
+package dev.httpmarco.osgan.networking.packet;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.experimental.Accessors;
+
+@Getter
+@Accessors(fluent = true)
+@AllArgsConstructor
+public class RegisterResponderPacket extends Packet {
+    private String id;
+
+    @Override
+    public void read(PacketBuffer buffer) {
+        this.id = buffer.readString();
+    }
+
+    @Override
+    public void write(PacketBuffer buffer) {
+        buffer.writeString(id);
+    }
+}
diff --git a/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/RequestResponsePacket.java b/osgan-netty/src/main/java/dev/httpmarco/osgan/networking/packet/RequestResponsePacket.java
@@ -5,30 +5,46 @@
 import lombok.SneakyThrows;
 import lombok.experimental.Accessors;
 
+import java.util.Objects;
 import java.util.UUID;
+import java.util.function.Function;
 
 @Getter
 @Accessors(fluent = true)
 @AllArgsConstructor
 public class RequestResponsePacket extends Packet {
 
     private UUID uuid;
-    private Packet response;
+    private String packetClass;
+    private PacketBuffer buffer;
+
+    public RequestResponsePacket(UUID uuid, Packet packet) {
+        this.uuid = uuid;
+        this.packetClass = packet.getClass().getName();
+        this.buffer = PacketBuffer.allocate();
+        packet.write(this.buffer);
+    }
+
+    @SneakyThrows
+    public Packet buildPacket() {
+        var packet = (Packet) PacketAllocator.allocate(Class.forName(this.packetClass));
+        Objects.requireNonNull(packet).read(this.buffer);
+        return packet;
+    }
 
     @Override
     @SneakyThrows
     public void read(PacketBuffer buffer) {
         this.uuid = buffer.readUniqueId();
-
-        this.response = (Packet) PacketAllocator.allocate(Class.forName(buffer.readString()));
-        this.response.read(buffer);
+        this.packetClass = buffer.readString();
+        this.buffer = new PacketBuffer(buffer.getOrigin().copy());
     }
 
     @Override
     public void write(PacketBuffer buffer) {
         buffer.writeUniqueId(uuid);
-        buffer.writeString(response.getClass().getName());
+        buffer.writeString(packetClass);
 
-        response.write(buffer);
+        buffer.writeBytes(this.buffer.getOrigin());
     }
 }