Event Stream Format (text/event-stream) — Complete Guide
In this tutorial, you will learn about Event Stream Format (text/event. We cover key concepts, practical examples, and best practices to help you master this topic.
Master the SSE event stream format: data fields, event types, event IDs, retry intervals, comments, line continuation, and Parsing text/event-stream responses.
What You Learn
You will learn every field in the SSE event stream format: data, event, id, retry, comments, how to format multi-line data, how to set event types, and how the client parses these fields.
Why It Matters
The event stream format is the foundation of SSE. Getting the format wrong causes silent failures: events not received, wrong event types, broken reconnection, or missing data. Understanding the format ensures reliable real-time streaming.
Real-World Use
DodaTech's notification SSE stream sends events with types: 'notification', 'alert', 'status'. Each event includes an ID for reconnection, a retry interval of 5 seconds, and JSON data with the notification payload.
Basic Event Fields
# SSE fields - each line is "field: value"
# Fields are separated by single newline
# Events are separated by double newline
def format_sse_event(data, event=None, event_id=None, retry=None):
lines = []
if event_id:
lines.append(f"id: {event_id}")
if event:
lines.append(f"event: {event}")
if retry:
lines.append(f"retry: {retry}")
lines.append(f"data: {data}")
lines.append("")
return "\n".join(lines)
# Simple data event
print(repr(format_sse_event("Hello World")))
# 'data: Hello World\n\n'
# Named event with ID
print(repr(format_sse_event(
'{"user": "alice", "action": "login"}',
event='user-login',
event_id='evt-42'
)))
# 'event: user-login\nid: evt-42\ndata: {"user": "alice", "action": "login"}\n\n'
Multi-Line Data
# Data can span multiple lines.
# Each line starts with "data: " and the client concatenates them with newline.
def format_multiline_data(lines):
result = []
for line in lines:
result.append(f"data: {line}")
result.append("")
return "\n".join(result)
log_lines = [
"[INFO] Starting backup",
"[INFO] Connecting to database",
"[INFO] Backup completed",
]
event = format_multiline_data(log_lines)
print(event)
Expected output:
data: [INFO] Starting backup
data: [INFO] Connecting to database
data: [INFO] Backup completed
Event Types
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
class EventTypeSSE(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.end_headers()
events = [
('status', {'level': 'info', 'message': 'Server starting'}),
('user-update', {'user': 'alice', 'status': 'online'}),
('notification', {'title': 'New message', 'count': 3}),
('heartbeat', {'timestamp': time.time()}),
('status', {'level': 'info', 'message': 'Server running'}),
]
for event_type, data in events:
self.wfile.write(f"event: {event_type}\n".encode())
self.wfile.write(f"data: {json.dumps(data)}\n\n".encode())
time.sleep(2)
# Client listens:
# source.addEventListener('status', (e) => ...)
# source.addEventListener('user-update', (e) => ...)
# source.addEventListener('notification', (e) => ...)
# source.addEventListener('heartbeat', (e) => ...)
// Listening for specific event types
const source = new EventSource('/events');
source.addEventListener('status', (event) => {
console.log('Status:', JSON.parse(event.data));
});
source.addEventListener('user-update', (event) => {
console.log('User:', JSON.parse(event.data));
});
source.addEventListener('notification', (event) => {
const notification = JSON.parse(event.data);
showNotification(notification.title, notification.count);
});
// Catch-all for unnamed events
source.onmessage = (event) => {
console.log('Unnamed event:', event.data);
};
Event IDs and Reconnection
# The "id" field sets the last event ID.
# On reconnection, the browser sends "Last-Event-ID" header.
# The server can resume from this point.
import time
import json
class SSEWithResume:
def __init__(self):
self.events = []
self.event_counter = 0
def add_event(self, data, event=None):
self.event_counter += 1
self.events.append({
'id': self.event_counter,
'event': event,
'data': data,
'time': time.time(),
})
def stream_from(self, last_event_id):
"""Generate events from a given point."""
start_id = int(last_event_id) if last_event_id else 0
for event in self.events:
if event['id'] > start_id:
yield event
def format_event(self, event):
lines = [f"id: {event['id']}"]
if event['event']:
lines.append(f"event: {event['event']}")
lines.append(f"data: {json.dumps(event['data'])}")
lines.append("")
return "\n".join(lines)
sse = SSEWithResume()
sse.add_event({'msg': 'first'})
sse.add_event({'msg': 'second'}, event='update')
sse.add_event({'msg': 'third'})
# Resume from ID 1 (skip first event)
for event in sse.stream_from(1):
print(sse.format_event(event))
Expected output:
id: 2
event: update
data: {"msg": "second"}
id: 3
data: {"msg": "third"}
Retry Interval
# The "retry" field tells the browser how long to wait before reconnecting
# after a connection loss. Value is in milliseconds.
def sse_with_retry(retry_ms=3000):
lines = [f"retry: {retry_ms}", "", "data: connected", ""]
return "\n".join(lines)
print(sse_with_retry(5000))
# retry: 5000
# (blank line - empty event)
# data: connected
# (blank line)
# The browser will now wait 5 seconds between reconnection attempts
# instead of the default 2-3 seconds.
Comments in SSE
# Lines starting with ":" are comments.
# Comments can be used for keep-alive or debugging.
def keep_alive_stream():
import time
while True:
# Comment line acts as keep-alive
yield ": heartbeat\n\n"
time.sleep(15)
# Comments are ignored by the EventSource API
# but keep the connection alive through proxies.
def sse_with_comments():
yield ": SSE stream started\n"
yield ": Server version: 1.0\n"
yield "event: start\n"
yield "data: Stream initialized\n\n"
Common Mistakes
1. Wrong Line Ending Format
SSE requires \n (LF), not \r\n (CRLF). Some browsers accept CRLF but the spec requires LF. Use \n for maximum compatibility.
2. Missing Blank Line Between Events
Events are separated by a blank line (double \n). Without it, multiple events are concatenated into one.
3. No Space After Colon
The format is "field: value" with a space after the colon. "field:value" (no space) may not be parsed correctly by all clients.
4. Sending Non-JSON Data Without Escaping
If data contains newlines, the event splits into multiple data lines. Escape or encode newlines in data.
5. Confusing Event ID Types
Event IDs must be string-comparable. Using integers is fine, but ensure format is consistent. "1" and "01" are different IDs.
Practice Questions
1. What is the purpose of the "id" field in SSE?
It sets the Last-Event-ID. On reconnection, the browser sends this value so the server can resume from the last successful event.
2. How do you send multi-line data in SSE?
Split the data into multiple "data:" lines. The client concatenates them with newline characters.
3. What does the "retry" field do?
It tells the browser how many milliseconds to wait before reconnecting after a connection loss. Default is 2-3 seconds.
4. How do you handle different event types on the client?
Use source.addEventListener('eventname', callback) for named events. Use source.onmessage for unnamed events.
Challenge
Create an SSE endpoint that: uses event types for different data categories (metrics, logs, alerts), sets event IDs for reconnection support, sends retry: 5000 for 5-second reconnection, includes keep-alive comments every 15 seconds, and handles multi-line log messages.
FAQ
{{< faq "Can I omit the "data:" field?" "No. Every event must have at least a data field. An event with no data may be ignored by the client." >}}
{{< faq "How does the browser handle the "event:" field?" "The browser dispatches a DOM event with the specified type. JavaScript can listen with addEventListener(type, handler)." >}}
Mini Project: SSE Format Validator
#!/usr/bin/env python3
"""Validate SSE event stream format."""
import sys
import re
def validate_sse(stream):
errors = []
lines = stream.split('\n')
in_event = False
for i, line in enumerate(lines):
if line == '':
in_event = False
continue
if line.startswith(':') or line == '':
continue
if not re.match(r'^(data|event|id|retry): ', line):
errors.append(f"Line {i+1}: Invalid field format: {line[:50]}")
if line.startswith('data: ') and in_event:
errors.append(f"Line {i+1}: Unexpected data field during event")
if line.startswith('event: ') or line.startswith('data: '):
in_event = True
return errors
def test_valid():
stream = "data: hello\n\nevent: update\ndata: world\n\n"
errors = validate_sse(stream)
assert len(errors) == 0, f"Expected no errors, got: {errors}"
print("Valid SSE: OK")
def test_invalid():
stream = "data: hello\nwrong: field\n\n"
errors = validate_sse(stream)
assert len(errors) > 0, "Expected errors"
print(f"Invalid SSE detected: {len(errors)} errors")
test_valid()
test_invalid()
What's Next
Now that you understand the event stream format, explore the EventSource API for client-side handling, then learn SSE with Express.js.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro