Event Types and Custom Events
In this tutorial, you will learn about Event Types and Custom Events. We cover key concepts, practical examples, and best practices to help you master this topic.
Use named event types in SSE: define custom event types with the event field, handle them on the client, build event-driven architectures, and design event schemas for real-time applications.
What You Learn
You will learn how to define and use custom event types in SSE, how the client dispatches named events, how to design event schemas, and how to build event-driven SSE architectures.
Why It Matters
Without named events, all SSE messages go to the same onmessage handler. Named events enable the client to route different messages to different handlers, creating a clean Event-Driven Architecture.
Real-World Use
DodaTech's SSE stream uses 5 event types: notification (user alerts), metrics (dashboard data), heartbeat (connection keep-alive), error (failure reports), and status (connection lifecycle). Each triggers different UI updates.
Server-Side Event Types
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
import random
class EventTypeHandler(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.send_header('Connection', 'keep-alive')
self.end_headers()
event_id = 0
while True:
event_id += 1
event_type = random.choice(['notification', 'metric', 'status', 'alert'])
if event_type == 'notification':
self._send_event(event_id, 'notification', {
'title': 'New message',
'body': 'You have a new notification',
'severity': random.choice(['info', 'warning']),
})
elif event_type == 'metric':
self._send_event(event_id, 'metric', {
'cpu': random.uniform(0, 100),
'memory': random.uniform(0, 100),
'timestamp': time.time(),
})
elif event_type == 'status':
self._send_event(event_id, 'status', {
'service': 'database',
'status': random.choice(['healthy', 'degraded']),
})
elif event_type == 'alert':
self._send_event(event_id, 'alert', {
'level': random.choice(['critical', 'warning']),
'message': 'Anomaly detected',
})
time.sleep(2)
def _send_event(self, event_id, event_type, data):
self.wfile.write(f"id: {event_id}\n".encode())
self.wfile.write(f"event: {event_type}\n".encode())
self.wfile.write(f"data: {json.dumps(data)}\n\n".encode())
Client-Side Event Dispatching
const source = new EventSource('/events');
// Named event handlers
source.addEventListener('notification', (event) => {
const data = JSON.parse(event.data);
showNotification(data.title, data.body, data.severity);
});
source.addEventListener('metric', (event) => {
const data = JSON.parse(event.data);
updateCpuGauge(data.cpu);
updateMemoryGauge(data.memory);
});
source.addEventListener('status', (event) => {
const data = JSON.parse(event.data);
updateServiceStatus(data.service, data.status);
});
source.addEventListener('alert', (event) => {
const data = JSON.parse(event.data);
triggerAlert(data.level, data.message);
});
// Default handler for unnamed events
source.onmessage = (event) => {
console.log('Unnamed event:', event.data);
};
// Connection lifecycle
source.onopen = () => console.log('Connected');
source.onerror = () => console.log('Connection error');
Event Schema Design
# Define clear schema for each event type
event_schemas = {
'notification': {
'description': 'User-facing notification',
'fields': {
'id': 'string - unique notification ID',
'type': 'string - info|warning|error',
'title': 'string - notification title',
'body': 'string - notification body',
'timestamp': 'number - Unix timestamp',
'action_url': 'string? - optional link URL',
},
},
'metric': {
'description': 'System metric update',
'fields': {
'name': 'string - metric name',
'value': 'number - current value',
'unit': 'string - measurement unit',
'timestamp': 'number - Unix timestamp',
'tags': 'object? - optional dimension tags',
},
},
'status': {
'description': 'Service status change',
'fields': {
'service': 'string - service name',
'status': 'string - healthy|degraded|down',
'message': 'string? - status details',
'timestamp': 'number - Unix timestamp',
},
},
}
def validate_event(event_type, data):
"""Validate event data against its schema."""
schema = event_schemas.get(event_type)
if not schema:
return False, f"Unknown event type: {event_type}"
required = {k.split('?')[0] for k in schema['fields'] if '?' not in k}
missing = required - set(data.keys())
if missing:
return False, f"Missing fields: {missing}"
return True, "Valid"
Event Router Pattern
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
import asyncio
class EventRouter:
def __init__(self):
self.handlers = {}
def on(self, event_type):
def decorator(func):
if event_type not in self.handlers:
self.handlers[event_type] = []
self.handlers[event_type].append(func)
return func
return decorator
def emit(self, event_type, data, event_id=None):
event = {
'type': event_type,
'data': data,
'id': event_id,
'timestamp': time.time(),
}
handlers = self.handlers.get(event_type, [])
for handler in handlers:
handler(event)
return event
router = EventRouter()
@router.on('user.login')
def handle_login(event):
print(f"User logged in: {event['data']['user']}")
@router.on('user.logout')
def handle_logout(event):
print(f"User logged out: {event['data']['user']}")
@router.on('file.upload')
def handle_upload(event):
print(f"File uploaded: {event['data']['filename']} ({event['data']['size']} bytes)")
# Usage
router.emit('user.login', {'user': 'alice', 'ip': '192.168.1.1'})
router.emit('file.upload', {'filename': 'report.pdf', 'size': 1024000})
Hierarchical Event Types
# Use dot-separated hierarchical event names
# server.event.subtype
events_hierarchy = {
'server': {
'start': 'Server started',
'stop': 'Server stopped',
'error': 'Server error occurred',
},
'user': {
'login': 'User logged in',
'logout': 'User logged out',
'updated': 'User profile updated',
},
'file': {
'uploaded': 'File uploaded',
'downloaded': 'File downloaded',
'deleted': 'File deleted',
},
'system': {
'backup': {
'start': 'Backup started',
'progress': 'Backup progress update',
'complete': 'Backup completed',
'failed': 'Backup failed',
},
'maintenance': {
'start': 'Maintenance window started',
'end': 'Maintenance window ended',
},
},
}
def flatten_hierarchy(hierarchy, prefix=''):
for key, value in hierarchy.items():
full_key = f"{prefix}.{key}" if prefix else key
if isinstance(value, dict):
yield from flatten_hierarchy(value, full_key)
else:
yield full_key, value
for event_path, description in flatten_hierarchy(events_hierarchy):
print(f" {event_path:30s} - {description}")
Expected output:
server.start - Server started
server.stop - Server stopped
server.error - Server error occurred
user.login - User logged in
user.logout - User logged out
user.updated - User profile updated
file.uploaded - File uploaded
file.downloaded - File downloaded
file.deleted - File deleted
system.backup.start - Backup started
system.backup.progress - Backup progress update
system.backup.complete - Backup completed
system.backup.failed - Backup failed
system.maintenance.start - Maintenance window started
system.maintenance.end - Maintenance window ended
Common Mistakes
1. Too Many Event Types
Creating dozens of similar event types (like user.login, user.logged_in, user:login) causes confusion. Keep a small, consistent set.
2. Not Using Namespacing
Event types like 'update' clash when multiple features use the same name. Use namespacing: 'user.update', 'file.update', 'system.update'.
3. Inconsistent Payload Structure
Events of the same type should have the same structure. Do not send {id: 1} sometimes and {userId: 1} other times.
4. Missing Event ID
Without event IDs, the client cannot resume after reconnection. Always include an id field for auditable events.
5. Sending Non-Serializable Data
SSE data must be text. Ensure all event data is JSON-serializable. Do not include functions, Date objects without Serialization, or circular references.
Practice Questions
1. How does the client distinguish between different SSE event types?
The server sets the "event:" field. The client uses source.addEventListener('type', handler) for each type.
2. What happens to unnamed events on the client?
They go to the default onmessage handler. The event.type property will be empty for unnamed events.
3. Why use hierarchical event names?
They organize events by domain, prevent name collisions, and enable wildcard matching in event routers.
4. How do you design event schemas for a real-time dashboard?
Define clear types (notification, metric, status, alert), use consistent field names across events, include timestamps and IDs, and document all schemas.
Challenge
Design an event system for a deployment pipeline with these event types: deploy.started, deploy.progress, deploy.completed, deploy.failed, test.started, test.result, test.completed, notification (for all human-readable alerts). Each event must have a schema with required and optional fields.
FAQ
Mini Project: Event-Driven SSE
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
import threading
class EventDrivenSSE(BaseHTTPRequestHandler):
clients = []
events_queue = []
@classmethod
def broadcast(cls, event_type, data):
event = {
'type': event_type,
'data': data,
'timestamp': time.time(),
}
cls.events_queue.append(event)
message = f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
for client in cls.clients:
try:
client.wfile.write(message.encode())
client.wfile.flush()
except Exception:
cls.clients.remove(client)
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
self.end_headers()
EventDrivenSSE.clients.append(self)
# Send queued events
for event in EventDrivenSSE.events_queue[-10:]:
self.wfile.write(f"event: {event['type']}\ndata: {json.dumps(event['data'])}\n\n".encode())
# Keep connection open
while True:
time.sleep(1)
def event_publisher():
time.sleep(2)
EventDrivenSSE.broadcast('server.start', {'message': 'Server ready'})
time.sleep(5)
EventDrivenSSE.broadcast('user.login', {'user': 'alice'})
time.sleep(3)
EventDrivenSSE.broadcast('file.uploaded', {'filename': 'doc.pdf', 'size': 1024})
threading.Thread(target=event_publisher, daemon=True).start()
server = HTTPServer(('localhost', 3000), EventDrivenSSE)
print("Event-driven SSE on :3000")
server.serve_forever()
What's Next
Now that you understand event types, explore auto-reconnection patterns, then learn about Last-Event-ID for resuming streams.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro