Cheat to win by setting a match context that avoids invoking the continuation

match-signal.ts

import {
  createContext,
  type Queue,
  resource,
  type Signal,
  type Stream,
} from "https://deno.land/x/effection@3.0.0-beta.0/mod.ts";

type Predicate = (value: unknown) => boolean;

const MatchContext = createContext<Predicate>("match-filter");

import type { Resolve } from "./types.ts";
import { pause } from "./pause.ts";

export function createMatchSignal<T, TClose = never>(): Signal<T, TClose> {
  let subscribers = new Set<Queue<T, TClose>>();

  let stream: Stream<T, TClose> = resource(function* Subscription(provide) {
    let queue = createMatchQueue<T, TClose>();
    subscribers.add(queue);

    try {
      yield* provide(queue.subscription);
    } finally {
      subscribers.delete(queue);
    }
  });

  function send(value: T) {
    for (let queue of [...subscribers]) {
      queue.add(value);
    }
  }

  function close(value: TClose) {
    for (let queue of [...subscribers]) {
      queue.close(value);
    }
  }

  return { send, close, stream };
}

export function createMatchQueue<T, TClose>(): Queue<T, TClose> {
  type Item = IteratorResult<T, TClose>;

  let items: Item[] = [];
  let consumers = new Set<Resolve<Item>>();

  function enqueue(item: Item) {
    items.unshift(item);
    while (items.length > 0 && consumers.size > 0) {
      let [consume] = consumers;
      let top = items.pop() as Item;
      consume(top);
    }
  }

  return {
    add: (value) => enqueue({ done: false, value }),
    close: (value) => enqueue({ done: true, value }),
    subscription: {
      *next() {
        let item = items.pop();
        if (item) {
          return item;
        } else {
          let predicate = yield* MatchContext;
          return yield* pause<Item>((resolve) => {
            let resolveIfMatches = (item: Item) => {
              if (item.done || predicate(item.value)) {
                resolve(item);
              }
            };
            consumers.add(resolveIfMatches);
            return () => consumers.delete(resolveIfMatches);
          });
        }
      },
    },
  };
}