queue.ts

@@ -0,0 +1,14 @@
+import { createQueue } from "./deps.ts";
+
+export function createFilterQueue<T, TClose>(predicate: (v: T) => boolean) {
+  const queue = createQueue<T, TClose>();
+
+  return {
+    ...queue,
+    add(value: T) {
+      if (predicate(value)) {
+        queue.add(value);
+      }
+    },
+  };
+}

redux/fx.ts

@@ -9,7 +9,6 @@ import {
 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<Signal<Action, void>>(
@@ -63,7 +62,7 @@ export function* select<S, R>(selectorFn: (s: S) => R) {
   return selectorFn(store.getState() as S);
 }
 
-export function useActions(pattern: ActionPattern): Stream<AnyAction, void> {
+function useActions(pattern: ActionPattern): Stream<AnyAction, void> {
   return {
     *subscribe() {
       const actions = yield* ActionContext;
@@ -123,10 +122,9 @@ export function* takeLeading<T>(
   op: (action: Action) => Operation<T>,
 ) {
   return yield* spawn(function* (): Operation<void> {
-    const fd = useActions(pattern);
-    for (const action of yield* each(fd)) {
+    while (true) {
+      const action = yield* take(pattern);
       yield* call(() => op(action));
-      yield* each.next();
     }
   });
 }

store/fx.ts

@@ -1,6 +1,5 @@
 import {
   Action,
-  createQueue,
   each,
   Operation,
   Signal,
@@ -13,6 +12,7 @@ import { ActionPattern, matcher } from "../matcher.ts";
 import type { ActionWPayload, AnyAction, AnyState } from "../types.ts";
 import type { FxStore, StoreUpdater, UpdaterCtx } from "./types.ts";
 import { ActionContext, StoreContext } from "./context.ts";
+import { createFilterQueue } from "../queue.ts";
 
 export function* updateStore<S extends AnyState>(
   updater: StoreUpdater<S> | StoreUpdater<S>[],
@@ -62,20 +62,7 @@ export function* put(action: AnyAction | AnyAction[]) {
   });
 }
 
-function createFilterQueue<T, TClose>(predicate: (v: T) => boolean) {
-  const queue = createQueue<T, TClose>();
-
-  return {
-    ...queue,
-    add(value: T) {
-      if (predicate(value)) {
-        queue.add(value);
-      }
-    },
-  };
-}
-
-export function useActions(pattern: ActionPattern): Stream<AnyAction, void> {
+function useActions(pattern: ActionPattern): Stream<AnyAction, void> {
   return {
     *subscribe() {
       const actions = yield* ActionContext;