async, asyncInterrupt, sleep

Author: SandroMaglione

packages/fpdart/lib/src/async_context.dart

@@ -0,0 +1,12 @@
+part of "effect.dart";
+
+class AsyncContext<L, R> {
+  final _deferred = Deferred<L, R>.unsafeMake();
+
+  void succeed(R value) => _deferred.unsafeCompleteExit(Right(value));
+  void fail(L error) => _deferred.unsafeCompleteExit(Left(Failure(error)));
+  void failCause(Cause<L> cause) => _deferred.unsafeCompleteExit(Left(cause));
+  void die(dynamic defect) => _deferred.unsafeCompleteExit(Left(
+        Die(defect, StackTrace.current),
+      ));
+}

packages/fpdart/lib/src/effect.dart

@@ -6,6 +6,7 @@ import './extension/future_or_extension.dart';
 import './extension/iterable_extension.dart';
 import 'unit.dart' as fpdart_unit;
 
+part 'async_context.dart';
 part 'context.dart';
 part 'deferred.dart';
 part 'either.dart';
@@ -152,18 +153,70 @@ final class Effect<E, L, R> extends IEffect<E, L, R> {
         (_) => f().then(Right.new),
       );
 
+  /// {@category constructors}
+  factory Effect.fail(L value) => Effect.from((_) => Left(Failure(value)));
+
+  /// {@category constructors}
+  factory Effect.failCause(Cause<L> cause) => Effect.from((_) => Left(cause));
+
+  /// {@category constructors}
+  factory Effect.succeed(R value) => Effect.from((_) => Right(value));
+
   /// {@category constructors}
   factory Effect.lazy(Effect<E, L, R> Function() effect) =>
       Effect.from((context) => effect()._unsafeRun(context));
 
   /// {@category constructors}
-  factory Effect.fail(L value) => Effect.from((_) => Left(Failure(value)));
+  factory Effect.async(void Function(AsyncContext<L, R> resume) callback) =>
+      Effect.from(
+        (context) {
+          final asyncContext = AsyncContext<L, R>();
+          callback(asyncContext);
+          return asyncContext._deferred.wait<E>()._unsafeRun(context);
+        },
+      );
 
   /// {@category constructors}
-  factory Effect.failCause(Cause<L> cause) => Effect.from((_) => Left(cause));
+  factory Effect.asyncInterrupt(
+    Effect<Null, Never, Unit> Function(AsyncContext<L, R> resume) callback,
+  ) =>
+      Effect.from((context) {
+        final asyncContext = AsyncContext<L, R>();
+
+        final finalizer = callback(asyncContext);
+        if (asyncContext._deferred.unsafeCompleted) {
+          return asyncContext._deferred.wait<E>()._unsafeRun(context);
+        }
+
+        final interruption = context.signal.wait<E>().alwaysIgnore(
+              finalizer.withEnv<E>(),
+            );
+
+        return asyncContext._deferred
+            .wait<E>()
+            .race(interruption)
+            ._unsafeRun(context.withoutSignal);
+      });
 
   /// {@category constructors}
-  factory Effect.succeed(R value) => Effect.from((_) => Right(value));
+  static Effect<E, L, void> sleep<E, L>(Duration duration) =>
+      Effect.asyncInterrupt(
+        (resume) {
+          final timer = Timer(duration, () {
+            resume.succeed(null);
+          });
+
+          if (resume._deferred.unsafeCompleted) {
+            timer.cancel();
+            return resume._deferred.wait<Null>().match(
+                  onFailure: (_) => fpdart_unit.unit,
+                  onSuccess: (_) => fpdart_unit.unit,
+                );
+          }
+
+          return Effect.unit();
+        },
+      );
 
   /// {@category constructors}
   factory Effect.raceAll(Iterable<Effect<E, L, R>> iterable) =>
@@ -189,14 +242,6 @@ final class Effect<E, L, R> extends IEffect<E, L, R> {
             );
       });
 
-  /// {@category constructors}
-  static Effect<E, L, void> sleep<E, L>(Duration duration) => Effect.from(
-        (_) => Future.delayed(
-          duration,
-          () => const Right(null),
-        ),
-      );
-
   /// {@category constructors}
   static Effect<E, Never, Never> die<E>(dynamic defect) => Effect.from(
         (_) => Left(Die.current(defect)),

packages/fpdart/test/src/effect/effect_constructors_test.dart

@@ -17,6 +17,44 @@ void main() {
         expect(result, "error");
       });
 
+      group('async', () {
+        test('succeed callback', () async {
+          final main = Effect<Null, String, int>.async(
+            (resume) => resume.succeed(10),
+          );
+          final result = await main.runFuture();
+          expect(result, 10);
+        });
+
+        test('fail callback', () async {
+          final main = Effect<Null, String, int>.async(
+            (resume) => resume.fail("error"),
+          );
+          final result = await main.flip().runFuture();
+          expect(result, "error");
+        });
+
+        test('succeed async callback', () async {
+          final main = Effect<Null, String, int>.async(
+            (resume) => Future.delayed(Duration(milliseconds: 100)).then(
+              (_) => resume.succeed(10),
+            ),
+          );
+          final result = await main.runFuture();
+          expect(result, 10);
+        });
+
+        test('fail async callback', () async {
+          final main = Effect<Null, String, int>.async(
+            (resume) => Future.delayed(Duration(milliseconds: 100)).then(
+              (_) => resume.fail("error"),
+            ),
+          );
+          final result = await main.flip().runFuture();
+          expect(result, "error");
+        });
+      });
+
       group('tryCatch', () {
         test('executes once', () {
           var mutable = 0;