repos / starfx

action.ts

  1import {
  2  type Callable,
  3  type Operation,
  4  type Signal,
  5  SignalQueueFactory,
  6  type Stream,
  7  call,
  8  createContext,
  9  createSignal,
 10  each,
 11  spawn,
 12} from "effection";
 13import { type ActionPattern, matcher } from "./matcher.js";
 14import { createFilterQueue } from "./queue.js";
 15import type { Action, ActionWithPayload, AnyAction } from "./types.js";
 16import type { ActionFnWithPayload } from "./types.js";
 17
 18export const ActionContext = createContext(
 19  "starfx:action",
 20  createSignal<AnyAction, void>(),
 21);
 22
 23export function useActions(pattern: ActionPattern): Stream<AnyAction, void> {
 24  return {
 25    [Symbol.iterator]: function* () {
 26      const actions = yield* ActionContext.expect();
 27      const match = matcher(pattern);
 28      yield* SignalQueueFactory.set(() => createFilterQueue(match) as any);
 29      return yield* actions;
 30    },
 31  };
 32}
 33
 34export function emit({
 35  signal,
 36  action,
 37}: {
 38  signal: Signal<AnyAction, void>;
 39  action: AnyAction | AnyAction[];
 40}) {
 41  if (Array.isArray(action)) {
 42    if (action.length === 0) {
 43      return;
 44    }
 45    action.map((a) => signal.send(a));
 46  } else {
 47    signal.send(action);
 48  }
 49}
 50
 51export function* put(action: AnyAction | AnyAction[]) {
 52  const signal = yield* ActionContext.expect();
 53  return emit({
 54    signal,
 55    action,
 56  });
 57}
 58
 59export function take<P>(
 60  pattern: ActionPattern,
 61): Operation<ActionWithPayload<P>>;
 62export function* take(pattern: ActionPattern): Operation<Action> {
 63  const fd = useActions(pattern);
 64  for (const action of yield* each(fd)) {
 65    return action;
 66  }
 67  return { type: "take failed, this should not be possible" };
 68}
 69
 70export function* takeEvery<T>(
 71  pattern: ActionPattern,
 72  op: (action: AnyAction) => Operation<T>,
 73) {
 74  const fd = useActions(pattern);
 75  for (const action of yield* each(fd)) {
 76    yield* spawn(() => op(action));
 77    yield* each.next();
 78  }
 79}
 80
 81export function* takeLatest<T>(
 82  pattern: ActionPattern,
 83  op: (action: AnyAction) => Operation<T>,
 84) {
 85  const fd = useActions(pattern);
 86  let lastTask;
 87
 88  for (const action of yield* each(fd)) {
 89    if (lastTask) {
 90      yield* lastTask.halt();
 91    }
 92    lastTask = yield* spawn(() => op(action));
 93    yield* each.next();
 94  }
 95}
 96
 97export function* takeLeading<T>(
 98  pattern: ActionPattern,
 99  op: (action: AnyAction) => Operation<T>,
100) {
101  while (true) {
102    const action = yield* take(pattern);
103    yield* call(() => op(action));
104  }
105}
106
107export function* waitFor(predicate: Callable<boolean>) {
108  const init = yield* call(predicate as any);
109  if (init) {
110    return;
111  }
112
113  while (true) {
114    yield* take("*");
115    const result = yield* call(predicate as any);
116    if (result) {
117      return;
118    }
119  }
120}
121
122export function getIdFromAction(
123  action: ActionWithPayload<{ key: string }> | ActionFnWithPayload,
124): string {
125  return typeof action === "function" ? action.toString() : action.payload.key;
126}
127
128export const API_ACTION_PREFIX = "";
129
130export function createAction(actionType: string): () => Action;
131export function createAction<P>(
132  actionType: string,
133): (p: P) => ActionWithPayload<P>;
134export function createAction(actionType: string) {
135  if (!actionType) {
136    throw new Error("createAction requires non-empty string");
137  }
138  const fn = (payload?: unknown) => ({
139    type: actionType,
140    payload,
141  });
142  fn.toString = () => actionType;
143  Object.defineProperty(fn, "_starfx", {
144    value: true,
145    enumerable: false,
146    configurable: false,
147    writable: false,
148  });
149  return fn;
150}