Anyone help me to implement connection adapter for react native. I have already do it but after tool calling follow up message not appear and isLoading keep true , new message not sending.

[devshahoriar]

import { type StreamChunk } from '@tanstack/ai'
import { type ConnectionAdapter } from '@tanstack/ai-react'
import EventSource from 'react-native-sse'

const isTerminalChunk = (chunk: StreamChunk) => {
console.log(JSON.stringify(chunk,null,2))
if (chunk.type === 'RUN_ERROR') {
return true
}

if (chunk.type === 'RUN_FINISHED') {
return chunk.finishReason !== 'tool_calls'
}

return false
}

export const createReactNativeSseConnection = (
url: string
): ConnectionAdapter => ({
connect(messages, data, abortSignal) {
return (async function* () {
const queue: StreamChunk[] = []
let resolver: ((value: StreamChunk | null) => void) | null = null
let done = false
let error: Error | null = null

  const push = (chunk: StreamChunk) => {
    if (done) return

    if (isTerminalChunk(chunk)) {
      done = true
    }

    if (resolver) {
      const next = resolver
      resolver = null
      next(chunk)
      return
    }

    queue.push(chunk)
  }

  const finish = () => {
    done = true
    if (resolver) {
      const next = resolver
      resolver = null
      next(null)
    }
  }

  const fail = (nextError: Error) => {
    error = nextError
    finish()
  }

  const es = new EventSource(url, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      messages,
      data,
      conversationId:
        typeof data?.conversationId === 'string'
          ? data.conversationId
          : undefined,
    }),
    lineEndingCharacter: '\n',
    pollingInterval: 0,
    
  })

  const onAbort = () => {
    es.close()
    finish()
  }

  abortSignal?.addEventListener('abort', onAbort, { once: true })

  es.addEventListener('message', (event) => {
    const payload = event.data
    if (!payload) return

    if (typeof payload === 'string' && payload.trim() === '[DONE]') {
      es.close()
      finish()
      return
    }

    try {
      const chunk = JSON.parse(payload) as StreamChunk
      push(chunk)
    } catch {
      // Ignore non-JSON chunks.
    }
  })

  es.addEventListener('error', (event) => {
    const message =
      event.type === 'error' ? 'SSE stream failed' : 'SSE stream error'
    es.close()
    fail(new Error(message))
  })

  es.addEventListener('close', () => {
    es.close()
    finish()
  })

  try {
    while (!done || queue.length > 0) {
      if (error) {
        throw error
      }

      if (queue.length > 0) {
        const nextChunk = queue.shift()
        if (nextChunk) {
          yield nextChunk
        }
        continue
      }

      const nextChunk = await new Promise<StreamChunk | null>((resolve) => {
        resolver = resolve
      })

      if (nextChunk) {
        yield nextChunk
      }
    }

    if (error) {
      throw error
    }
  } finally {
    abortSignal?.removeEventListener('abort', onAbort)
    es.removeAllEventListeners()
    es.close()
  }
})()

},
})

This is my code 👆

[devshahoriar]

after change and save it start working

2026-04-28.17-24-23.mp4

[AlemTuzlak]

We've added support and docs for reaxt native!