Is it possible to create an SSE handler without a generator?

[szagi3891]

Describe the feature

    class SourceOfEvents {
        async process(id: string, emit: (event) => void) {
            for (let i=0; i<100; i++) {
                emit({
                    message: `event id=${id} count: ${i}`
                });

                await timeout(1000);
            }
        }
    }

    const sourceOfEvents = new SourceOfEvents();

    test: os
        .input(z.obiect({
            id: z.string(),
        }))
        .handler(async ({ input, emit }) => {
            await sourceOfEvents.process(input.id, emit);
            //end of stream SSE
        })

I would like to be able to create an SSE handler without generators.
In my application, I would like to have an object or function that allows sending new events directly to an SSE stream (without generators).

Additional information

• Would you be willing to help implement this feature?

[coderabbitai]

📝 CodeRabbit Plan Mode

Generate an implementation plan and prompts that you can use with your favorite coding agent.

• Create Plan

Examples

• Example 1
• Example 2

---

🔗 Similar Issues

Related Issues

• https://github.com/unnoq/orpc/issues/1058
• https://github.com/unnoq/orpc/issues/993

👤 Suggested Assignees

• unnoq
• marbemac

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord or schedule a call!

[dosubot]

Hi @szagi3891! I'm Dosu and I’m helping the orpc team.

orpc currently only supports SSE handlers using async generators—there's no built-in way to create an SSE handler that receives an emit callback or lets you push events directly from an object or function without a generator. This generator-based pattern is used for portability and resource management across all supported environments. While the Publisher and EventPublisher helpers let you decouple event production (so you can publish events from anywhere in your code), the handler that streams events to the client still needs to be an async generator that yields those events as they arrive. The callback-based subscription in Publisher is for internal event handling, not for SSE handler APIs. There are discussions about supporting alternative streaming APIs, but for now, generator-based handlers are the only portable and supported approach in orpc (reference, reference, reference).

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[dinwwwh]

An Event Iterator is simply a return value that satisfies the AsyncIteratorObject interface. You can implement this by returning an async generator directly:

const procedure = os
  .input(z.object({
    id: z.string(),
  }))
  .handler(async ({ input, emit }) => {
    return (async function* () {
      yield 'something';
    })();
  });

Alternatively, for more complex logic or state management, you can return an instance of a dedicated class, such as AsyncIteratorClass.

[jedwards1211]

It would make more sense to support returning async iterables (anything with a Symbol.asyncIterator method). Then returning a ReadableStream of events would work. The web streams API is a lot more effective than basic async iteration for efficiently transforming sequential content. Each link in the chain can prefetch chunks if desired so there's less of a bottleneck, and it's harder to leak resources AFAIK when mapping/filtering with web streams than with async generators.

In graphql-js for example, subscription resolvers must return async iterables, not iterators. A lot of people in the JS world misunderstand when to use an iterable or an iterator, because a generator returns an iterable iterator that satisfies both contracts.

[dinwwwh]

It would make more sense to support returning async iterables (anything with a Symbol.asyncIterator method)

You can easily convert an async iterable into an async iterator object (which includes the [Symbol.asyncIterator] and next methods). We have opted not to support generic iterables because our client-side implementation specifically expects an async iterator.

Returning a ReadableStream on server while the client re-create an async iterator object could lead to unpredictable behavior. Furthermore, we want to avoid complex TypeScript transformations (changing Type A to Type B) solely for the sake of performance, as this adds unnecessary architectural overhead.

[jedwards1211]

So a client can't iterate with for await then? Or is the return type actually an AsyncIterableIterator? (Afaik AsyncIterator can't be for awaited, only AsyncIterable and derived interfaces)

Returning a ReadableStream on server while the client re-create an async iterator object could lead to unpredictable behavior.

Well one would want to declare a less specific return type like AsyncIterable<T> that's valid for the client too. I could use a ReadableStream under the hood even with the existing design, it would just take more boilerplate

[dinwwwh]

@jedwards1211 async iterator object is an async iterator & async iteratable but does not have method of ReadableStream

[jedwards1211]

I was never saying the return type visible to the client should be a ReadableStream. Just that the runtime characteristics of an async generator aren't always ideal. If the runtime code of the clients only relied on the Symbol.asyncIterator method and allowed the return type to be only AsyncIterable<T> (which it already is a more specific subtype of) then we wouldn't have to write an adapter layer to add the other methods of an async generator return value.

[jedwards1211]

I'll try to resurrect an example soon of how async generators tend to cause subtle memory leaks when you're filtering or a mapping an iterable of redis events. It's something that I stumbled on long ago with graphql subscriptions and found a discussion thread in graphql about, but that few people realize.

[dinwwwh]

async generators tend to cause subtle memory leaks when you're filtering or a mapping an iterable of redis events.

I don't think they’re related. An async iterator / async generator is just a protocol (a way to communicate). If there's a memory leak, it's likely due to an implementation issue somewhere, not the protocol itself.

[jedwards1211]

@unnoq they are related. This is hard to realize and easy to forget, I will demonstrate. It's kind of horrifying, kind of felt like the ground dropping out from beneath me when I realized it while debugging memory leaks.

A common use for RPC subscriptions is subscribing to pubsub events from Redis, Postgres, etc. Yielding them from an async iterator properly requires some kind of push/pull queue. But for the sake of demonstration let's make a simple mock where an event arrives every 10 seconds:

let start = Date.now()

function log(...args) {
  console.log(`+${((Date.now() - start) / 1000).toFixed(3).padStart(6)}s`, ...args)
}

function events() {
  let i = 0
  let start = Date.now()
  return {
    async next() {
      const index = i++
      log('next() called', index)
      await new Promise(r => setTimeout(r, Math.max(0, start + 10000 * index - Date.now())))
      log('mock pubsub event', index)
      return {value: index, done: false}
    },
    async return() {
      log('mock unsubscribing from pubsub')
      return {value: undefined, done: true}
    },
    async throw() {
      return {value: undefined, done: true}
    },
    [Symbol.asyncIterator]() {
      log('mock subscribing to pubsub')
      return this
    }
  }
}

Scenario: A client subscribes, receives one event, and then unsubscribes while the framework is waiting for the second event. The RPC framework return()s the iterator to make it clean up:

console.log('Basic Events')
console.log('=================')
const iterator = events()[Symbol.asyncIterator]()
log(await iterator.next())
iterator.next().then(log, log)
log('client ending subscription')
await iterator.return()
log('done cleaning up')

This works fine, everything is cleaned up immediately when the client unsubscribes:

Basic Events
=================
+ 0.011s mock subscribing to pubsub
+ 0.011s next() called 0
+ 0.013s mock pubsub event 0
+ 0.013s { value: 0, done: false }
+ 0.013s next() called 1
+ 0.014s client ending subscription
+ 0.014s mock unsubscribing from pubsub
+ 0.014s done cleaning up

But now look what happens when we involve an async generator. Let's say we use one to transform the events from pubsub:

async function * events2() {
  try {
    for await (const event of events()) {
      log('events2 got', event)
      yield event * 2
    }
  } finally {
    log('events2 returning')
  }
}

console.log('Async Generator')
console.log('=================')
const iterator = events2()[Symbol.asyncIterator]()
log(await iterator.next())
iterator.next().then(log, log)
log('client ending subscription')
await iterator.return()
log('done cleaning up')

This time, nothing gets cleaned up until after the second event arrives from pubsub 10s after the client unsubscribes (but in the real world who knows how long it will take):

Async Generator
=================
+ 0.024s mock subscribing to pubsub
+ 0.024s next() called 0
+ 0.025s mock pubsub event 0
+ 0.025s events2 got 0
+ 0.025s { value: 0, done: false }
+ 0.026s next() called 1
+ 0.026s client ending subscription
+10.025s mock pubsub event 1
+10.026s events2 got 1
+10.026s { value: 2, done: false }
+10.026s mock unsubscribing from pubsub
+10.026s events2 returning
+10.026s done cleaning up

The framework could alternatively throw() the iterator, but the delay is the same.

Important

The return() (or throw()) call will hang for as long as it takes the next pubsub event to arrive, potentially forever. And the Promise returned by return() leaks memory as long as it is outstanding. This is because async generators wait for the previous next() promise to fulfill before finishing.

For this reason, servers must avoid using async generators anywhere unpredictable events are involved to avoid eventual OOM.

One workaround is to do mapping and filtering in manual iterator implementations:

function events3() {
  const underlying = events()
  return {
    async next() {
      const { value, done } = await underlying.next()
      log('events3 got', value)
      return { value: value != null ? value * 2 : undefined, done }
    },
    return() {
      log('events3 returning')
      return underlying.return()
    },
    throw(e) {
      return underlying.throw(e)
    },
    [Symbol.asyncIterator]() {
      return this
    }
  }
}

Manual Iteration
=================
+ 0.012s next() called 0
+ 0.013s mock pubsub event 0
+ 0.014s events3 got 0
+ 0.014s { value: 0, done: false }
+ 0.014s next() called 1
+ 0.014s client ending subscription
+ 0.015s events3 returning
+ 0.015s mock unsubscribing from pubsub
+ 0.015s done cleaning up

Another general alternative to async iteration is to use web streams, because you can do mapping and filtering in a TransformStream and stream.cancel() immediately cancels the entire chain instead of waiting for the next chunk. When I can, I prefer to use web streams for working with unpredictable event sequences because they have fewer pitfalls.

However, I did discover in the course of this exercise that iterating web streams (via stream[Symbol.asyncIterator]()) unfortunately has the same problem as async generators: return() waits for the previous next() to resolve instead of canceling the chain of streams immediately. So returning a web stream directly from a subscription handler wouldn't be the solution I thought it would (unless the framework called iterator.cancel?.() in case it's a stream).

Otherwise, the safest way for me to work with pubsub subscriptions with oRPC is to use chains of web streams and then wrap them with a carefully written AsyncIterableIterator that cancel()s the stream immediately when it's return()ed.

Warning

I haven't checked if it does, but if oRPC itself iterates over our subscriptions in any async generators, we will be unable to prevent leaks like this with any userland workarounds.

[dinwwwh]

@jedwards1211 First of all, thanks for taking the time to write such a detailed message and for providing example code.

As I mentioned earlier, this issue is not related to the async iterator protocol itself. If there is a memory leak, it comes from the implementation, not from the protocol.

For example, why not do something like this? (we must rely on signal to deal with this problem)

async function* events2() {
  const eventsR = events()
  try {
    for await (const event of eventsR) {
      log('events2 got', event)
      yield event * 2
    }
  } finally {
    await eventsR.return()
    log('events2 returning')
  }
}

The only real downside here is that this syntax makes it easier for developers to forget the cleanup logic if they are not careful.

Now, about an async function that potentially never resolves or rejects: when you invoke such a function, it may wait forever, regardless of whether it is triggered from an async generator, a ReadableStream, or anything else. That is exactly why AbortSignal exists. oRPC already fully supports it, and you can even access it inside .handler.

You also mentioned preferring ReadableStream | TransformStream over async generators. The main reason I see for that preference is that streams make cleanup logic easier to manage, since it is handled automatically for you.

To be clear, I am not arguing that async generators or iterators should be used everywhere. My point is simply that this syntax itself does not cause memory leaks. Any leak would come from how it is implemented. Also, it is straightforward to write a small helper to transform a ReadableStream into an async generator or async iterator and reuse it across the project.