Use a stream combinator to filter the actions from a channel and return a stream that is only

take-filter.ts

import { filter } from "effection";

export function* useActions(pattern: ActionPattern): Stream<AnyAction> {
  let match = matcher(pattern);
  let { output } = yield* ActionContext;
  
  // return a subscription to the filtered actions.
  return yield* filter(match)(output);
}

export function* take(pattern: ActionPattern): Operation<AnyAction> {
  let actions = yield* useActions(pattern);
  let first = yield* actions.next();
  return first.value as AnyAction;
}

export function takeEvery<T>(
  pattern: ActionPattern,
  op: (action: AnyAction) => Operation<T>,
): Operation<Task<void>> {
  return spawn(function*() {
    let actions = yield* useActions(pattern);
    let next = yield* actions.next();
    while (!next.done) {
      yield* spawn(() => op(next.value));
      next = yield* actions.next();
    }
  });
}