bloomBUILD•SHIP•BLOOM
Bloom ServerRealtime & WebSockets
.md
Realtime Streams

Realtime & WebSockets

bloom_realtime bridges backend mutations to connected clients over WebSockets, featuring zero-allocation channel pub/sub, multi-isolate cluster scaling, and automated query cache invalidation.

BloomChannelHub Server Pub/Sub

BloomChannelHub manages active WebSocket connections, topic subscriptions, broadcast fanout, and automatic dead-socket cleanup:

bin/server.dart
dart
// bin/server.dart - Server-Side Pub/Sub & WebSockets
import 'dart:io';
import 'package:bloom_framework/bloom_server.dart';
import 'package:bloom_security/bloom_security.dart';
import 'package:bloom_realtime/bloom_realtime.dart';

void main() async {
  final router = BloomApiRouter();
  final hub = BloomChannelHub();

  // POST endpoint triggering real-time broadcast to a list's channel
  router.post('/api/lists/:id/todos', (req) async {
    final listId = req.params['id']!;
    final body = await req.json();

    final newTodo = {'id': 101, 'listId': listId, 'title': body['title']};

    // Broadcast to all clients subscribed to this list channel
    hub.broadcast('lists:$listId', {
      'action': 'todo_created',
      'todo': newTodo,
    });

    return BloomResponse.json(newTodo, statusCode: 201);
  });

  // Serve HTTP and WebSocket upgrades with TCP_NODELAY & zero-overhead compressionOff
  final wsServer = BloomWebSocketServer(apiRouter: router);
  wsServer.register('/ws/realtime', (socket, httpRequest) async {
    hub.registerConnection(socket);
  });

  final server = await HttpServer.bind(InternetAddress.anyIPv4, 8080);
  server.listen(wsServer.handleIoRequest);
}

Multi-Core Isolate Cluster (50k+ msgs/s)

Scale real-time throughput past Node.js by distributing incoming TCP connections evenly across all CPU cores with BloomRealtimeCluster:

bin/cluster_server.dart
dart
// bin/cluster_server.dart - 50k+ msgs/s Multi-Core Isolate Cluster
import 'dart:io';
import 'package:bloom_realtime/bloom_realtime.dart';

void main() async {
  // Spawns N isolate workers listening on the same port via kernel shared: true
  final cluster = await BloomRealtimeCluster.bind(
    port: 8080,
    wsPath: '/ws',
    workers: Platform.numberOfProcessors, // Scales across all CPU cores
    compression: CompressionOptions.compressionOff,
    tcpNoDelay: true,
  );

  print('Realtime cluster running with ${cluster.workerCount} workers!');

  // Broadcasts event across all isolate workers and connected sockets
  cluster.broadcast('room:general', {
    'event': 'announcement',
    'text': 'Hello from clustered Bloom!',
  });
}

Kernel TCP_NODELAY & Binary Framing

Bloom Realtime achieves ultra-low latency and minimal memory overhead via three key production optimizations:

TCP_NODELAY

Eliminates Nagle packet buffering to ensure immediate message flushing for realtime chat and live cursor updates.

compressionOff

Disables per-message deflate by default to save 15-25% CPU on small JSON events during high fanout bursts.

Zero-Allocation Hub

Direct pointer set indexing eliminates hash-map queries and heap list allocations in the broadcast hot-path.

Server vs. Client Barrel Split

package:bloom_realtime/bloom_realtime.dart

Pure Dart, server-safe barrel export with zero Flutter framework dependencies. Used inside server entrypoints and backend route handlers.

package:bloom_realtime/bloom_realtime_client.dart

Client barrel adding the RealtimeQueryBridge and BloomData integration for Flutter applications.

Live Query Cache Invalidation

Connect channel broadcasts directly to BloomData.invalidateQueries(...) on the client. When a backend event occurs, cached queries automatically refetch in the background:

lib/client.dart
dart
// client.dart - Flutter / Client-Side Live Invalidation
import 'package:bloom_framework/bloom_data.dart';
import 'package:bloom_realtime/bloom_realtime.dart';

void main() async {
  // 1. Initialize Reconnecting Client
  final realtime = BloomRealtimeClient(
    uri: Uri.parse('ws://localhost:8080/ws/realtime'),
  );
  await realtime.connect();

  const listId = '42';
  final queryKey = ['lists', listId, 'todos'];

  // 2. Wire broadcast events to automatically invalidate BloomData query cache
  realtime.invalidateQueriesOnBroadcast(
    channel: 'lists:$listId',
    key: queryKey,
  );

  // 3. Define the reactive BloomQuery
  final todosQuery = query<List<dynamic>>(
    key: queryKey,
    fetch: () => fetchTodosFromApi(listId),
  );
}

Presence Tracking

BloomPresenceTracker tracks real-time join/leave states with initial state snapshots and departure events when sockets disconnect.

Previous: Admin PanelNext: Caching