---
title: "Stream"
description: "Subscribe to wide events flowing through evlog: in-process with createStreamDrain, or over the network with the local SSE stream server."
canonical_url: "https://www.evlog.dev/extend/stream"
---
# Stream

> Subscribe to wide events flowing through evlog: in-process with createStreamDrain, or over the network with the local SSE stream server.

evlog ships a **stream primitive** so any local consumer can subscribe to wide events without re-implementing a drain. There are two layers, building on each other:

- **In-process bus**: `createStreamDrain()`, the canonical pub/sub. Sync listeners, async iterators, ring-buffered replay.
- **Network bridge**: `startStreamServer()`, an opt-in HTTP mini-server that exposes the in-process bus over Server-Sent Events for browsers, CLIs, or external devtools.

<callout icon="i-lucide-shield">

**Local-only by design.** Both layers live inside a single Node / Bun / Deno process. They work in `pnpm dev`, on long-lived self-hosted servers, on VMs and containers (Fly, Railway, Coolify…). They do **not** work on serverless platforms (Vercel Functions, Cloudflare Workers, AWS Lambda), each invocation is an isolated process. Use a real broker (Redis Streams, NATS, Pub/Sub) for cross-instance fan-out there.

</callout>

## In-process bus

<stream-bus>



</stream-bus>

`createStreamDrain()` is just a drain. Register it on the evlog drain hook and subscribe to events as they're emitted: no HTTP, no serialization, no extra hops.

<prompt :actions="["copy","cursor","claude"]" description="Subscribe to wide events in-process" icon="i-lucide-radio-tower">

Wire an in-process subscriber on top of evlog's stream drain.

- Import `createStreamDrain` from `evlog/stream` and call it once at app boot
- Register the returned `drain` on the evlog drain hook for my framework (Nitro: `nitroApp.hooks.hook('evlog:drain', stream.drain)`; Next/standalone: pass to `initLogger({ drain })`)
- Subscribe with `stream.subscribe((event) => ...)` for sync listeners or `for await (const event of stream.events())` for async iteration
- Seed history for late subscribers with `stream.recent()` (snapshot of the ring buffer) before opening the live iterator
- Tune `buffer` for replay history and `perSubscriberQueue` for slow-consumer backpressure
- Skip on serverless platforms. The stream is in-process, so isolated invocations won't share it

Docs: [https://www.evlog.dev/extend/stream](https://www.evlog.dev/extend/stream)

</prompt>

```ts
import { createStreamDrain } from 'evlog/stream'

const stream = createStreamDrain({ buffer: 200 })

nitroApp.hooks.hook('evlog:drain', stream.drain)

const off = stream.subscribe((event) => {
  if (event.level === 'error') notify(event)
})
off()

for (const past of stream.recent()) {
  bootstrap(past)
}
for await (const event of stream.events()) {
  bootstrap(event)
}
```

### Options

<table>
<thead>
  <tr>
    <th>
      Option
    </th>
    
    <th>
      Type
    </th>
    
    <th>
      Description
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        buffer
      </code>
    </td>
    
    <td>
      <code>
        number
      </code>
    </td>
    
    <td>
      Ring-buffer size for <code>
        recent()
      </code>
      
       snapshots. Set to <code>
        0
      </code>
      
       to disable. Default: <code>
        500
      </code>
      
      .
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        perSubscriberQueue
      </code>
    </td>
    
    <td>
      <code>
        number
      </code>
    </td>
    
    <td>
      Max queued events per <code>
        events()
      </code>
      
       async iterator before the oldest are dropped. The drain itself is never blocked. Default: <code>
        1000
      </code>
      
      .
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        filter
      </code>
    </td>
    
    <td>
      <code>
        (event) => boolean
      </code>
    </td>
    
    <td>
      Optional predicate run on each drained event — return <code>
        false
      </code>
      
       to skip the event entirely (neither buffered nor delivered).
    </td>
  </tr>
</tbody>
</table>

`stream.drain(events)` accepts either a single event or a batch. `stream.recent()` returns a snapshot of the buffered events (oldest first, most recent last) for ad-hoc inspection or to seed a UI panel.

## Network bridge: stream server

<sse-wire>



</sse-wire>

`startStreamServer()` boots a tiny `node:http` server in the same process as your app, on its own ephemeral port, and exposes the in-process bus over Server-Sent Events. Any consumer (browser tab, CLI, Tauri/Electron devtool) can subscribe, and your application API is untouched.

<callout icon="i-lucide-shield">

**Strict opt-in.** Nothing starts unless you set the option explicitly. There is no auto-enable in dev, and the server only boots when you ask for it.

</callout>

<prompt :actions="["copy","cursor","claude"]" description="Expose the evlog stream over SSE" icon="i-lucide-radio">

Turn on the local stream server so I can subscribe to wide events from a browser tab, CLI, or Tauri/Electron devtool.

- Detect my framework and opt in explicitly (Nuxt: `evlog.stream: true` in `nuxt.config.ts`; Next.js: `defineStreamedInstrumentation({ stream: true })` in `instrumentation.ts`; Hono/Express/Fastify/Elysia/standalone: call `startStreamServer()` once at boot and register the returned `drain` on the evlog drain hook)
- Never enable in production by default; gate it behind `process.env.NODE_ENV !== 'production'` or a feature flag
- For shared dev environments, set `token: process.env.EVLOG_STREAM_TOKEN` and have the consumer send it as `Authorization: Bearer <token>` on every request
- Discover the URL from `.evlog/stream.url` (or `/api/_evlog/stream-info` on Nuxt), and never hard-code the port, which is ephemeral
- Skip on serverless platforms. The server is in-process

Docs: [https://www.evlog.dev/extend/stream](https://www.evlog.dev/extend/stream)

</prompt>

### What boots up

When you opt in, evlog calls `startStreamServer()` and:

1. Picks an ephemeral free port (or the one you specify)
2. Spins a tiny `node:http` server on `127.0.0.1` (or your host)
3. Writes the discovered URL to `.evlog/stream.url`
4. Prints a one-line banner in the server console
5. Hooks the in-process stream drain into the evlog pipeline
6. Exposes `GET /` for the SSE stream and `GET /info` for handshake metadata

On shutdown (SIGINT / SIGTERM / process exit) the server cleans up listeners and removes the URL file.

### Per-framework opt-in

<framework-tabs>

```ts [Nuxt]nuxt.config.ts
export default defineNuxtConfig({
  modules: ['evlog/nuxt'],
  evlog: {
    stream: true,
    // or: stream: { port: 4444, token: process.env.EVLOG_STREAM_TOKEN }
  },
})
```

```ts [Next.js]instrumentation.ts
import { defineStreamedInstrumentation } from 'evlog/next/stream'

export const { register } = defineStreamedInstrumentation({
  stream: true,
})
```

```ts [Hono / Express / Fastify / standalone]
import { startStreamServer } from 'evlog/stream'

if (process.env.NODE_ENV !== 'production' && process.env.EVLOG_STREAM === '1') {
  const { drain } = await startStreamServer()
  // Plug `drain` into the evlog drain hook for your framework
}
```

</framework-tabs>

The Hono / Express / Fastify integrations don't need a "feature PR", because `startStreamServer()` is **orthogonal** to your framework. You boot it once and connect its `drain` to the evlog pipeline like any other drain.

### Discovery

The mini-server runs on a random port, so any consumer must discover it.

```ts [.evlog/stream.url]
http://127.0.0.1:53942
```

Read directly from disk:

```ts
import { readFile } from 'node:fs/promises'

const url = (await readFile('.evlog/stream.url', 'utf-8')).trim()
```

Or, for same-origin browser tabs in a Nuxt app, hit the discovery route:

```ts
const { url } = await fetch('/api/_evlog/stream-info').then(r => r.json())
```

### Wire format

Every SSE message has the shape `{ evlog: '1', type, data }`:

<table>
<thead>
  <tr>
    <th>
      <code>
        type
      </code>
    </th>
    
    <th>
      When
    </th>
    
    <th>
      <code>
        data
      </code>
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        hello
      </code>
    </td>
    
    <td>
      First frame after connect
    </td>
    
    <td>
      <code>
        { evlogVersion, bufferSize, heartbeatMs }
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        replay
      </code>
    </td>
    
    <td>
      Right after <code>
        hello
      </code>
      
      , replays buffered events when the consumer passed <code>
        ?since=<iso>
      </code>
    </td>
    
    <td>
      <code>
        WideEvent
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        event
      </code>
    </td>
    
    <td>
      Every emitted event after that
    </td>
    
    <td>
      <code>
        WideEvent
      </code>
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        ping
      </code>
    </td>
    
    <td>
      Heartbeat (default every 15s, configurable via <code>
        heartbeatMs
      </code>
      
      )
    </td>
    
    <td>
      <code>
        { ts: number }
      </code>
    </td>
  </tr>
</tbody>
</table>

```ts [Browser consumer]
const { url } = await fetch('/api/_evlog/stream-info').then(r => r.json())
const es = new EventSource(url)

es.onmessage = (msg) => {
  const { type, data } = JSON.parse(msg.data)
  if (type === 'event') events.push(data)
}
```

### Auth + CORS

<table>
<thead>
  <tr>
    <th>
      Option
    </th>
    
    <th>
      Behavior
    </th>
  </tr>
</thead>

<tbody>
  <tr>
    <td>
      <code>
        token
      </code>
    </td>
    
    <td>
      When set, the server requires <code>
        Authorization: Bearer <token>
      </code>
      
       on every request and 401s otherwise.
    </td>
  </tr>
  
  <tr>
    <td>
      (no token)
    </td>
    
    <td>
      Connections are accepted only when the <code>
        Host
      </code>
      
       header is <code>
        localhost
      </code>
      
      , <code>
        127.0.0.1
      </code>
      
       or <code>
        [::1]
      </code>
      
      , and the <code>
        Origin
      </code>
      
       header, if sent, names one of those hosts exactly. <code>
        https://localhost.example.com
      </code>
      
       and a page reached through a domain that resolves to <code>
        127.0.0.1
      </code>
      
       receive 403.
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        host
      </code>
    </td>
    
    <td>
      Default <code>
        127.0.0.1
      </code>
      
       — never reachable from the LAN. Override to <code>
        0.0.0.0
      </code>
      
       only with a token set.
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        heartbeatMs
      </code>
    </td>
    
    <td>
      Heartbeat interval (default <code>
        15000
      </code>
      
      ).
    </td>
  </tr>
  
  <tr>
    <td>
      <code>
        buffer
      </code>
    </td>
    
    <td>
      Ring buffer kept on the underlying default stream — replayed for late-joining clients via <code>
        ?since=<iso>
      </code>
      
      . Default <code>
        500
      </code>
      
      .
    </td>
  </tr>
</tbody>
</table>

## Going further

- [FS reader](https://www.evlog.dev/extend/fs-reader): replay or tail historic NDJSON files (cross-process, survives restarts)
- [Consumer recipes](https://www.evlog.dev/extend/consumer-recipes): build a minimal devtool, pipe to curl + jq, replay history then go live


## Sitemap

See the full [sitemap](https://www.evlog.dev/sitemap.md) for all pages.