Stream messages

A stream delivers conversations and messages to a callback as they arrive. Streams are infinite. They do not end on their own.

Stream methods

Stream	Fires when	Value
`conversations.stream`	A conversation is created or received	Conversation
`conversations.streamGroups`	A group is created or received	Group
`conversations.streamDms`	A DM is created or received	DM
`conversations.streamAllMessages`	A message arrives in any conversation	Message
`conversation.stream` / `streamMessages`	A message arrives in one conversation	Message
`preferences.streamConsent`	A consent record changes	Consent record
`preferences.streamPreferences`	A preference changes	Preference type

Kotlin and Swift use a type filter instead of the separate group and DM stream methods. They name the preference method streamPreferenceUpdates.

  const 
const stream: AsyncStreamProxy<Group<BuiltInContentTypes> | Dm<BuiltInContentTypes>>
stream = await 
client: Client<BuiltInContentTypes>
client.
Client<BuiltInContentTypes>.conversations: Conversations<BuiltInContentTypes>
Gets the conversations manager for this client
conversations.
Conversations<BuiltInContentTypes>.stream<Group<BuiltInContentTypes> | Dm<BuiltInContentTypes>>(options?: (StreamOptions<SafeConversation, Group<BuiltInContentTypes> | Dm<BuiltInContentTypes>> & {
    conversationType?: ConversationType;
}) | undefined): Promise<AsyncStreamProxy<Group<BuiltInContentTypes> | Dm<BuiltInContentTypes>>>
Creates a stream for new conversations
@param ― options - Optional stream options
@param ― options.conversationType - Optional type to filter conversations
@returns ― Stream instance for new conversations
stream({
    
onValue?: ((value: Group<BuiltInContentTypes> | Dm<BuiltInContentTypes>) => void | Promise<void>) | undefined
Called when a value is emitted from the stream.
For message streams, this selects callback mode. Do not also iterate that stream.
Message delivery is acknowledged after this callback returns successfully.
onValue: 
handleConversation: (conversation: Conversation | undefined) => void
handleConversation,
  });

  const 
const stream: AsyncStreamProxy<Group<BuiltInContentTypes> | Dm<BuiltInContentTypes>>
stream = await 
client: Client<BuiltInContentTypes>
client.
Client<BuiltInContentTypes>.conversations: Conversations<BuiltInContentTypes>
Gets the conversations manager for this client
@throws ― {ClientNotInitializedError} if the client is not initialized
conversations.
Conversations<BuiltInContentTypes>.stream(options?: (StreamOptions<Conversation, Group<BuiltInContentTypes> | Dm<BuiltInContentTypes> | undefined> & {
    conversationType?: ConversationType;
}) | undefined): Promise<AsyncStreamProxy<Group<BuiltInContentTypes> | Dm<BuiltInContentTypes>>>
Creates a stream for new conversations
@param ― options - Optional stream options
@param ― options.conversationType - Optional conversation type to filter by
@returns ― Stream instance for new conversations
@see ― https://docs.xmtp.org/sdk/stream/#stream-methods
stream({
    
onValue?: ((value: Group<BuiltInContentTypes> | Dm<BuiltInContentTypes> | undefined) => void | Promise<void>) | undefined
Called when a value is emitted from the stream.
For message streams, this selects callback mode. Do not also iterate that stream.
Message delivery is acknowledged after this callback returns successfully.
onValue: 
handleConversation: (conversation: Conversation | undefined) => void
handleConversation,
  });

client.conversations.stream(type = ConversationType.ALL).collect(::handleConversation)

for try await conversation in client.conversations.stream(type: .all) {
  handleConversation(conversation)
}

By default, streamAll streams only conversations with a consent state of allowed or unknown.

We recommend streaming messages for allowed conversations only. This ensures that spammy conversations with a consent state of unknown don’t take up networking resources. This also ensures that unwanted spam messages aren’t stored in the user’s local database.

Important

The stream is infinite. Therefore, any looping construct used with the stream won’t terminate unless you explicitly initiate the termination. You can initiate the termination by breaking the loop or by making an external call to return.

Catch-up

When a stream registers a topic, it captures a fixed catch-up target. It delivers every retained envelope above the local cursor through that target, then continues with new envelopes. New publications do not move the target.

A topic is caught up only after the SDK safely processes the range through the target. To receive only live traffic, sync before starting the stream.

Retry

SDK	Attempts	Delay	Automatic
Node	10	60,000 ms	Yes
Browser	6	10,000 ms	Yes
Kotlin	—	—	No; restart after `onClose`
Swift	—	—	No; restart after `onClose`

Set retryOnFail: false to turn off Browser or Node retries. Node calls onFail; Browser throws.

Agent SDK events

The Agent SDK wraps Node streams and emits events named for the content type. It also emits message, unknownMessage, conversation, group, dm, start, stop, and unhandledError. Its additional retry layer uses up to 10 attempts with exponential backoff from 1 second to 30 seconds.