new: preDowngrade hook

Author: ishland

FlowSched

@@ -6,8 +6,6 @@
 import io.reactivex.rxjava3.core.Completable;
 import net.minecraft.world.chunk.ChunkStatus;
 
-import java.util.concurrent.CompletableFuture;
-
 public class Deferred extends NewChunkStatus {
 
     public Deferred(int ordinal) {

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/statuses/Deferred.java

@@ -20,6 +20,7 @@
 import com.ishland.flowsched.util.Assertions;
 import io.reactivex.rxjava3.annotations.NonNull;
 import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.CompletableObserver;
 import io.reactivex.rxjava3.core.Single;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import net.minecraft.nbt.NbtElement;
@@ -120,18 +121,24 @@ protected Completable finalizeLoading(ChunkLoadingContext context, Single<ProtoC
     }
 
     @Override
-    public Completable downgradeFromThis(ChunkLoadingContext context, Cancellable cancellable) {
+    public Completable preDowngradeFromThis(ChunkLoadingContext context, Cancellable cancellable) {
         return Completable.defer(() -> Completable.fromCompletionStage(syncWithLightEngine(context)))
                 .observeOn(Schedulers.from(((IThreadedAnvilChunkStorage) context.tacs()).getMainThreadExecutor()))
-                .doOnEvent(throwable -> {
-                    if (throwable != null) return;
+                .andThen((CompletableObserver observer) -> {
+                    if (context.holder().getTargetStatus().ordinal() >= this.ordinal()) { // saving cancelled
+                        cancellable.cancel();
+                        observer.onError(new CancellationException());
+                    } else {
+                        observer.onComplete();
+                    }
+                });
+    }
+
+    @Override
+    public Completable downgradeFromThis(ChunkLoadingContext context, Cancellable cancellable) {
+        return Completable.defer(() -> {
                     Assertions.assertTrue(((IThreadedAnvilChunkStorage) context.tacs()).getMainThreadExecutor().isOnThread());
                     try (var ignored = ThreadInstrumentation.getCurrent().begin(new ChunkTaskWork(context, this, false))) {
-                        if (context.holder().getTargetStatus().ordinal() >= this.ordinal()) { // saving cancelled
-                            cancellable.cancel();
-                            throw new CancellationException();
-                        }
-
                         final ChunkState chunkState = context.holder().getItem().get();
                         Chunk chunk = chunkState.chunk();
                         if (chunk instanceof WrapperProtoChunk protoChunk) chunk = protoChunk.getWrappedChunk();
@@ -174,6 +181,8 @@ public Completable downgradeFromThis(ChunkLoadingContext context, Cancellable ca
                         ((IPOIUnloading) ((IThreadedAnvilChunkStorage) context.tacs()).getPointOfInterestStorage()).c2me$unloadPoi(context.holder().getKey());
 
                         context.holder().getItem().set(new ChunkState(null, null, null));
+
+                        return Completable.complete();
                     }
                 })
                 .doOnError((throwable) -> {

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/statuses/ReadFromDisk.java

@@ -28,7 +28,6 @@
 import com.ishland.flowsched.util.Assertions;
 import io.reactivex.rxjava3.annotations.NonNull;
 import io.reactivex.rxjava3.core.Completable;
-import io.reactivex.rxjava3.core.CompletableSource;
 import io.reactivex.rxjava3.core.Single;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import it.unimi.dsi.fastutil.Pair;
@@ -50,9 +49,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.CancellationException;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Function;
 
 public class ReadFromDiskAsync extends ReadFromDisk {
 
@@ -140,19 +137,18 @@ protected Single<ProtoChunk> invokeAsyncLoad(ChunkLoadingContext context) {
                 });
     }
 
+    @Override
+    public Completable preDowngradeFromThis(ChunkLoadingContext context, Cancellable cancellable) {
+        return super.preDowngradeFromThis(context, cancellable);
+    }
+
     @Override
     public Completable downgradeFromThis(ChunkLoadingContext context, Cancellable cancellable) {
         final AtomicBoolean loadedToWorld = new AtomicBoolean(false);
-        return Completable.defer(() -> Completable.fromCompletionStage(syncWithLightEngine(context)))
-                .observeOn(Schedulers.from(((IThreadedAnvilChunkStorage) context.tacs()).getMainThreadExecutor()))
-                .andThen(Completable.defer(() -> {
+        return Completable
+                .defer(() -> {
                     Assertions.assertTrue(((IThreadedAnvilChunkStorage) context.tacs()).getMainThreadExecutor().isOnThread());
                     try (var ignored = ThreadInstrumentation.getCurrent().begin(new ChunkTaskWork(context, this, false))) {
-                        if (context.holder().getTargetStatus().ordinal() >= this.ordinal()) { // saving cancelled
-                            cancellable.cancel();
-                            return Completable.error(new CancellationException());
-                        }
-
                         final ChunkState chunkState = context.holder().getItem().get();
                         Chunk chunk = chunkState.chunk();
                         if (chunk instanceof WrapperProtoChunk protoChunk) chunk = protoChunk.getWrappedChunk();
@@ -173,10 +169,9 @@ public Completable downgradeFromThis(ChunkLoadingContext context, Cancellable ca
                             return asyncSave(context.tacs(), chunk);
                         }
                     }
-                }))
+                })
                 .observeOn(Schedulers.from(((IThreadedAnvilChunkStorage) context.tacs()).getMainThreadExecutor()))
-                .doOnEvent(throwable -> {
-                    if (throwable != null) return;
+                .doOnComplete(() -> {
                     Assertions.assertTrue(((IThreadedAnvilChunkStorage) context.tacs()).getMainThreadExecutor().isOnThread());
                     try (var ignored = ThreadInstrumentation.getCurrent().begin(new ChunkTaskWork(context, this, false))) {
                         Chunk chunk = context.holder().getItem().get().chunk();

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/statuses/ReadFromDiskAsync.java

@@ -27,7 +27,6 @@
 import net.minecraft.world.chunk.WrapperProtoChunk;
 
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
 
 public class ServerAccessible extends NewChunkStatus {
 

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/statuses/ServerAccessible.java

@@ -10,8 +10,6 @@
 import net.minecraft.util.math.ChunkPos;
 import net.minecraft.world.chunk.ChunkStatus;
 
-import java.util.concurrent.CompletableFuture;
-
 public class ServerAccessibleChunkSending extends NewChunkStatus {
 
     private static final KeyStatusPair<ChunkPos, ChunkState, ChunkLoadingContext>[] deps;

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/statuses/ServerAccessibleChunkSending.java

@@ -20,8 +20,6 @@
 import net.minecraft.world.chunk.ChunkStatus;
 import net.minecraft.world.chunk.WorldChunk;
 
-import java.util.concurrent.CompletableFuture;
-
 public class ServerEntityTicking extends NewChunkStatus {
 
     private static final KeyStatusPair<ChunkPos, ChunkState, ChunkLoadingContext>[] deps;