Will there be support for signals for `fromEvent` and `fromEventPattern`?

[seniorkonung]

Currently, event binding is not sensitive to AbortSignal at all.

[trxcllnt]

What's wrong with using withAbort(), e.g. fromEvent(tgt, "evt").pipe(withAbort(sig))?

[seniorkonung]

What's wrong with using withAbort(), e.g. fromEvent(tgt, "evt").pipe(withAbort(sig))?

Is your signal with fromEvent working well? Actually, it's my assumption that it doesn't work, because I saw in the source code that it's based on fromEventPattern.

My problem is exactly with fromEventPattern.

import { fromEventPattern } from "ix/asynciterable"

const controller = new AbortController
setTimeout(() => {
    console.log("controller.abort()")
    controller.abort()
}, 8000)

let intervalId
fromEventPattern((handler) => {
    let count = 0
    intervalId = setInterval(() => handler(count++), 1000)
}, (_) => {
    console.log("clearInterval")
    clearInterval(intervalId)
})
.forEach(console.log, undefined, controller.signal)

Output:

0 0 AbortSignal { aborted: false }
1 1 AbortSignal { aborted: false }
2 2 AbortSignal { aborted: false }
3 3 AbortSignal { aborted: false }
4 4 AbortSignal { aborted: false }
5 5 AbortSignal { aborted: false }
6 6 AbortSignal { aborted: false }
controller.abort()
7 7 AbortSignal { aborted: true }
8 8 AbortSignal { aborted: true }
9 9 AbortSignal { aborted: true }
10 10 AbortSignal { aborted: true }
11 11 AbortSignal { aborted: true }
12 12 AbortSignal { aborted: true }
13 13 AbortSignal { aborted: true }

Expected behavior: the removeHandler closure fromEventPattern should be triggered when the signal is triggered. This does not happen.

I tried to do my own mini-investigation and found that the memoize operator does not pass the signal upstream. And it is memoize that the fromEventPattern operator is based on. This prevented me from writing my quick implementation of fromEventPattern, also based on memoize. But the signal was passed to the operator I created if I didn't use memoize.

But I was doing my mini-investigation at the time of publishing this issue and I'm speaking from memory. Besides, perhaps I didn't look deep enough into the matter, so these conclusions should be treated with skepticism.

There is only one fact: the fromEventPattern behavior does not match the user's expectations and does not cancel its behavior when the signal is triggered.

Platform: node v22.12.0
Version ix: 7.0.0

[trxcllnt]

Oh right, I forgot fromEventPattern uses memoize. The correct fix is to make memoize aggregate all its subscriber's cancellations and cancel upstream when they're all done. @mattpodwysocki

[ghbuzer]

To check whether I face the same issue. The following code should normally stop printing after 5 seconds, but due to this issue it does not. Is this correct?

const emitter = new EventTarget();
const abortController = new AbortController();
let asyncIterable = fromEvent(emitter, 'my-event').pipe(withAbort(abortController.signal)).pipe(tap(console.log))
setInterval(()=>emitter.dispatchEvent(new CustomEvent('my-event', {detail: { foo: 'bar' }})), 1000)
setTimeout(()=>abortController.abort(), 5000);
toArray(asyncIterable);

Also FYI: this is the workaround I use for now, but it seems to only work with rxjs. In IxJs I get: this._other is not a function at TakeUntilAsyncIterable._a

const emitter = new EventTarget();
let o = fromEvent(emitter, 'my-event').pipe(tap(console.log)).pipe(takeUntil(fromEvent(emitter, 'abort')))
let interval = setInterval(()=>emitter.dispatchEvent(new CustomEvent('my-event', {detail: { foo: 'bar' }})), 1000)
setTimeout(()=>(console.log("aborting"), emitter.dispatchEvent(new CustomEvent('abort', {detail: {}}))), 5000);
o.subscribe()

[bilby91]

I'm facing a similar issue. I would have expected this to call the remove handler but it's not happening.

async consume(options: ConsumeOptions): Promise<AsyncIterable<DomainEvent>> {
    const opId = crypto.randomUUID();
    const queueName = `ephemeral-${opId}`;
    const consumerTag = `consumer-${opId}`;

    try {
      this.logger.debug(`Creating ephemeral queue: ${queueName}`);

      // Assert ephemeral queue (auto-delete when consumer disconnects)
      await this.amqpConnection.channel.assertQueue(queueName, {
        durable: false,
        autoDelete: true,
        exclusive: true,
        arguments: options.queueArguments,
      });

      // Bind the queue to routing patterns
      for (const pattern of options.routingPatterns) {
        await this.amqpConnection.channel.bindQueue(
          queueName,
          EVENT_EXCHANGE_NAME,
          pattern,
        );
        this.logger.debug(`Bound queue ${queueName} to pattern: ${pattern}`);
      }

      const messageStream = fromEventPattern<DomainEvent>(
        (handler) =>
          this.amqpConnection.channel.consume(queueName, handler, {
            noAck: true,
            consumerTag: consumerTag,
          }),
        () => {
          this.amqpConnection.channel
            .cancel(consumerTag)
            .then(() => {
              this.logger.debug(
                `Consumer ${consumerTag} cancelled successfully`,
              );
            })
            .catch((err) => {
              this.logger.warn(
                `Failed to cancel consumer ${consumerTag} during cleanup`,
                err,
              );
            });
        },
        (message: ConsumeMessage) => {
          return JSON.parse(message.content.toString()) as DomainEvent;
        },
      );

      return from(wrapWithAbort(messageStream, options.signal));
}

[trxcllnt]

Also FYI: this is the workaround I use for now, but it seems to only work with rxjs. In IxJs I get: this._other is not a function at TakeUntilAsyncIterable._a

The takeUntil() operator accepts a function that returns a Promise, so you'd need to do something like this:

let o = fromEvent(emitter, 'my-event')
  .pipe(tap(console.log))
  .pipe(takeUntil(() => first(fromEvent(emitter, 'abort'))))

[trxcllnt]

I'm pretty swamped at the moment, but I'll try to take a look at implementing this in the next few weeks. Or If someone wants to file a PR before me, I can review it. Like I mentioned above, the fix is to add cancellation to memoize, which would be a nice feature to have regardless.