Function createSubscribeAdapter

packages/ai-client/tests/chat-client.test.ts:275–316

(chunksToSend: Array<StreamChunk>)

  describe('subscribe/send connection mode', () => {
    function createSubscribeAdapter(chunksToSend: Array<StreamChunk>) {
      let hasPendingSend = false
      let wakeSubscriber: (() => void) | null = null
      let removeAbortListener: (() => void) | null = null

      const subscribe = vi.fn((signal?: AbortSignal) => {
        return (async function* () {
          while (!signal?.aborted) {
            if (!hasPendingSend) {
              await new Promise<void>((resolve) => {
                removeAbortListener?.()
                removeAbortListener = null
                wakeSubscriber = resolve
                const onAbort = () => resolve()
                signal?.addEventListener('abort', onAbort, { once: true })
                removeAbortListener = () => {
                  signal?.removeEventListener('abort', onAbort)
                }
              })
              continue
            }

            hasPendingSend = false
            for (const chunk of chunksToSend) {
              yield chunk
            }
          }
          removeAbortListener?.()
          removeAbortListener = null
        })()
      })

      const send = vi.fn(async () => {
        removeAbortListener?.()
        removeAbortListener = null
        hasPendingSend = true
        wakeSubscriber?.()
        wakeSubscriber = null
      })

      return { subscribe, send }
    }

    it('should use subscribe/send adapter mode', async () => {
      const adapter = createSubscribeAdapter(