refactor(redux): use signal instead of channel

deps.ts

@@ -5,6 +5,7 @@ export type {
   Port,
   Result,
   Scope,
+  Signal,
   Stream,
   Subscription,
   Task,
@@ -13,7 +14,9 @@ export {
   action,
   createChannel,
   createContext,
+  createQueue,
   createScope,
+  createSignal,
   each,
   Err,
   expect,

mod.ts

@@ -7,7 +7,9 @@ export {
   action,
   createChannel,
   createContext,
+  createQueue,
   createScope,
+  createSignal,
   each,
   Err,
   getframe,

redux/fx.ts

@@ -1,52 +1,48 @@
-import { Action, Channel, Operation } from "../deps.ts";
-import { createChannel, createContext, spawn } from "../deps.ts";
-import { call, parallel } from "../fx/mod.ts";
+import { Action, Operation, Signal } from "../deps.ts";
+import { createContext, createSignal, each, spawn } from "../deps.ts";
+import { call } from "../fx/mod.ts";
 import { ActionPattern, matcher } from "../matcher.ts";
 import type { ActionWPayload, AnyAction } from "../types.ts";
 
 import type { StoreLike } from "./types.ts";
 
-export const ActionContext = createContext<Channel<Action, void>>(
+export const ActionContext = createContext<Signal<Action, void>>(
   "redux:action",
-  createChannel<Action, void>(),
+  createSignal<Action, void>(),
 );
 
 export const StoreContext = createContext<StoreLike>("redux:store");
 
-export function* emit({
-  channel,
+export function emit({
+  signal,
   action,
 }: {
-  channel: Operation<Channel<AnyAction, void>>;
+  signal: Signal<AnyAction, void>;
   action: AnyAction | AnyAction[];
 }) {
-  const { input } = yield* channel;
   if (Array.isArray(action)) {
     if (action.length === 0) {
       return;
     }
-    yield* parallel(action.map((a) => () => input.send(a)));
+    action.map((a) => signal.send(a));
   } else {
-    yield* input.send(action);
+    signal.send(action);
   }
 }
 
 export function* once({
-  channel,
+  signal,
   pattern,
 }: {
-  channel: Operation<Channel<Action, void>>;
+  signal: Signal<Action, void>;
   pattern: ActionPattern;
 }) {
-  const { output } = yield* channel;
-  const msgList = yield* output;
-  let next = yield* msgList.next();
-  while (!next.done) {
+  for (const action of yield* each(signal.stream)) {
     const match = matcher(pattern);
-    if (match(next.value)) {
-      return next.value;
+    if (match(action)) {
+      return action;
     }
-    next = yield* msgList.next();
+    yield* each.next;
   }
 }
 
@@ -57,8 +53,9 @@ export function* select<S, R>(selectorFn: (s: S) => R) {
 
 export function take<P>(pattern: ActionPattern): Operation<ActionWPayload<P>>;
 export function* take(pattern: ActionPattern): Operation<Action> {
+  const signal = yield* ActionContext;
   const action = yield* once({
-    channel: ActionContext,
+    signal,
     pattern,
   });
   return action as Action;

redux/middleware.ts

@@ -1,36 +1,21 @@
-import {
-  Action,
-  BATCH,
-  BatchAction,
-  ReducersMapObject,
-  Scope,
-} from "../deps.ts";
+import { BATCH, BatchAction, ReducersMapObject, Scope } from "../deps.ts";
 import { combineReducers, createScope, enableBatching } from "../deps.ts";
-import { parallel } from "../fx/mod.ts";
 import type { AnyAction } from "../types.ts";
-
 import { ActionContext, emit, StoreContext } from "./fx.ts";
 import { reducers as queryReducers } from "./query.ts";
 import type { StoreLike } from "./types.ts";
 
 function* send(action: AnyAction) {
+  const signal = yield* ActionContext;
   if (action.type === BATCH) {
     const actions = action.payload as BatchAction[];
-    const group = yield* parallel(
-      actions.map(
-        (a) =>
-          function* () {
-            yield* emit({
-              channel: ActionContext,
-              action: a as Action,
-            });
-          },
-      ),
-    );
-    yield* group;
+    emit({
+      signal,
+      action: actions,
+    });
   } else {
-    yield* emit({
-      channel: ActionContext,
+    emit({
+      signal,
       action,
     });
   }

redux/put.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "../test.ts";
-import { sleep, spawn } from "../deps.ts";
+import { each, sleep, spawn } from "../deps.ts";
 
 import { ActionContext, put, take } from "./mod.ts";
 import { createTestStore } from "./util.ts";
@@ -12,11 +12,9 @@ it(putTests, "should send actions through channel", async () => {
   function* genFn(arg: string) {
     yield* spawn(function* () {
       const actions = yield* ActionContext;
-      const msgs = yield* actions.output;
-      let action = yield* msgs.next();
-      while (!action.done) {
-        actual.push(action.value.type);
-        action = yield* msgs.next();
+      for (const action of yield* each(actions.stream)) {
+        actual.push(action.type);
+        yield* each.next;
       }
     });