Server-Sent Events (SSE)

const stream = await ofetch('/api/sse', { responseType: 'stream' })
const reader = stream.getReader()
const decoder = new TextDecoder()

while (true) {
  const { done, value } = await reader.read()
  if (done) break
  
  const text = decoder.decode(value)
  console.log('Received:', text)
}

case "stream": {
  context.response._data =
    context.response.body || (context.response as any)._bodyInit;
  break;
}

export interface ResponseMap {
  blob: Blob;
  text: string;
  arrayBuffer: ArrayBuffer;
  stream: ReadableStream<Uint8Array>;
}

export type ResponseType = keyof ResponseMap | "json";

export type MappedResponseType<
  R extends ResponseType,
  JsonType = any,
> = R extends keyof ResponseMap ? ResponseMap[R] : JsonType;

const stream = await ofetch("/sse");
const reader = stream.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  // Here is the chunked text of the SSE response.
  const text = decoder.decode(value);
}

const stream = await ofetch('/api/notifications', {
  responseType: 'stream'
})

const reader = stream.getReader()
const decoder = new TextDecoder()

while (true) {
  const { done, value } = await reader.read()
  if (done) break
  
  const chunk = decoder.decode(value)
  const lines = chunk.split('\n')
  
  for (const line of lines) {
    if (line.startsWith('data: ')) {
      const data = JSON.parse(line.slice(6))
      displayNotification(data)
    }
  }
}

async function streamChat(prompt: string) {
  const stream = await ofetch('/api/chat', {
    method: 'POST',
    body: { prompt },
    responseType: 'stream'
  })
  
  const reader = stream.getReader()
  const decoder = new TextDecoder()
  let fullResponse = ''
  
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    
    const text = decoder.decode(value, { stream: true })
    fullResponse += text
    
    // Update UI progressively
    updateChatDisplay(fullResponse)
  }
  
  return fullResponse
}

async function trackProgress(taskId: string) {
  const stream = await ofetch(`/api/tasks/${taskId}/progress`, {
    responseType: 'stream'
  })
  
  const reader = stream.getReader()
  const decoder = new TextDecoder()
  
  try {
    while (true) {
      const { done, value } = await reader.read()
      if (done) break
      
      const chunk = decoder.decode(value)
      const events = chunk.split('\n\n')
      
      for (const event of events) {
        if (event.startsWith('data: ')) {
          const progress = JSON.parse(event.slice(6))
          updateProgressBar(progress.percentage)
          
          if (progress.status === 'complete') {
            reader.cancel()
            break
          }
        }
      }
    }
  } finally {
    reader.releaseLock()
  }
}

interface SSEMessage {
  event?: string
  data: string
  id?: string
  retry?: number
}

async function parseSSE(url: string) {
  const stream = await ofetch(url, { responseType: 'stream' })
  const reader = stream.getReader()
  const decoder = new TextDecoder()
  
  let buffer = ''
  let currentMessage: Partial<SSEMessage> = {}
  
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    
    buffer += decoder.decode(value, { stream: true })
    const lines = buffer.split('\n')
    buffer = lines.pop() || ''
    
    for (const line of lines) {
      if (line === '') {
        // Empty line = message complete
        if (currentMessage.data) {
          yield currentMessage as SSEMessage
        }
        currentMessage = {}
        continue
      }
      
      const colonIndex = line.indexOf(':')
      if (colonIndex === -1) continue
      
      const field = line.slice(0, colonIndex)
      const value = line.slice(colonIndex + 1).trim()
      
      switch (field) {
        case 'event':
          currentMessage.event = value
          break
        case 'data':
          currentMessage.data = (currentMessage.data || '') + value
          break
        case 'id':
          currentMessage.id = value
          break
        case 'retry':
          currentMessage.retry = parseInt(value, 10)
          break
      }
    }
  }
}

// Usage
for await (const message of parseSSE('/api/events')) {
  console.log('Event:', message.event)
  console.log('Data:', message.data)
}

const controller = new AbortController()

const stream = await ofetch('/api/stream', {
  responseType: 'stream',
  signal: controller.signal
})

const reader = stream.getReader()

// Cancel after 10 seconds
setTimeout(() => {
  controller.abort()
  reader.cancel()
}, 10000)

try {
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    processChunk(value)
  }
} catch (error) {
  if (error.name === 'AbortError') {
    console.log('Stream cancelled')
  }
}

const sseClient = ofetch.create({
  baseURL: '/api',
  responseType: 'stream',
  headers: {
    'Accept': 'text/event-stream',
    'Cache-Control': 'no-cache'
  }
})

const stream = await sseClient('/events')
const reader = stream.getReader()

const stream = await ofetch<ReadableStream<Uint8Array>>('/api/sse', {
  responseType: 'stream'
})

// Or let it infer
const stream = await ofetch('/api/sse', {
  responseType: 'stream' as const
})
// stream is typed as ReadableStream<Uint8Array>

const responseType =
  (context.options.parseResponse
    ? "json"
    : context.options.responseType) ||
  detectResponseType(context.response.headers.get("content-type") || "");