> ## Documentation Index
> Fetch the complete documentation index at: https://docs.expys.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming messages

> Live concierge messages over Server-Sent Events, consumed as an AsyncIterable, AsyncStream, or Flow.

The concierge delivers new messages live over **Server-Sent Events (SSE)** at
`GET /v1/conversations/{id}/stream`. Because the response is an open
`text/event-stream` rather than a single request/response, it is documented here
as a concept rather than in the interactive reference - the "Try it" playground
cannot represent a long-lived stream.

<Info>
  Streaming is **member-mode**: it uses the member token, same as the rest of the
  concierge. See [Conversations](/guides/conversations) for the
  request/response message operations.
</Info>

## Consuming the stream

Each SDK exposes the stream as the idiomatic async sequence for its language, so
you consume new messages with a normal loop:

| Language   | Type                     | Consume with                       |
| ---------- | ------------------------ | ---------------------------------- |
| TypeScript | `AsyncIterable<Message>` | `for await (const message of ...)` |
| Swift      | `AsyncStream<Message>`   | `for try await message in ...`     |
| Kotlin     | `Flow<Message>`          | `.collect { message -> ... }`      |

A common pattern is to print the recent backlog with
[`listMessages`](/guides/conversations) first, then live-stream what follows:

<CodeGroup>
  ```typescript TypeScript theme={null}
  import { initialize } from "@expys/sdk";

  const token = process.env.EXPYS_MEMBER_TOKEN;
  if (!token) {
    throw new Error(
      "Set EXPYS_MEMBER_TOKEN (a member token from your backend's /v1/auth/exchange)",
    );
  }

  const conversationId = process.env.EXPYS_CONVERSATION_ID;
  if (!conversationId) {
    throw new Error("Set EXPYS_CONVERSATION_ID (a conversation to stream)");
  }
  // Narrow for use inside main() (a module-level const isn't narrowed across it).
  const cnv: string = conversationId;

  const expys = initialize({
    baseUrl: process.env.EXPYS_BASE_URL,
    environment: "sandbox",
    token,
  });

  async function main(): Promise<void> {
    // Optional: print the recent backlog first, then live-stream what follows.
    const { messages } = await expys.listMessages(cnv, { limit: 20 });
    for (const message of messages) {
      console.log(`[history ${message.authorID}] ${message.body ?? "(no body)"}`);
    }

    let received = 0;
    console.log("listening for new messages (stops after 5)...");

    // `for await` consumes the AsyncIterable lazily. Breaking the loop (here, after
    // five messages) tears down the underlying HTTP connection and any reconnect
    // timer - no leaked sockets. In a real app you would break on a signal /
    // unmount instead of a fixed count.
    for await (const message of expys.streamMessages(cnv)) {
      received += 1;
      console.log(`[live ${message.authorID}] ${message.body ?? "(no body)"}`);
      if (received >= 5) {
        break; // closes the connection
      }
    }

    console.log(`done: received ${received} live message(s)`);
  }
  ```

  ```swift Swift theme={null}
  import ExpysSDK
  import Foundation

  @main
  struct StreamMessagesExample {
    static func main() async throws {
      let environment = ProcessInfo.processInfo.environment
      guard let token = environment["EXPYS_MEMBER_TOKEN"] else {
        fatalError("Set EXPYS_MEMBER_TOKEN (a member token from your backend's /v1/auth/exchange)")
      }
      guard let conversationID = environment["EXPYS_CONVERSATION_ID"] else {
        fatalError("Set EXPYS_CONVERSATION_ID (a conversation to stream)")
      }
      let baseURL =
        environment["EXPYS_BASE_URL"].flatMap(URL.init(string:))
        ?? ExpysConfiguration.defaultBaseURL

      let client = ExpysClient(
        configuration: ExpysConfiguration(token: token, environment: .sandbox, baseURL: baseURL)
      )

      // Optional: print the recent backlog first, then live-stream what follows.
      let history = try await client.listMessages(id: conversationID, limit: 20)
      for message in history.messages {
        print("[history \(message.authorID)] \(message.body ?? "(no body)")")
      }

      print("listening for new messages (stops after 5)...")

      // Drive the stream from a Task so cancellation tears down the connection.
      // `for try await` consumes the AsyncThrowingStream lazily; breaking the loop
      // (here, after five messages) cancels the consuming Task, which severs the
      // underlying connection and any pending reconnect timer - no leaked sockets.
      let task = Task {
        var received = 0
        for try await message in client.streamMessages(id: conversationID) {
          received += 1
          print("[live \(message.authorID)] \(message.body ?? "(no body)")")
          if received >= 5 {
            break  // closes the connection
          }
        }
        return received
      }

      let received = try await task.value
      print("done: received \(received) live message(s)")
    }
  }
  ```

  ```kotlin Kotlin theme={null}
  package com.expys.sdk.examples.streammessages

  import com.expys.sdk.ExpysClient
  import com.expys.sdk.ExpysConfiguration
  import kotlinx.coroutines.flow.take
  import kotlinx.coroutines.runBlocking

  fun main() = runBlocking {
    val token = System.getenv("EXPYS_MEMBER_TOKEN")
      ?: error("Set EXPYS_MEMBER_TOKEN (a member token from your backend's /v1/auth/exchange)")
    val conversationId = System.getenv("EXPYS_CONVERSATION_ID")
      ?: error("Set EXPYS_CONVERSATION_ID (a conversation to stream)")

    val client = ExpysClient.create(
      ExpysConfiguration(
        token = token,
        baseUrl = System.getenv("EXPYS_BASE_URL") ?: ExpysConfiguration.DEFAULT_BASE_URL,
      ),
    )

    // Optional: print the recent backlog first, then live-stream what follows.
    val history = client.listMessages(conversationId, limit = 20)
    for (message in history.messages) {
      println("[history ${message.authorID}] ${message.body ?: "(no body)"}")
    }

    println("listening for new messages (stops after 5)...")

    // collect() consumes the cold Flow lazily; `take(5)` cancels collection after
    // five messages, which tears down the underlying connection and any pending
    // reconnect timer - no leaked coroutines. In a real app you would cancel the
    // collecting coroutine's scope (e.g. a ViewModel scope) instead.
    var received = 0
    client.streamMessages(conversationId).take(5).collect { message ->
      received++
      println("[live ${message.authorID}] ${message.body ?: "(no body)"}")
    }

    println("done: received $received live message(s)")
  }
  ```
</CodeGroup>

## Reconnect and backoff

The SDKs reconnect automatically if the stream drops, using the same full-jitter
backoff as the rest of the client (base 500ms, capped at 10s). You consume one
continuous sequence of messages; transient disconnects are handled underneath.

## Cancellation

Stopping consumption tears down the underlying HTTP connection and any pending
reconnect timer - no leaked sockets:

<Tabs>
  <Tab title="TypeScript">
    `break` out of the `for await` loop (or `return`), and the connection closes.
  </Tab>

  <Tab title="Swift">
    Cancel the enclosing `Task`; the `AsyncStream` terminates and the connection
    closes.
  </Tab>

  <Tab title="Kotlin">
    Cancel the collecting coroutine (its `Job` or scope); the `Flow` collection
    stops and the connection closes.
  </Tab>
</Tabs>

<Tip>
  In a real app, end the stream on a lifecycle signal - a screen unmount, a
  cancelled task, or a closed scope - rather than after a fixed number of
  messages as the example does.
</Tip>