Streaming messages

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)`);
}

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)")
  }
}

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)")
}