Incomplete something

2023-07-19 15:54:17 +05:00
parent a42f8367a4
commit 218411902e
12 changed files with 167 additions and 52 deletions
@@ -1,50 +1,44 @@
 package com.seibel.distanthorizons.core.file.fullDatafile;

-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.Multimap;
 import com.seibel.distanthorizons.core.dataObjects.fullData.sources.interfaces.IFullDataSource;
 import com.seibel.distanthorizons.core.file.structure.AbstractSaveStructure;
 import com.seibel.distanthorizons.core.level.IDhLevel;
 import com.seibel.distanthorizons.core.network.ChildNetworkEventSource;
 import com.seibel.distanthorizons.core.network.NetworkClient;
+import com.seibel.distanthorizons.core.network.future.NetworkRequestTracker;
+import com.seibel.distanthorizons.core.network.messages.ChunkRequestMessage;
+import com.seibel.distanthorizons.core.network.messages.ChunkResponseMessage;
 import com.seibel.distanthorizons.core.pos.DhSectionPos;
-import io.netty.channel.ChannelHandlerContext;

 import java.util.concurrent.CompletableFuture;

 public class RemoteFullDataFileHandler extends FullDataFileHandler
 {
-    private final Multimap<ChannelHandlerContext, ChunkRequest> chunkRequests = HashMultimap.create();
-
-    public RemoteFullDataFileHandler(IDhLevel level, AbstractSaveStructure saveStructure, ChildNetworkEventSource<NetworkClient> eventSource) {
+    private final NetworkClient networkClient;
+    private final NetworkRequestTracker<ChunkResponseMessage, DhSectionPos> chunkRequestTracker;
+    
+    public RemoteFullDataFileHandler(IDhLevel level, AbstractSaveStructure saveStructure, NetworkClient networkClient) {
        super(level, saveStructure);
-        this.registerNetworkHandlers(eventSource);
-    }
-
-    private void registerNetworkHandlers(ChildNetworkEventSource<NetworkClient> eventSource) {
-        //eventSource.registerHandler();
+        this.networkClient = networkClient;
+        this.chunkRequestTracker = new NetworkRequestTracker<>(networkClient, ChunkResponseMessage.class);
    }

    @Override
    public CompletableFuture<IFullDataSource> read(DhSectionPos pos) {
        // TODO read and force update somehow instead ????
-        return super.read(pos).handle((fullDataSource, throwable) -> {
-            if (fullDataSource == null) {
-
+        return super.read(pos).thenCompose((fullDataSource) -> {
+            if (fullDataSource != null) {
+                return CompletableFuture.completedFuture(fullDataSource);
            }

-            return fullDataSource;
+            CompletableFuture<ChunkResponseMessage> responseFuture = chunkRequestTracker.sendRequest(networkClient.getChannel(), new ChunkRequestMessage());
+            
        });
    }

    @Override
    public void close() {
        super.close();
-
-
-    }
-
-    private static class ChunkRequest {
-
+        chunkRequestTracker.close();
    }
 }
@@ -32,11 +32,11 @@ public class DhClientLevel extends DhLevel implements IDhClientLevel
 	// constructor //
 	//=============//
 	
-	public DhClientLevel(AbstractSaveStructure saveStructure, IClientLevelWrapper clientLevelWrapper, ChildNetworkEventSource<NetworkClient> eventSource)
+	public DhClientLevel(AbstractSaveStructure saveStructure, IClientLevelWrapper clientLevelWrapper, NetworkClient networkClient)
 	{
 		this.levelWrapper = clientLevelWrapper;
 		this.saveStructure = saveStructure;
-		dataFileHandler = new RemoteFullDataFileHandler(this, saveStructure, eventSource);
+		dataFileHandler = new RemoteFullDataFileHandler(this, saveStructure, networkClient);
 		clientside = new ClientLevelModule(this);
 		clientside.startRenderer();
 		LOGGER.info("Started DHLevel for "+this.levelWrapper+" with saves at "+this.saveStructure);
@@ -134,6 +134,10 @@ public class NetworkClient extends NetworkEventSource implements AutoCloseable
 		this.channel.disconnect();
    }
 	
+	public Channel getChannel() {
+		return channel;
+	}
+	
    @Override
    public void close() 
 	{
@@ -0,0 +1,8 @@
+package com.seibel.distanthorizons.core.network.future;
+
+import com.seibel.distanthorizons.core.network.protocol.INetworkMessage;
+
+public interface IFutureTrackableNetworkMessage<TKey extends Comparable<TKey>> extends INetworkMessage
+{
+	TKey getRequestKey();
+}
@@ -0,0 +1,73 @@
+package com.seibel.distanthorizons.core.network.future;
+
+import com.google.common.collect.HashBasedTable;
+import com.google.common.collect.Table;
+import com.seibel.distanthorizons.core.network.ChildNetworkEventSource;
+import com.seibel.distanthorizons.core.network.NetworkEventSource;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundInvoker;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+public class NetworkRequestTracker
+		<TResponse extends IFutureTrackableNetworkMessage<TKey>, TKey extends Comparable<TKey>>
+	implements AutoCloseable
+{
+	private final ChildNetworkEventSource<?> eventSource;
+	
+	private final Set<Channel> knownChannels = new HashSet<>();
+	private final Table<Channel, TKey, CompletableFuture<TResponse>> pendingFutures = HashBasedTable.create();
+	
+	public NetworkRequestTracker(
+			NetworkEventSource eventSource,
+			Class<TResponse> responseClass)
+	{
+		this.eventSource = new ChildNetworkEventSource<>(eventSource);
+		registerNetworkHandlers(responseClass);
+	}
+	
+	private void registerNetworkHandlers(Class<TResponse> responseClass)
+	{
+		this.eventSource.registerHandler(responseClass, (msg, ctx) -> {
+			CompletableFuture<TResponse> future = pendingFutures.remove(ctx.channel(), msg.getRequestKey());
+			if (future != null) {
+				future.complete(msg);
+			}
+		});
+	}
+	
+	public CompletableFuture<TResponse> sendRequest(Channel channel, IFutureTrackableNetworkMessage<TKey> msg)
+	{
+		if (knownChannels.add(channel))
+		{
+			channel.closeFuture().addListener(closeFuture ->
+			{
+				pendingFutures.row(channel).values().removeIf(responseFuture ->
+				{
+					responseFuture.completeExceptionally(closeFuture.cause());
+					return true;
+				});
+			});
+		}
+		
+		CompletableFuture<TResponse> responseFuture = new CompletableFuture<>();
+		pendingFutures.put(channel, msg.getRequestKey(), responseFuture);
+		
+		channel.writeAndFlush(msg).addListener(writeFuture -> {
+			if (writeFuture.cause() != null) {
+				responseFuture.completeExceptionally(writeFuture.cause());
+			}
+		});
+		return responseFuture;
+	}
+	
+	@Override public void close()
+	{
+		this.eventSource.close();
+	}
+}
@@ -0,0 +1,24 @@
+package com.seibel.distanthorizons.core.network.messages;
+
+import com.seibel.distanthorizons.core.network.future.IFutureTrackableNetworkMessage;
+import com.seibel.distanthorizons.core.pos.DhSectionPos;
+import io.netty.buffer.ByteBuf;
+
+public class ChunkRequestMessage implements IFutureTrackableNetworkMessage<DhSectionPos>
+{
+	public DhSectionPos dhSectionPos;
+	
+	@Override public DhSectionPos getRequestKey() { return dhSectionPos; }
+	
+    @Override
+    public void encode(ByteBuf out) 
+	{
+
+    }
+
+    @Override
+    public void decode(ByteBuf in) 
+	{
+
+    }
+}
@@ -0,0 +1,22 @@
+package com.seibel.distanthorizons.core.network.messages;
+
+import com.seibel.distanthorizons.core.network.future.IFutureTrackableNetworkMessage;
+import com.seibel.distanthorizons.core.pos.DhSectionPos;
+import io.netty.buffer.ByteBuf;
+
+public class ChunkResponseMessage implements IFutureTrackableNetworkMessage<DhSectionPos>
+{
+	public DhSectionPos dhSectionPos;
+	
+	@Override public DhSectionPos getRequestKey() { return dhSectionPos; }
+	
+	@Override public void encode(ByteBuf out)
+	{
+	
+	}
+	
+	@Override public void decode(ByteBuf in)
+	{
+	
+	}
+}
@@ -1,21 +0,0 @@
-package com.seibel.distanthorizons.core.network.messages;
-
-import com.seibel.distanthorizons.core.network.protocol.INetworkMessage;
-import io.netty.buffer.ByteBuf;
-
-public class RequestChunksMessage implements INetworkMessage
-{
-
-    @Override
-    public void encode(ByteBuf out) 
-	{
-
-    }
-
-    @Override
-    public void decode(ByteBuf in) 
-	{
-
-    }
-	
-}
@@ -29,7 +29,7 @@ public class MessageRegistry
 		this.registerMessage(AckMessage.class, AckMessage::new);
 		this.registerMessage(PlayerUUIDMessage.class, PlayerUUIDMessage::new);
 		this.registerMessage(RemotePlayerConfigMessage.class, RemotePlayerConfigMessage::new);
-		this.registerMessage(RequestChunksMessage.class, RequestChunksMessage::new);
+		this.registerMessage(ChunkRequestMessage.class, ChunkRequestMessage::new);
 	}
 	
 	
@@ -3,6 +3,7 @@ package com.seibel.distanthorizons.core.pos;
 import com.seibel.distanthorizons.core.enums.ELodDirection;
 import com.seibel.distanthorizons.coreapi.util.BitShiftUtil;
 import com.seibel.distanthorizons.core.util.LodUtil;
+import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;

 import java.util.function.Consumer;
@@ -18,7 +19,7 @@ import java.util.function.Consumer;
 * @author Leetom
 * @version 2022-11-6
 */
-public class DhSectionPos
+public class DhSectionPos implements Comparable<DhSectionPos>
 {
 	/** 
 	 * The lowest detail level a Section position can hold.
@@ -238,6 +239,18 @@ public class DhSectionPos
 				this.sectionZ == that.sectionZ;
 	}
 	
+	@Override public int compareTo(@NotNull DhSectionPos o)
+	{
+		if (this.sectionDetailLevel != o.sectionDetailLevel)
+			return this.sectionDetailLevel - o.sectionDetailLevel;
+		if (this.sectionX != o.sectionX)
+			return this.sectionX - o.sectionX;
+		if (this.sectionZ != o.sectionZ)
+			return this.sectionZ - o.sectionZ;
+		
+		return 0;
+	}
+	
 	@Override
 	public int hashCode()
 	{
@@ -245,5 +258,4 @@ public class DhSectionPos
 				Integer.hashCode(this.sectionX) ^ // XOR
 				Integer.hashCode(this.sectionZ);
 	}
-	
 }
@@ -4,7 +4,6 @@ import com.seibel.distanthorizons.core.dependencyInjection.SingletonInjector;
 import com.seibel.distanthorizons.core.file.structure.ClientOnlySaveStructure;
 import com.seibel.distanthorizons.core.level.IDhLevel;
 import com.seibel.distanthorizons.core.level.DhClientLevel;
-import com.seibel.distanthorizons.core.network.ChildNetworkEventSource;
 import com.seibel.distanthorizons.core.network.NetworkClient;
 import com.seibel.distanthorizons.core.network.messages.*;
 import com.seibel.distanthorizons.core.network.messages.PlayerUUIDMessage;
@@ -65,7 +64,7 @@ public class DhClientWorld extends AbstractDhWorld implements IDhClientWorld

        networkClient.registerAckHandler(RemotePlayerConfigMessage.class, ctx -> {
            // TODO Actually request chunks
-            ctx.writeAndFlush(new RequestChunksMessage());
+            ctx.writeAndFlush(new ChunkRequestMessage());
        });
    }

@@ -86,7 +85,7 @@ public class DhClientWorld extends AbstractDhWorld implements IDhClientWorld
 				return null;
 			}

-			return new DhClientLevel(this.saveStructure, clientLevelWrapper, new ChildNetworkEventSource<>(networkClient));
+			return new DhClientLevel(this.saveStructure, clientLevelWrapper, networkClient);
        });
    }

@@ -7,7 +7,7 @@ import com.seibel.distanthorizons.core.level.DhServerLevel;
 import com.seibel.distanthorizons.core.level.IDhLevel;
 import com.seibel.distanthorizons.core.network.NetworkServer;
 import com.seibel.distanthorizons.core.network.messages.*;
-import com.seibel.distanthorizons.core.network.messages.RequestChunksMessage;
+import com.seibel.distanthorizons.core.network.messages.ChunkRequestMessage;
 import com.seibel.distanthorizons.core.network.objects.RemotePlayer;
 import com.seibel.distanthorizons.core.util.LodUtil;
 import com.seibel.distanthorizons.core.wrapperInterfaces.misc.IServerPlayerWrapper;
@@ -86,7 +86,7 @@ public class DhServerWorld extends AbstractDhWorld implements IDhServerWorld
 			channelContext.writeAndFlush(new AckMessage(RemotePlayerConfigMessage.class));
 		});
 		
-		this.networkServer.registerHandler(RequestChunksMessage.class, (msg, ctx) -> 
+		this.networkServer.registerHandler(ChunkRequestMessage.class, (msg, ctx) ->
 		{
 			LOGGER.info("RequestChunksMessage");
 			// hasReceivedChunkRequest should be false somewhere ???