Spry with Server-Sent Events (SSE): Real‑Time Updates Without WebSockets

Server‑Sent Events (SSE) provide a simple, efficient way to push real‑time updates from server to client over a single HTTP connection. Unlike WebSockets, SSE is unidirectional (server → client) and built on standard HTTP, making it ideal for notifications, live feeds, and progress updates.

In this tutorial, we'll implement SSE in Spry, covering:

1. SSE fundamentals – Event streams, MIME type, and connection lifecycle
2. Spry implementation – Creating event‑stream endpoints and managing connections
3. Client‑side integration – JavaScript EventSource API and Dart clients
4. Advanced patterns – Authentication, reconnection, and error handling
5. Use‑case comparison – When to choose SSE vs WebSockets vs GraphQL subscriptions

All examples are production‑ready and can be adapted to your own projects.

Why Server‑Sent Events?

• Simple protocol – Plain text over HTTP, easy to debug
• Automatic reconnection – Built‑in client reconnection with last‑event‑ID
• No extra ports – Uses same HTTP(S) port as your API
• Browser support – Native EventSource API in all modern browsers
• Efficient – Lightweight compared to WebSockets for server‑push scenarios

Common use cases:

• Live notifications and alerts
• Progress bars for long‑running operations
• Stock tickers and news feeds
• Dashboard updates (metrics, logs)
• Chat applications (server‑push side)

1. Basic SSE Endpoint in Spry

Let's start with a minimal SSE endpoint that sends periodic updates:

import 'dart:async';
import 'package:spry/spry.dart';

final app = Application();

// Basic SSE endpoint
app.get('/sse', (request) async {
  // Set SSE headers
  request.response.headers
    ..set('Content-Type', 'text/event-stream')
    ..set('Cache-Control', 'no-cache')
    ..set('Connection', 'keep-alive')
    ..set('Access-Control-Allow-Origin', '*');

  // Create a broadcast stream for events
  final controller = StreamController<String>.broadcast();
  final stream = controller.stream;

  // Send initial connection event
  controller.add('event: connected\ndata: {"time": "${DateTime.now()}"}\n\n');

  // Send periodic updates every 2 seconds
  final timer = Timer.periodic(const Duration(seconds: 2), (timer) {
    controller.add('''
event: update
data: {"value": ${DateTime.now().millisecondsSinceEpoch}, "count": ${timer.tick}}
id: ${timer.tick}

''');
  });

  // Clean up when client disconnects
  request.response.onClose = () {
    timer.cancel();
    controller.close();
    print('SSE connection closed');
  };

  // Return the event stream
  return Response.stream(stream);
});

Key points:

• Content-Type: text/event-stream – Required for SSE
• Cache-Control: no-cache – Prevent caching of events
• Connection: keep-alive – Keep HTTP connection alive
• Events follow the format: event: name\ndata: ...\n\n
• Each event is separated by double newline (\n\n)
• Optional id: field enables automatic reconnection

2. Structured Event Streaming with Models

For production applications, structure your events with typed models:

// lib/models/sse_event.dart
abstract class SseEvent {
  final String event;
  final Map<String, dynamic> data;
  final String? id;

  SseEvent(this.event, this.data, {this.id});

  String format() {
    final buffer = StringBuffer();

    if (event.isNotEmpty) {
      buffer.write('event: $event\n');
    }

    buffer.write('data: ${_formatData(data)}\n');

    if (id != null) {
      buffer.write('id: $id\n');
    }

    buffer.write('\n');
    return buffer.toString();
  }

  String _formatData(Map<String, dynamic> data) {
    if (data.isEmpty) return '{}';

    // Convert to JSON, handling nested objects
    final json = StringBuffer('{');
    final entries = data.entries.toList();

    for (int i = 0; i < entries.length; i++) {
      final entry = entries[i];
      final key = entry.key;
      final value = entry.value;

      json.write('"$key": ');

      if (value is String) {
        json.write('"${value.replaceAll('"', '\\"')}"');
      } else if (value is num || value is bool) {
        json.write(value);
      } else if (value is Map) {
        json.write(_formatData(Map<String, dynamic>.from(value)));
      } else if (value is List) {
        json.write('[${value.map((v) => v is Map ? _formatData(Map<String, dynamic>.from(v)) : '"$v"').join(',')}]');
      } else {
        json.write('"$value"');
      }

      if (i < entries.length - 1) {
        json.write(',');
      }
    }

    json.write('}');
    return json.toString();
  }
}

// Specific event types
class ConnectionEvent extends SseEvent {
  ConnectionEvent() : super('connected', {'time': DateTime.now().toIso8601String()});
}

class ProgressEvent extends SseEvent {
  ProgressEvent(double progress, String taskId)
      : super('progress', {
          'progress': progress,
          'taskId': taskId,
          'timestamp': DateTime.now().millisecondsSinceEpoch,
        });
}

class NotificationEvent extends SseEvent {
  NotificationEvent(String title, String message, {String? id})
      : super('notification', {
          'title': title,
          'message': message,
          'type': 'info',
        }, id: id);
}

// Usage in route handler
app.get('/progress/{taskId}', (request) async {
  final taskId = request.params['taskId'];

  request.response.headers
    ..set('Content-Type', 'text/event-stream')
    ..set('Cache-Control', 'no-cache');

  final controller = StreamController<SseEvent>.broadcast();

  // Send initial connection event
  controller.add(ConnectionEvent());

  // Simulate progress updates
  for (int i = 0; i <= 10; i++) {
    await Future.delayed(const Duration(seconds: 1));
    final progress = i / 10.0;
    controller.add(ProgressEvent(progress, taskId));

    if (i == 10) {
      controller.add(NotificationEvent(
        'Task completed',
        'Task $taskId has finished processing',
        id: 'task-$taskId-complete',
      ));
    }
  }

  request.response.onClose = () => controller.close();

  return Response.stream(
    controller.stream.map((event) => event.format()),
  );
});

3. Client‑Side Integration

JavaScript (Browser)

<!DOCTYPE html>
<html>
<head>
  <title>SSE Client Example</title>
</head>
<body>
  <div id="events"></div>
  <script>
    const eventSource = new EventSource('/sse');

    eventSource.addEventListener('connected', (event) => {
      console.log('Connected:', JSON.parse(event.data));
      displayEvent('Connected to server');
    });

    eventSource.addEventListener('update', (event) => {
      const data = JSON.parse(event.data);
      displayEvent(`Update #${data.count}: ${data.value}`);
    });

    eventSource.addEventListener('error', (event) => {
      console.error('SSE error:', event);
      displayEvent('Connection error - attempting reconnect...');
    });

    eventSource.addEventListener('open', () => {
      console.log('Connection reopened');
    });

    function displayEvent(text) {
      const div = document.createElement('div');
      div.textContent = `[${new Date().toLocaleTimeString()}] ${text}`;
      document.getElementById('events').appendChild(div);
    }
  </script>
</body>
</html>

Dart Client

// lib/client/sse_client.dart
import 'dart:async';
import 'dart:convert';

class SseClient {
  final String url;
  late EventSource _eventSource;
  final Map<String, StreamController<dynamic>> _controllers = {};

  SseClient(this.url);

  Future<void> connect() async {
    _eventSource = EventSource(url);

    _eventSource.onOpen.listen((event) {
      print('SSE connection opened');
    });

    _eventSource.onError.listen((error) {
      print('SSE error: $error');
    });

    _eventSource.onMessage.listen((message) {
      // Handle messages without event type (default "message")
      _dispatchEvent('message', message.data);
    });

    // Listen for all event types
    _eventSource.addEventListener('*', (event) {
      _dispatchEvent(event.type, event.data);
    });
  }

  void _dispatchEvent(String type, String data) {
    final controller = _controllers[type];
    if (controller != null) {
      try {
        final parsed = jsonDecode(data);
        controller.add(parsed);
      } catch (e) {
        controller.add(data);
      }
    }
  }

  Stream<dynamic> listen(String eventType) {
    final controller = StreamController<dynamic>.broadcast();
    _controllers[eventType] = controller;
    return controller.stream;
  }

  Future<void> close() async {
    await _eventSource.close();
    for (final controller in _controllers.values) {
      controller.close();
    }
    _controllers.clear();
  }
}

// Usage
final client = SseClient('http://localhost:8080/sse');
await client.connect();

client.listen('progress').listen((data) {
  print('Progress: ${data['progress'] * 100}%');
});

client.listen('notification').listen((data) {
  print('Notification: ${data['title']} - ${data['message']}');
});

4. Advanced SSE Patterns

Authentication with SSE

SSE connections can be authenticated via:

• URL tokens – GET /sse?token=abc123
• Cookies – Standard session cookies
• HTTP headers – Custom headers (requires CORS configuration)

app.get('/sse/secure', (request) async {
  // Verify authentication token
  final token = request.query['token'];
  if (token != 'valid-token') {
    return Response.json({'error': 'Unauthorized'}, status: 401);
  }

  // Set SSE headers
  request.response.headers.set('Content-Type', 'text/event-stream');

  // Create authenticated stream
  final user = await authenticateToken(token);
  final controller = StreamController<String>.broadcast();

  // Send personalized events
  controller.add('''
event: authenticated
data: {"userId": "${user.id}", "name": "${user.name}"}

''');

  // ... rest of SSE implementation

  return Response.stream(controller.stream);
});

Connection Management and Broadcasting

For broadcasting events to multiple clients:

class SseConnectionManager {
  final Map<String, List<StreamController<String>>> _rooms = {};

  void join(String roomId, StreamController<String> controller) {
    _rooms.putIfAbsent(roomId, () => []).add(controller);

    // Remove controller when connection closes
    controller.done.whenComplete(() {
      _rooms[roomId]?.remove(controller);
      if (_rooms[roomId]?.isEmpty ?? false) {
        _rooms.remove(roomId);
      }
    });
  }

  void broadcastToRoom(String roomId, String event) {
    final controllers = _rooms[roomId];
    if (controllers != null) {
      for (final controller in controllers) {
        if (!controller.isClosed) {
          controller.add(event);
        }
      }
    }
  }

  void broadcastToAll(String event) {
    for (final roomControllers in _rooms.values) {
      for (final controller in roomControllers) {
        if (!controller.isClosed) {
          controller.add(event);
        }
      }
    }
  }
}

// Usage in route
final connectionManager = SseConnectionManager();

app.get('/chat/{roomId}', (request) async {
  final roomId = request.params['roomId'];

  request.response.headers.set('Content-Type', 'text/event-stream');

  final controller = StreamController<String>.broadcast();
  connectionManager.join(roomId, controller);

  // Send join notification
  connectionManager.broadcastToRoom(roomId, '''
event: user-joined
data: {"roomId": "$roomId", "time": "${DateTime.now()}"}

''');

  request.response.onClose = () {
    controller.close();
    connectionManager.broadcastToRoom(roomId, '''
event: user-left
data: {"roomId": "$roomId", "time": "${DateTime.now()}"}

''');
  };

  return Response.stream(controller.stream);
});

Reconnection and Event IDs

SSE clients automatically reconnect using the last received event ID:

class ResilientSseEndpoint {
  final Map<String, int> _lastEventIds = {};

  Future<Response> handle(Request request) async {
    final lastEventId = request.headers['last-event-id'];
    final clientId = request.query['clientId'] ?? 'anonymous';

    request.response.headers.set('Content-Type', 'text/event-stream');

    final controller = StreamController<String>.broadcast();

    // Resume from last event if provided
    int counter = _lastEventIds[clientId] ?? 0;
    if (lastEventId != null) {
      counter = int.tryParse(lastEventId) ?? counter;
      print('Resuming from event ID: $counter for client: $clientId');
    }

    // Send missed events if any (simplified example)
    if (counter < _lastEventIds[clientId] ?? 0) {
      for (int i = counter; i <= (_lastEventIds[clientId] ?? 0); i++) {
        controller.add('''
event: catchup
data: {"eventId": $i, "message": "Missed event #$i"}
id: $i

''');
      }
    }

    // Start sending new events
    final timer = Timer.periodic(const Duration(seconds: 5), (timer) {
      counter++;
      _lastEventIds[clientId] = counter;

      controller.add('''
event: heartbeat
data: {"eventId": $counter, "timestamp": ${DateTime.now().millisecondsSinceEpoch}}
id: $counter

''');
    });

    request.response.onClose = () {
      timer.cancel();
      controller.close();
    };

    return Response.stream(controller.stream);
  }
}

5. SSE vs WebSockets vs GraphQL Subscriptions

Choose SSE when:

• You only need server‑push (no client→server messages)
• You want simplicity and minimal overhead
• You need automatic reconnection
• You're building dashboards, notifications, or live feeds

Choose WebSockets when:

• You need bidirectional communication
• You're building chat, collaborative editing, or games
• You need low‑latency, full‑duplex communication

Choose GraphQL Subscriptions when:

• You already use GraphQL
• You need fine‑grained, typed real‑time data
• You want to leverage existing GraphQL tooling

6. Production Considerations

Scaling SSE Connections

SSE connections are long‑lived HTTP connections. Consider:

• Connection limits – Adjust server connection timeouts and limits
• Load balancing – Use sticky sessions or connection‑aware load balancers
• Resource management – Monitor memory usage of long‑lived connections

Monitoring and Debugging

// Add logging middleware for SSE endpoints
app.use((request, next) async {
  if (request.headers['accept']?.contains('text/event-stream') ?? false) {
    print('[SSE] New connection: ${request.uri}');
    final start = DateTime.now();

    final response = await next(request);

    response.onClose = () {
      final duration = DateTime.now().difference(start);
      print('[SSE] Connection closed after $duration');
    };

    return response;
  }

  return await next(request);
});

Security Best Practices

1. Validate origins – Use CORS headers to restrict allowed origins
2. Rate limiting – Implement connection rate limits per client
3. Event sanitization – Validate and sanitize event data before sending
4. HTTPS only – Always use HTTPS in production
5. Connection timeouts – Close idle connections after reasonable timeout

Conclusion

Server‑Sent Events provide a lightweight, standards‑based solution for real‑time server‑push scenarios. With Spry's streaming response support, implementing SSE is straightforward and production‑ready.

SSE excels at:

• Live notifications – Instant updates without polling
• Progress tracking – Long‑running operation status
• Dashboard updates – Real‑time metrics and logs
• News feeds – Chronological event streams

By leveraging SSE where appropriate, you can build responsive applications with minimal complexity while preserving resources for scenarios that truly require bidirectional communication.

Next Steps:

1. Implement SSE in your Spry application for live notifications
2. Add authentication to secure your event streams
3. Monitor connection health with heartbeat events
4. Compare performance with WebSockets for your specific use case
5. Explore advanced patterns like room‑based broadcasting and event persistence

All code examples in this tutorial are production‑ready and available in the spry‑sse‑example GitHub repository.

Happy streaming! 🚀