refactor(fx): use call from effection (#20)

deps.ts

@@ -15,6 +15,7 @@ export type {
 } from "https://deno.land/x/[email protected]/mod.ts";
 export {
   action,
+  call,
   createChannel,
   createContext,
   createQueue,

fx/call.test.ts

@@ -1,6 +1,5 @@
 import { describe, expect, it } from "../test.ts";
 import { run } from "../deps.ts";
-
 import { call } from "./call.ts";
 
 const tests = describe("call()");
@@ -31,28 +30,6 @@ it(tests, "should return an Err()", async () => {
   });
 });
 
-it(tests, "should call a normal function with no params", async () => {
-  function me() {
-    return "valid";
-  }
-
-  await run(function* () {
-    const result = yield* call(me);
-    expect(result).toEqual("valid");
-  });
-});
-
-it(tests, "should call a normal function with params", async () => {
-  function me(v: string) {
-    return "valid " + v;
-  }
-
-  await run(function* () {
-    const result = yield* call(() => me("fn"));
-    expect(result).toEqual("valid fn");
-  });
-});
-
 it(tests, "should call a promise", async () => {
   const me = () =>
     new Promise<string>((resolve) => {

fx/call.ts

@@ -1,38 +1,12 @@
-import type { OpFn } from "../types.ts";
 import type { Operation, Result } from "../deps.ts";
-import { action, Err, expect, Ok } from "../deps.ts";
+import { call, Err, Ok } from "../deps.ts";
+import type { Operator } from "../types.ts";
 
-export const isFunc = (f: unknown) => typeof f === "function";
-export const isPromise = (p: unknown) =>
-  p && isFunc((p as PromiseLike<unknown>).then);
-export const isIterator = (it: unknown) =>
-  it &&
-  isFunc((it as Iterator<unknown>).next) &&
-  isFunc((it as Iterator<unknown>).throw);
+export { call };
 
-export function* toOperation<T>(opFn: OpFn<T>): Operation<T> {
-  const op = opFn();
-  let result: T;
-  if (isPromise(op)) {
-    result = yield* expect(op as Promise<T>);
-  } else if (isIterator(op)) {
-    result = yield* op as Operation<T>;
-  } else {
-    result = op as T;
-  }
-  return result;
-}
-
-export function call<T>(op: OpFn<T>): Operation<T> {
-  return action(function* (resolve) {
-    const result = yield* toOperation(op);
-    resolve(result);
-  });
-}
-
-export function* safe<T>(opFn: OpFn<T>): Operation<Result<T>> {
+export function* safe<T>(operator: Operator<T>): Operation<Result<T>> {
   try {
-    const value = yield* call(opFn);
+    const value: T = yield* call(operator as any) as any;
     return Ok(value);
   } catch (error) {
     return Err(error);

fx/parallel.ts

@@ -1,5 +1,5 @@
 import type { Channel, Operation, Result } from "../deps.ts";
-import type { Computation, OpFn } from "../types.ts";
+import type { Computation, Operator } from "../types.ts";
 import { createChannel, resource, spawn } from "../deps.ts";
 
 import { safe } from "./call.ts";
@@ -9,7 +9,7 @@ export interface ParallelRet<T> extends Computation<Result<T>[]> {
   immediate: Channel<Result<T>, void>;
 }
 
-export function parallel<T>(operations: OpFn<T>[]) {
+export function parallel<T>(operations: Operator<T>[]) {
   const sequence = createChannel<Result<T>>();
   const immediate = createChannel<Result<T>>();
   const results: Result<T>[] = [];

fx/race.ts

@@ -1,16 +1,21 @@
 import type { Operation, Task } from "../deps.ts";
 import { action, resource, spawn } from "../deps.ts";
-import type { OpFn } from "../types.ts";
-
-import { toOperation } from "./call.ts";
+import type { Operator } from "../types.ts";
+import { call } from "./call.ts";
 
 interface OpMap<T = unknown> {
-  [key: string]: OpFn<T>;
+  [key: string]: Operator<T>;
 }
 
-export function race(
+export function race<T>(
   opMap: OpMap,
-): Operation<{ [K in keyof OpMap<unknown>]: ReturnType<OpMap[K]> }> {
+): Operation<
+  {
+    [K in keyof OpMap<T>]: OpMap[K] extends (...args: any[]) => any
+      ? ReturnType<OpMap[K]>
+      : OpMap[K];
+  }
+> {
   return resource(function* Race(provide) {
     const keys = Object.keys(opMap);
     const taskMap: { [key: string]: Task<unknown> } = {};
@@ -20,7 +25,9 @@ export function race(
       for (let i = 0; i < keys.length; i += 1) {
         const key = keys[i];
         yield* spawn(function* () {
-          const task = yield* spawn(() => toOperation(opMap[key]));
+          const task = yield* spawn(function* () {
+            yield* call(opMap[key] as any);
+          });
           taskMap[key] = task;
           (resultMap as any)[key] = yield* task;
           resolve(task);

fx/request.ts

@@ -1,12 +1,12 @@
-import { expect, useAbortSignal } from "../deps.ts";
+import { call, useAbortSignal } from "../deps.ts";
 
 export function* request(url: string | URL | Request, opts?: RequestInit) {
   const signal = yield* useAbortSignal();
-  const response = yield* expect(fetch(url, { signal, ...opts }));
+  const response = yield* call(fetch(url, { signal, ...opts }));
   return response;
 }
 
 export function* json(response: Response) {
-  const result = yield* expect(response.json());
+  const result = yield* call(response.json());
   return result;
 }

fx/watch.ts

@@ -1,17 +1,16 @@
-import type { OpFn } from "../types.ts";
-
+import type { Operator } from "../types.ts";
 import { safe } from "./call.ts";
 import { parallel } from "./parallel.ts";
 
-export function supervise<T>(op: OpFn<T>) {
+export function supervise<T>(op: Operator<T>) {
   return function* () {
     while (true) {
       yield* safe(op);
     }
   };
 }
 
-export function* keepAlive(ops: OpFn[]) {
+export function* keepAlive(ops: Operator<unknown>[]) {
   const results = yield* parallel(ops.map(supervise));
   return yield* results;
 }

query/middleware.ts

@@ -1,6 +1,6 @@
 import { call } from "../fx/mod.ts";
 import { compose } from "../compose.ts";
-import type { OpFn } from "../types.ts";
+import type { Operator } from "../types.ts";
 
 import type {
   Action,
@@ -153,9 +153,13 @@ export function* performanceMonitor<Ctx extends PerfCtx = PerfCtx>(
 /**
  * This middleware will call the `saga` provided with the action sent to the middleware pipeline.
  */
-export function wrap<Ctx extends PipeCtx = PipeCtx>(op: (a: Action) => OpFn) {
+export function wrap<Ctx extends PipeCtx = PipeCtx, T = any>(
+  op: (a: Action) => Operator<T>,
+) {
   return function* (ctx: Ctx, next: Next) {
-    yield* call(() => op(ctx.action));
+    yield* call(function* () {
+      return op(ctx.action);
+    });
     yield* next();
   };
 }

query/pipe.ts

@@ -1,5 +1,5 @@
 import { compose } from "../compose.ts";
-import type { OpFn, Payload } from "../types.ts";
+import type { Operator, Payload } from "../types.ts";
 import { parallel } from "../mod.ts";
 
 // TODO: remove store deps
@@ -24,7 +24,7 @@ import { Ok } from "../deps.ts";
 export interface SagaApi<Ctx extends PipeCtx> {
   use: (fn: Middleware<Ctx>) => void;
   routes: () => Middleware<Ctx>;
-  bootup: OpFn;
+  bootup: Operator<unknown>;
 
   /**
    * Name only
@@ -127,7 +127,7 @@ export function createPipe<Ctx extends PipeCtx = PipeCtx<any>>(
   } = { supervisor: takeEvery },
 ): SagaApi<Ctx> {
   const middleware: Middleware<Ctx>[] = [];
-  const visors: { [key: string]: OpFn } = {};
+  const visors: { [key: string]: Operator<unknown> } = {};
   const middlewareMap: { [key: string]: Middleware<Ctx> } = {};
   const actionMap: {
     [key: string]: CreateActionWithPayload<Ctx, any>;

store/store.ts

@@ -9,7 +9,7 @@ import {
   Task,
 } from "../deps.ts";
 import { BaseMiddleware, compose } from "../compose.ts";
-import type { AnyAction, AnyState, OpFn } from "../types.ts";
+import type { AnyAction, AnyState, Operator } from "../types.ts";
 import { safe } from "../fx/mod.ts";
 import { Next } from "../query/types.ts";
 import type { FxStore, Listener, StoreUpdater, UpdaterCtx } from "./types.ts";
@@ -132,7 +132,7 @@ export function createStore<S extends AnyState>({
     });
   }
 
-  function run<T>(op: OpFn<T>): Task<Result<T>> {
+  function run<T>(op: Operator<T>): Task<Result<T>> {
     return scope.run(function* () {
       return yield* safe(op);
     });

store/supervisor.ts

@@ -1,7 +1,7 @@
 import { call, race } from "../fx/mod.ts";
 import { take } from "./fx.ts";
 import { Operation, sleep, spawn, Task } from "../deps.ts";
-import type { ActionWPayload, AnyAction, OpFn } from "../types.ts";
+import type { ActionWPayload, AnyAction, Operator } from "../types.ts";
 import type { CreateActionPayload } from "../query/mod.ts";
 
 const MS = 1000;
@@ -44,12 +44,14 @@ export function poll(parentTimer: number = 5 * 1000, cancelType?: string) {
 export function timer(timer: number = 5 * MINUTES) {
   return function* onTimer(
     actionType: string,
-    op: (action: AnyAction) => OpFn,
+    op: (action: AnyAction) => Operator<unknown>,
   ) {
     const map: { [key: string]: Task<unknown> } = {};
 
     function* activate(action: ActionWPayload<CreateActionPayload>) {
-      yield* call(() => op(action));
+      yield* call(function* () {
+        return op(action);
+      });
       yield* sleep(timer);
       delete map[action.payload.key];
     }

store/types.ts

@@ -1,6 +1,6 @@
 import type { Operation, Patch, Result, Scope, Task } from "../deps.ts";
 import { BaseCtx } from "../mod.ts";
-import type { AnyAction, AnyState, OpFn } from "../types.ts";
+import type { AnyAction, AnyState, Operator } from "../types.ts";
 
 export type StoreUpdater<S extends AnyState> = (s: S) => S | void;
 
@@ -32,7 +32,7 @@ export interface FxStore<S extends AnyState> {
   getState: () => S;
   subscribe: (fn: Listener) => () => void;
   update: (u: StoreUpdater<S> | StoreUpdater<S>[]) => Operation<UpdaterCtx<S>>;
-  run: <T>(op: OpFn<T>) => Task<Result<T>>;
+  run: <T>(op: Operator<T>) => Task<Result<T>>;
   // deno-lint-ignore no-explicit-any
   dispatch: (a: AnyAction) => any;
   replaceReducer: (r: (s: S, a: AnyAction) => S) => void;

types.ts

@@ -5,10 +5,11 @@ export interface Computation<T = unknown> {
   [Symbol.iterator](): Iterator<Instruction, T, any>;
 }
 
-export type OpFn<T = unknown> =
+export type Operator<T> =
+  | Operation<T>
+  | Promise<T>
   | (() => Operation<T>)
-  | (() => PromiseLike<T>)
-  | (() => T);
+  | (() => Promise<T>);
 
 export interface QueryState {
   "@@starfx/loaders": Record<string, LoaderItemState>;