SSE Real-time Push

EnerOS provides Server-Sent Events (SSE) endpoints for browser-native unidirectional real-time event push. Compared to WebSocket, SSE is lighter, auto-reconnects on disconnection, and uses standard HTTP ports, making it suitable for read-only subscription scenarios such as alarm dashboards, event display screens, and operations notifications. SSE is based on HTTP long connections, with the server streaming event pushes via text/event-stream, and browsers natively receive them through the EventSource API.

SSE Protocol Description

Protocol Characteristics

SSE is part of the HTML5 specification, based on HTTP/1.1 long connections, with the server continuously pushing text event streams. Main features:

• Unidirectional communication: Server→Client only; clients cannot send data through the same connection
• Text protocol: Supports UTF-8 text only, not suitable for binary data
• Auto-reconnection: Browser natively implements disconnection reconnection, no client code needed
• Resumption: Missed events are replayed via the Last-Event-ID header
• HTTP compatibility: Uses standard 80/443 ports, passes through proxies and firewalls
• Connection reuse: Under HTTP/2, can reuse the same TCP connection with other requests

Protocol Format

SSE response Content-Type is text/event-stream, with each event consisting of several field lines separated by blank lines:

event: <event type>
id: <event ID>
retry: <reconnection interval milliseconds>
data: <data line 1>
data: <data line 2>

Lines starting with : are comment lines, commonly used for heartbeat keepalive.

Endpoint Information

Subscription Parameters

Filter the event stream via Query parameters:

Event Types

The SSE endpoint pushes the following event types, with each event field corresponding to a category:

Message Format Examples

alarm Event

event: alarm
id: evt_20260706_001
retry: 3000
data: {"id":"alm_001","severity":"critical","status":"active","type":"BRANCH_OVERLOAD","message":"Branch 1-2 overload","network_id":"net_8f3a2b","raised_at":"2026-07-06T08:30:00Z"}

agent Event

event: agent
id: evt_20260706_002
data: {"agent_id":"dispatch_01","name":"DispatchAgent-01","old_status":"idle","new_status":"running","reason":"manual_start","timestamp":"2026-07-06T08:30:05Z"}

Multi-line data Example

event: topology
id: evt_20260706_003
data: {"network_id":"net_8f3a2b","kind":"TopologyChanged"}
data: {"changes":[{"branch":"1-2","status":"open"},{"branch":"2-3","status":"closed"}]}
data: {"actor":{"type":"agent","id":"agent_self_healing_01"}}

Heartbeat Comment Line

:keepalive

System Notification

event: system
id: evt_20260706_004
data: {"type":"maintenance_scheduled","message":"System will enter maintenance mode in 30 minutes","scheduled_at":"2026-07-06T09:00:00Z"}

Browser Examples

Basic Subscription

const es = new EventSource(
  'http://localhost:8080/api/v1/events/stream?topics=alarm,agent&severity=critical',
  { withCredentials: true }
);

es.addEventListener('alarm', (e) => {
  const data = JSON.parse(e.data);
  console.log('Alarm:', data.severity, data.message);
  showAlarmBanner(data);
});

es.addEventListener('agent', (e) => {
  const data = JSON.parse(e.data);
  console.log('Agent status:', data.agent_id, data.new_status);
  updateAgentStatus(data.agent_id, data.new_status);
});

es.addEventListener('open', () => {
  console.log('SSE connection established');
});

es.addEventListener('error', (e) => {
  if (es.readyState === EventSource.CLOSED) {
    console.log('Connection closed');
  } else {
    console.log('Connection error, browser will auto-reconnect');
  }
});

Subscription with Token

Browser EventSource does not support custom Headers, so the Token must be passed via Query parameter:

const token = localStorage.getItem('eneros_token');
const es = new EventSource(
  `http://localhost:8080/api/v1/events/stream?topics=alarm&token=${encodeURIComponent(token)}`,
  { withCredentials: true }
);

es.addEventListener('alarm', (e) => {
  const data = JSON.parse(e.data);
  console.log('Alarm:', data);
});

Resumption

The server automatically handles resumption via the Last-Event-ID header. The browser automatically attaches this Header on reconnection, no manual handling required. However, for non-browser clients, manual maintenance is needed:

let lastEventId = null;

function connect() {
  const url = lastEventId
    ? `http://localhost:8080/api/v1/events/stream?topics=alarm&last_event_id=${lastEventId}`
    : 'http://localhost:8080/api/v1/events/stream?topics=alarm';
  const es = new EventSource(url);

  es.addEventListener('alarm', (e) => {
    lastEventId = e.lastEventId;
    const data = JSON.parse(e.data);
    console.log('Alarm:', data);
  });

  es.addEventListener('error', () => {
    // Browser will auto-reconnect and attach Last-Event-ID header
    console.log('Waiting for auto-reconnection...');
  });
}

connect();

cURL Example

curl -N -H "Authorization: Bearer <token>" \
  "http://localhost:8080/api/v1/events/stream?topics=alarm&severity=critical"

The -N parameter disables buffering, ensuring events are output in real-time.

Python Example

import requests
import json

url = "http://localhost:8080/api/v1/events/stream"
headers = {"Authorization": "Bearer <token>"}
params = {"topics": "alarm,agent", "severity": "critical"}

response = requests.get(url, headers=headers, params=params, stream=True)
for line in response.iter_lines(decode_unicode=True):
    if not line:
        continue
    if line.startswith("event:"):
        event_type = line[6:].strip()
    elif line.startswith("id:"):
        event_id = line[3:].strip()
    elif line.startswith("data:"):
        data = json.loads(line[5:].strip())
        print(f"[{event_type}] {data}")

Rust Example

use reqwest::header;
use futures_util::StreamExt;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let client = reqwest::Client::new();
    let mut headers = header::HeaderMap::new();
    headers.insert(
        header::AUTHORIZATION,
        header::HeaderValue::from_static("Bearer <token>"),
    );

    let response = client
        .get("http://localhost:8080/api/v1/events/stream")
        .headers(headers)
        .query(&[("topics", "alarm"), ("severity", "critical")])
        .send()
        .await?;

    let mut stream = response.bytes_stream();
    let mut buffer = String::new();

    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        buffer.push_str(&String::from_utf8_lossy(&chunk));

        while let Some(pos) = buffer.find("\n\n") {
            let event_str = buffer[..pos].to_string();
            buffer = buffer[pos + 2..].to_string();

            for line in event_str.lines() {
                if line.starts_with("event:") {
                    println!("Type: {}", line[6..].trim());
                } else if line.starts_with("data:") {
                    println!("Data: {}", line[5..].trim());
                } else if line.starts_with("id:") {
                    println!("ID: {}", line[3..].trim());
                }
            }
            println!("---");
        }
    }

    Ok(())
}

JavaScript Complete Dashboard Example

class EnerOSDashboard {
  constructor(token) {
    this.token = token;
    this.es = null;
    this.alarmCount = 0;
    this.agentStatus = new Map();
  }

  start() {
    const url = new URL('http://localhost:8080/api/v1/events/stream');
    url.searchParams.set('topics', 'alarm,agent,topology,constraint');
    url.searchParams.set('token', this.token);

    this.es = new EventSource(url.toString());

    this.es.addEventListener('alarm', (e) => this.onAlarm(e));
    this.es.addEventListener('agent', (e) => this.onAgent(e));
    this.es.addEventListener('topology', (e) => this.onTopology(e));
    this.es.addEventListener('constraint', (e) => this.onConstraint(e));
    this.es.addEventListener('system', (e) => this.onSystem(e));

    this.es.addEventListener('open', () => {
      console.log('[Dashboard] Connected');
    });

    this.es.addEventListener('error', (e) => {
      if (this.es.readyState === EventSource.CLOSED) {
        console.error('[Dashboard] Connection closed, reconnecting in 5s');
        setTimeout(() => this.start(), 5000);
      }
    });
  }

  onAlarm(e) {
    const data = JSON.parse(e.data);
    this.alarmCount++;
    console.log(`[Alarm #${this.alarmCount}] ${data.severity}: ${data.message}`);
  }

  onAgent(e) {
    const data = JSON.parse(e.data);
    this.agentStatus.set(data.agent_id, data.new_status);
    console.log(`[Agent] ${data.agent_id}: ${data.old_status} → ${data.new_status}`);
  }

  onTopology(e) {
    const data = JSON.parse(e.data);
    console.log(`[Topology] ${data.network_id}:`, data.changes);
  }

  onConstraint(e) {
    const data = JSON.parse(e.data);
    console.log(`[Constraint] ${data.element} ${data.quantity}: ${data.value} / ${data.limit}`);
  }

  onSystem(e) {
    const data = JSON.parse(e.data);
    console.log(`[System] ${data.type}: ${data.message}`);
  }

  stop() {
    if (this.es) {
      this.es.close();
      this.es = null;
    }
  }
}

const dashboard = new EnerOSDashboard('<bearer-token>');
dashboard.start();

Reconnection Mechanism

Browser Auto-Reconnection

Browser EventSource automatically reconnects after connection disconnection, with a default interval of 3 seconds. The server can adjust this via the retry field:

retry: 5000

Last-Event-ID Resumption

The browser automatically attaches the Last-Event-ID HTTP header on reconnection, containing the last received event ID. The server uses this to replay missed events:

GET /api/v1/events/stream?topics=alarm HTTP/1.1
Last-Event-ID: evt_20260706_001

Server processing logic:

1. Parse the Last-Event-ID Header (or Query parameter last_event_id)
2. Query all events after this ID from event storage
3. Replay in ID order
4. After replay completes, continue pushing real-time events

Reconnection Backoff Strategy

1st reconnection: 3s
2nd reconnection: 3s
3rd reconnection: 6s
4th reconnection: 12s
5th and above: 30s (cap)

The server dynamically adjusts via the retry field:

retry: 3000
event: system
data: {"type":"reconnect_backoff","interval_ms":6000}

SSE vs WebSocket Comparison

When to Choose SSE

• Only need server push, no need for client to send data
• Need browser-native auto-reconnection
• Deployment environment has high proxy traversal requirements
• Alarm dashboards, event display screens, operations notifications

When to Choose WebSocket

• Need client to send commands to server
• Need to transmit binary data
• Need higher real-time performance (millisecond level)
• Agent remote control, bidirectional collaboration

Limitations

Error Handling

Connection Errors

In-Stream Errors

After the SSE connection is established, errors are pushed via system events:

event: system
id: evt_err_001
data: {"type":"error","code":"SUBSCRIPTION_EXPIRED","message":"Subscription expired, please reconnect"}

Performance Tuning

• Use topics filtering wisely: Avoid subscribing to all topics to prevent bandwidth waste
• Use HTTP/2: Multiplexing avoids connection count limits
• Client-side rate limiting: In high-frequency event scenarios, clients should debounce or throttle
• Compression transmission: Enable gzip compression to reduce bandwidth usage
• CDN caching: Historical event queries can use REST endpoints + CDN caching

Related Documentation

• API Overview — Comparison of four API types
• WebSocket — Bidirectional real-time communication
• Rate Limit and Quota — Connection limits
• REST API — Historical event queries
• eneros-eventbus crate — Event bus implementation