Streams

External event sources like server-sent events, WebSockets, DOM events, or anything else that pushes values over time are subscribed to with the $stream function. Events are written into reactive state, so templates and effects update through the normal reactivity model:

export default function Chat($props: { room: string }) {
	let $state = $watch({
		messages: [] as string[]
	})

	$stream(fromServer(() => "/sse/" + $props.room), (e) => {
		$state.messages.push(e.data)
	})

	@render {
		<ul>
			@for (let m of $state.messages) {
				<li>{m}</li>
			}
		</ul>
	}
}

Managed subscriptions

The subscription is managed by the framework:

• It starts when the component is mounted to the DOM, so &ref-bound elements already exist, and it is unsubscribed when the component unmounts or its region is cleared.
• The source is never invoked during a server render, so browser-only APIs like EventSource are safe to use without any typeof window guards.
• Errors are values like any other: a source that can fail should push an error-shaped value and own its own reconnection (EventSource reconnects automatically).

Re-subscribing on state changes

Reactive state read inside the source is tracked. When it changes, the old subscription is torn down and a fresh one is opened — for example, re-opening the connection when a room id changes. Pass the URL as a getter for this to work; a plain string URL is read once.

Capture tracked reads in a local variable inside the source, so its cleanup function unsubscribes the listener it actually added (not whatever the value is at teardown time).

The fromServer function

fromServer(url) receives message events from a server-sent events endpoint, and fromWebSocket(url) does the same for a WebSocket. The URL may be a string or a getter, as above.

The fromElement function

fromElement(el, type) pushes DOM events of the given type. Pass a getter returning a &ref-bound variable, since elements only exist once they are rendered:

export default function AutoSave() {
	let input: HTMLInputElement
	let $state = $watch({
		saved: ""
	})

	$stream(fromElement(() => input, "input"), () => {
		$state.saved = "Saved"
	}, { debounce: 500 })

	@render {
		<input &ref={input} oninput={(e) => { $state.draft = e.target.value }} />
		<p>{$state.saved}</p>
	}
}

For simple cases, a plain event handler is fine — see Events. Reach for fromElement when you want the events debounced, counted, or handled by the framework's lifecycle.

The debounce option

Pass { debounce: ms } to delay each handler call until the source has been quiet for that many milliseconds. Every event resets the timer, so only the last event of a burst is handled. A pending debounced call is dropped if the subscription is torn down or re-subscribed first.

Custom sources

A source is just a function that takes a push callback and returns an unsubscribe function (the StreamSource<T> type), so wrapping things like an EventEmitter, a BroadcastChannel or a third-party SDK takes just a few lines:

$stream((push) => {
	const socket = new WebSocket(url)
	socket.onmessage = (e) => push(e.data)
	return () => socket.close()
}, (message) => {
	$state.inbox.push(message)
})

Timing options belong on $stream; anything shapeful like windowing, aggregation, or combining multiple streams is plain closure logic in the handler. To react to each event in an effect, write it to state and read it from $run. For async data (once-off fetches rather than ongoing streams) see Async.