(S2S sample-codes/websocket-nodejs) createSessionAsyncIterable does not work with native async iterable with nodenext

[pr1ntr]

Presently this code

compiles down to this with the current tsconfig settings.

with my nodenext settings it compiles like this

So since newer node (Node16 for instance) supports doing this natively. When compiled to node16 or nodenext it doesn't use that generator polyfill and just hangs forever.

@prettyprettyprettygood

[prettyprettyprettygood]

Thanks for reaching out @pr1ntr
Here is the suggested fix for below compile settings
"target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext",

private createSessionAsyncIterable(sessionId: string): AsyncIterable<InvokeModelWithBidirectionalStreamInput> {

    if (!this.isSessionActive(sessionId)) {
      console.log(`Cannot create async iterable: Session ${sessionId} not active`);
      return {
        [Symbol.asyncIterator]() {
          return {
            async next() {
              return { value: undefined, done: true };
            }
          };
        }
      };
    }

    const session = this.activeSessions.get(sessionId);
    if (!session) {
      throw new Error(`Cannot create async iterable: Session ${sessionId} not found`);
    }

    let eventCount = 0;
    const self = this; // Store reference to 'this' for use in the iterator

    return {
      [Symbol.asyncIterator]() {
        console.log(`AsyncIterable iterator requested for session ${sessionId}`);

        return {
          async next(): Promise<IteratorResult<InvokeModelWithBidirectionalStreamInput>> {
            try {
              // Check if session is still active
              if (!session.isActive || !self.activeSessions.has(sessionId)) {
                console.log(`Iterator closing for session ${sessionId}, done = true`);
                return { value: undefined, done: true };
              }
              // Wait for items in the queue or close signal
              if (session.queue.length === 0) {
                try {
                  await Promise.race([
                    firstValueFrom(session.queueSignal.pipe(take(1))),
                    firstValueFrom(session.closeSignal.pipe(take(1))).then(() => {
                      throw new Error("Stream closed");
                    })
                  ]);
                } catch (error) {
                  if (error instanceof Error) {
                    if (error.message === "Stream closed" || !session.isActive) {
                      // This is an expected condition when closing the session
                      if (self.activeSessions.has(sessionId)) {
                        console.log(`Session ${sessionId} closed during wait`);
                      }
                      return { value: undefined, done: true };
                    }
                  }
                  else {
                    console.error(`Error on event close`, error)
                  }
                }
              }

              // If queue is still empty or session is inactive, we're done
              if (session.queue.length === 0 || !session.isActive) {
                console.log(`Queue empty or session inactive: ${sessionId} `);
                return { value: undefined, done: true };
              }

              // Get next item from the session's queue
              const nextEvent = session.queue.shift();
              eventCount++;

              //console.log(`Sending event #${ eventCount } for session ${ sessionId }: ${ JSON.stringify(nextEvent).substring(0, 100) }...`);

              return {
                value: {
                  chunk: {
                    bytes: new TextEncoder().encode(JSON.stringify(nextEvent))
                  }
                },
                done: false
              };
            } catch (error) {
              console.error(`Error in session ${sessionId} iterator: `, error);
              session.isActive = false;
              return { value: undefined, done: true };
            }
          },

          async return(): Promise<IteratorResult<InvokeModelWithBidirectionalStreamInput>> {
            console.log(`Iterator return () called for session ${sessionId}`);
            session.isActive = false;
            return { value: undefined, done: true };
          },

          throw: async (error: any): Promise<IteratorResult<InvokeModelWithBidirectionalStreamInput>> => {
            console.log(`Iterator throw () called for session ${sessionId} with error: `, error);
            session.isActive = false;
            throw error;
          }
        };
      }
    };
  }