allow interruptions in Result

tim-smart · Feb 7, 2024 · ea179de · ea179de
1 parent 1d6a8e0
commit ea179de
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 8 deletions.
diff --git a/.changeset/soft-crabs-admire.md b/.changeset/soft-crabs-admire.md
@@ -0,0 +1,5 @@
+---
+"@effect-rx/rx": patch
+---
+
+allow interruptions in Result
diff --git a/packages/rx/src/Result.ts b/packages/rx/src/Result.ts
@@ -175,6 +175,13 @@ export interface Failure<E, A> extends Result.Proto<E, A> {
  */
 export const isFailure = <E, A>(result: Result<E, A>): result is Failure<E, A> => result._tag === "Failure"
 
+/**
+ * @since 1.0.0
+ * @category refinements
+ */
+export const isInterrupted = <E, A>(result: Result<E, A>): result is Failure<E, A> =>
+  result._tag === "Failure" && Cause.isInterruptedOnly(result.cause)
+
 /**
  * @since 1.0.0
  * @category constructors

diff --git a/packages/rx/src/Rx.ts b/packages/rx/src/Rx.ts
@@ -463,9 +463,7 @@ function makeEffect<E, A>(
   const cancel = runCallbackSync(runtime)(
     scopedEffect,
     function(exit) {
-      if (!Exit.isInterrupted(exit)) {
-        ctx.setSelfSync(Result.fromExitWithPrevious(exit, previous))
-      }
+      ctx.setSelfSync(Result.fromExitWithPrevious(exit, previous))
     }
   )
   if (cancel !== undefined) {
@@ -592,9 +590,7 @@ function makeStream<E, A>(
     ),
     (exit) => {
       if (exit._tag === "Failure") {
-        if (!Exit.isInterrupted(exit)) {
-          ctx.setSelfSync(Result.failureWithPrevious(exit.cause, previous))
-        }
+        ctx.setSelfSync(Result.failureWithPrevious(exit.cause, previous))
       } else {
         pipe(
           ctx.self<Result.Result<E | NoSuchElementException, A>>(),

diff --git a/packages/rx/src/internal/runtime.ts b/packages/rx/src/internal/runtime.ts
@@ -2,7 +2,7 @@
  * @since 1.0.0
  */
 import { NoSuchElementException } from "effect/Cause"
-import * as Effect from "effect/Effect"
+import type * as Effect from "effect/Effect"
 import * as Exit from "effect/Exit"
 import * as Runtime from "effect/Runtime"
 import { SyncScheduler } from "effect/Scheduler"
@@ -49,6 +49,7 @@ export const runCallbackSync =
     }
     fiberRuntime.addObserver(onExit)
     return function() {
-      Effect.runFork(fiberRuntime.interruptAsFork(fiberRuntime.id()))
+      fiberRuntime.removeObserver(onExit)
+      fiberRuntime.unsafeInterruptAsFork(fiberRuntime.id())
     }
   }