Stream.send()

Use Stream.send() to publish one or more records to an existing Stream when the producer can import your Moose SDK models and Stream definition directly. If the producer should only send JSON over HTTP, use an Ingest API instead.

This page covers in-process producers. To define the stream itself, start with Stream.

When to use

Use Stream.send() when the producer can import the stream definition directly, such as an existing app in your monorepo or a Moose workflow.

Use an Ingest API when the producer should only send JSON over HTTP or cannot import Moose objects directly.

Send records

Stream.send() takes one argument. Pass a single record to publish one message, or pass an array or list to publish multiple messages. Moose publishes one stream record per element, not one array payload. There are no per-call options on send().

import { userEvents } from "./stream"; await userEvents.send({  id: "evt_1",  userId: "user_123",  timestamp: new Date(),  eventType: "click",});

import { userEvents } from "./stream"; await userEvents.send([  { id: "evt_2", userId: "user_456", timestamp: new Date(), eventType: "view" },  { id: "evt_3", userId: "user_789", timestamp: new Date(), eventType: "signup" },]);

Send nothing

This is useful in helper code that batches optional records or filters a collection down to zero results. Top-level empty inputs are valid and send nothing:

await userEvents.send(undefined);await userEvents.send(null);await userEvents.send([]);

Behavior

Stream.send() publishes to the stream topic. If the stream also has a destination table, records become queryable only after Moose syncs them into ClickHouse.

If the stream is configured for Schema Registry JSON, send() publishes using the Confluent wire format automatically. See Schema Registry.

Related capabilities

• Stream for defining the stream resource
• Sync to table when records should land in ClickHouse
• Schema registry when payloads should use Confluent Schema Registry
• Ingest API for external HTTP ingestion into streams