Broadcasting

Introduction#

Magic provides a Laravel Echo-equivalent broadcasting system that lets your Flutter app receive real-time events over WebSocket connections. The Echo facade mirrors the Laravel Echo JavaScript client API — if you know how to use Laravel Echo, you already know how to use this.

The broadcasting system is:

• Pusher-compatible: Works with Laravel Reverb, Soketi, and any Pusher-protocol server out of the box.
• Resilient: Automatic reconnection with exponential backoff and application-level heartbeat.
• Extensible: Register custom drivers via BroadcastManager.extend().
• Testable: Echo.fake() swaps the real driver for an in-memory fake with assertion helpers.

The BroadcastServiceProvider is not auto-registered. You must add it explicitly to your providers list.

Configuration#

Copy defaultBroadcastingConfig into your application config and register BroadcastServiceProvider:

// lib/config/broadcasting.dart
import 'package:magic/magic.dart';

final Map<String, dynamic> broadcastingConfig = {
  'broadcasting': {
    'default': Env.get('BROADCAST_CONNECTION', 'null'),

    'connections': {
      'reverb': {
        'driver': 'reverb',
        'host': Env.get('REVERB_HOST', 'localhost'),
        'port': int.parse(Env.get('REVERB_PORT', '8080')!),
        'scheme': Env.get('REVERB_SCHEME', 'ws'),
        'app_key': Env.get('REVERB_APP_KEY', ''),
        'auth_endpoint': '/broadcasting/auth',
        'reconnect': true,
        'max_reconnect_delay': 30000,
        'activity_timeout': 120,
      },
      'null': {'driver': 'null'},
    },
  },
};

Register the provider in Magic.init():

await Magic.init(
  configFactories: [
    () => appConfig,
    () => broadcastingConfig,
  ],
  configs: [
    {
      'app': {
        'providers': [
          (app) => BroadcastServiceProvider(app),
        ],
      },
    },
  ],
);

Connection Options#

Environment Variables#

BROADCAST_CONNECTION=reverb
REVERB_HOST=localhost
REVERB_PORT=8080
REVERB_SCHEME=ws
REVERB_APP_KEY=your-app-key

Echo Facade#

The Echo facade provides static access to the broadcasting system, proxying all calls to the bound BroadcastManager.

API Reference#

Excluding the Sender#

Laravel's broadcast($event)->toOthers() skips the socket that made the request, but only when the request carries the X-Socket-ID header, which Laravel Echo adds to its axios calls. BroadcastServiceProvider registers a SocketIdInterceptor on the network driver that sets the header from Echo.socketId on every request, so toOthers() works with no extra code. While the socket is down the id is null and the request is sent unchanged; the header is never empty.

The interceptor is skipped when broadcasting.default is 'null' and when no network driver is bound (add NetworkServiceProvider to the providers list to receive it).

Channels#

Public Channels#

Public channels require no authentication. Any connected client may subscribe.

// Subscribe and listen for a specific event
Echo.channel('orders').listen('OrderShipped', (event) {
  final orderId = event.data['id'];
  print('Order $orderId has shipped!');
});

// Fluent chaining for multiple events on the same channel
Echo.channel('orders')
  .listen('OrderShipped', onShipped)
  .listen('OrderCancelled', onCancelled);

// Stop listening to a specific event
Echo.channel('orders').stopListening('OrderShipped');

// Leave the channel entirely
Echo.leave('orders');

The BroadcastEvent envelope provides full context for each received message:

Every frame the server sends reaches your listener, including one whose payload repeats an earlier frame exactly: a "something changed, refetch" event usually does. The server can also send the same event twice, when a queued broadcast job is retried after a timeout. Write handlers so a second delivery is harmless (refetch, or upsert by an id the payload carries) rather than counting or toasting once per frame.

Private Channels#

Private channels perform an HTTP auth handshake at auth_endpoint before subscribing. The driver sends the socket_id and channel_name to your server, which validates the request and returns an auth token.

// Subscribe to a private channel (driver adds 'private-' prefix automatically)
Echo.private('user.${Auth.user<User>()!.id}')
  .listen('ProfileUpdated', (event) {
    print('Profile updated: ${event.data}');
  });

// Convenience shorthand
Echo.listen('user.1', 'ProfileUpdated', (event) {
  print(event.data);
});

On the Laravel server side, the channel authorization lives in routes/channels.php:

Broadcast::channel('user.{id}', function ($user, $id) {
    return (int) $user->id === (int) $id;
});

Presence Channels#

Presence channels extend private channels with real-time member tracking. The server returns member data on subscription success, and the driver emits onJoin/onLeave streams as membership changes.

final channel = Echo.join('room.1');

// Members currently in the channel
print('Online: ${channel.members.length}');

// React to member join/leave
channel.onJoin.listen((member) {
  print('${member['name']} joined the room');
});

channel.onLeave.listen((member) {
  print('${member['name']} left the room');
});

// Presence channels also support event listening
channel.listen('MessagePosted', (event) {
  print('New message: ${event.data['body']}');
});

BroadcastPresenceChannel API:

Interceptors#

BroadcastInterceptor hooks into the driver message pipeline — identical in spirit to MagicNetworkInterceptor in the HTTP layer. All three hook methods have pass-through default implementations; subclass only what you need.

Creating a Custom Interceptor#

// lib/app/broadcasting/logging_broadcast_interceptor.dart
import 'package:magic/magic.dart';

class LoggingBroadcastInterceptor extends BroadcastInterceptor {
  @override
  BroadcastEvent onReceive(BroadcastEvent event) {
    Log.debug('Broadcast received', {
      'event': event.event,
      'channel': event.channel,
      'data': event.data,
    });
    return event;
  }

  @override
  dynamic onError(dynamic error) {
    Log.error('Broadcast error', {'error': error.toString()});
    return error;
  }
}

Registering Interceptors#

Register interceptors in a Service Provider's boot() phase, after the connection is established:

class BroadcastingServiceProvider extends ServiceProvider {
  BroadcastingServiceProvider(super.app);

  @override
  Future<void> boot() async {
    Echo.addInterceptor(LoggingBroadcastInterceptor());
  }
}

Custom Drivers#

Implementing a Custom Driver#

Implement the BroadcastDriver abstract class:

// lib/app/broadcasting/pusher_broadcast_driver.dart
import 'package:magic/magic.dart';

class PusherBroadcastDriver implements BroadcastDriver {
  PusherBroadcastDriver(this._config);

  final Map<String, dynamic> _config;

  @override
  Future<void> connect() async {
    // Establish connection to Pusher.
  }

  @override
  Future<void> disconnect() async {
    // Close the connection.
  }

  @override
  String? get socketId => /* ... */;

  @override
  bool get isConnected => /* ... */;

  @override
  Stream<BroadcastConnectionState> get connectionState => /* ... */;

  @override
  Stream<void> get onReconnect => /* ... */;

  @override
  BroadcastChannel channel(String name) => /* ... */;

  @override
  BroadcastChannel private(String name) => /* ... */;

  @override
  BroadcastPresenceChannel join(String name) => /* ... */;

  @override
  void leave(String name) { /* ... */ }

  @override
  void addInterceptor(BroadcastInterceptor interceptor) { /* ... */ }
}

Registering the Custom Driver#

Register your driver via BroadcastManager.extend() in a Service Provider's boot() phase:

class AppServiceProvider extends ServiceProvider {
  AppServiceProvider(super.app);

  @override
  Future<void> boot() async {
    BroadcastManager.extend('pusher', (config) => PusherBroadcastDriver(config));
  }
}

Then reference the driver name in config:

'connections': {
  'pusher': {
    'driver': 'pusher',
    'app_key': Env.get('PUSHER_APP_KEY'),
    'cluster': Env.get('PUSHER_CLUSTER', 'mt1'),
  },
},

This follows the same extend() pattern used by Auth.manager.extend(...) for custom auth guards and LogManager.extend(...) for custom log drivers.

Testing#

Using FakeBroadcastManager#

Echo.fake() swaps the real BroadcastManager binding with an in-memory FakeBroadcastManager. All channel operations are recorded but no WebSocket connection is opened.

import 'package:flutter_test/flutter_test.dart';
import 'package:magic/magic.dart';
import 'package:magic/testing.dart';

void main() {
  MagicTest.init();

  test('subscribes to orders channel on init', () async {
    final fake = Echo.fake();

    // Exercise code under test
    final controller = OrderController();
    await controller.onInit();

    // Assert
    fake.assertSubscribed('orders');
    fake.assertConnected();
  });
}

Always call Echo.unfake() in tearDown — or use MagicTest.init() which resets the container automatically.

Assertion Helpers#

FakeBroadcastManager exposes assertion methods that throw AssertionError with descriptive messages on failure:

Access the underlying FakeBroadcastDriver via fake.driver for low-level inspection:

final fake = Echo.fake();

Echo.channel('orders');
Echo.private('user.1');

expect(fake.driver.subscribedChannels, contains('orders'));
expect(fake.driver.subscribedChannels, contains('private-user.1'));
expect(fake.driver.isConnected, isFalse);

Every fake channel starts with isSubscribed false. Stand in for the server's confirmation, or a drop, with fake.driver.setSubscribed(name, value), which takes the fully-qualified name and emits on the channel's onSubscriptionChanged when the value changes:

fake.driver.setSubscribed('private-user.1', true);
expect(Echo.private('user.1').isSubscribed, isTrue);

Simulate received events by publishing directly to a channel in tests:

// Inject a fake event into the channel stream
final channel = Echo.channel('orders') as _FakeBroadcastChannel;
// ... or use fake.driver to inspect subscriptions and simulate state changes

Connection#

Connection Lifecycle#

BroadcastServiceProvider starts the default connection during boot and does not wait for it. A server that is down or slow never holds or fails the app's launch: the failure is logged with Log.warning() and the driver keeps retrying with the backoff below, including when the server refused the socket outright. Code that needs the socket awaits Echo.connect(), which joins the attempt already in flight.

BroadcastConnectionState tracks the lifecycle of the WebSocket connection:

Subscribe to Echo.connectionState to react to transitions in your UI:

Echo.connectionState.listen((state) {
  switch (state) {
    case BroadcastConnectionState.connected:
      showOnlineBadge();
    case BroadcastConnectionState.reconnecting:
      showReconnectingBanner();
    case BroadcastConnectionState.disconnected:
      showOfflineBanner();
    default:
      break;
  }
});

Every handshake subscribes every channel the app holds (public, private, and presence) with the new socket id, as pusher-js does, whichever path opened the socket: a timer retry, an explicit Echo.connect() after a drop, or the first connect. A channel created while the socket is down is subscribed once, by the next handshake. Private and presence channels re-authenticate via the auth_endpoint, since the signature covers the socket id. onReconnect emits after each reconnect handshake, once every channel has been sent its subscribe (a private or presence channel after its first auth attempt), and never for the first connection. It does not mean the server confirmed anything: a channel whose auth is still being retried receives nothing yet, so a refetch on onReconnect can miss events published until the channel is confirmed. Refetch on the channel's own confirmation instead (see Subscription State), as AuthChannelSubscription does:

final channel = Echo.private('teams.1');
channel.onSubscriptionChanged.where((subscribed) => subscribed).listen((_) {
  // Events flow again from here; fetch what was missed while they did not.
  refetchIncidents();
});

If a private/presence channel auth fails, the error is routed through the interceptor onError() chain and logged, as a warning while it is being retried and as an error when it is not. Other channels continue subscribing; one failure does not block the rest.

Subscription State

A connected driver does not mean a subscribed channel: a private channel still waits on its auth call, and an auth refusal leaves it unsubscribed. Each channel reports the server's confirmation itself:

final channel = Echo.private('team.1');

channel.isSubscribed; // true once the server sent subscription_succeeded
channel.onSubscriptionChanged.listen((subscribed) {
  // true on subscription_succeeded, false on a drop or a leave
  subscribed ? stopPolling() : startPolling();
});

onSubscriptionChanged emits only on a change and never replays, so read isSubscribed when you attach. The null driver's channels are never subscribed.

Auth Failures

The default auth call throws a BroadcastAuthException carrying the statusCode when the auth_endpoint answers outside 2xx (0 when no response arrived). A transient failure, no response or a 5xx, is retried with the reconnect backoff for as long as the socket lasts; a 401 or 403 is logged and not retried, since retrying cannot change the answer. An answer that arrives after the socket changed, or after the channel was left, is dropped: its signature covers the old socket id, and a left channel must not be joined again.

Reconnection Backoff and Jitter#

ReverbBroadcastDriver reconnects automatically with exponential backoff and random jitter. The formula is:

base  = min(500ms x 2^attempt, max_reconnect_delay)
delay = base + random(0 .. base x 0.3)

• The base delay starts at 500 ms and doubles each attempt, capped at max_reconnect_delay (default 30,000 ms).
• Up to 30% random jitter is added on top of the base delay. This prevents a thundering herd: when a server restarts all clients do not hit it at the exact same moment.
• Set reconnect: false in config to disable auto-reconnect entirely.

You can lower the cap to keep reconnects aggressive under controlled network conditions:

'connections': {
  'reverb': {
    'driver': 'reverb',
    // ... other options
    'reconnect': true,
    'max_reconnect_delay': 10000, // cap at 10 seconds instead of the default 30s
  },
},

The WebSocket close code the server sends determines which strategy is applied, following the Pusher protocol:

A close without a code is the common case: a restarting Reverb drops its connections without a close frame. A pusher:error frame on an open socket never reconnects; it is routed through the interceptor onError() chain only. Reverb sends one (code 4200) for any exception it hits while handling a message and keeps the socket open, so reconnecting on it would tear down a healthy connection.

Activity Monitor and Heartbeat#

ReverbBroadcastDriver implements client-side activity monitoring per the Pusher protocol to detect silent connection loss (server crashes, network partitions, or deployments that close the connection without sending a WebSocket close frame).

How it works:

1. After the pusher:connection_established handshake, the driver reads the activity_timeout from the server payload (falling back to the activity_timeout config key, default 120 seconds).
2. An inactivity timer is started. If no message arrives within activity_timeout seconds, the driver sends a pusher:ping frame to the server.
3. If no pusher:pong response arrives within 30 seconds (the fixed pongTimeout), the connection is declared dead and dropped as a close without a code: the state moves to reconnecting, every channel reports unsubscribed, and the standard reconnection flow with exponential backoff begins. The driver does not wait for the dead socket to finish closing, which it may never do.
4. The timer resets on any inbound message, not only pong responses.
5. All timers are cancelled on disconnect() and at the start of each reconnection cycle to prevent stale-timer interference.

The driver also responds to server-initiated pusher:ping frames automatically (sends pusher:pong), satisfying the server-side keepalive requirement.

Connection Timeout#

The connection_timeout config key (default: 15 seconds) controls how long ReverbBroadcastDriver waits for the pusher:connection_established handshake after the WebSocket upgrades. If the server does not complete the handshake within that window:

• The socket is closed.
• A reconnect is scheduled (subject to backoff and the reconnect config flag).
• A TimeoutException is thrown from Echo.connect() so callers can surface an error state.

ReverbBroadcastDriver.connect() is idempotent, so calling it again is always safe: it returns at once when already connected, joins an attempt already in flight (another connect() or a timer-driven retry), and when a reconnect is armed (after a drop, a failed retry, or this timeout) it cancels the armed retry and reconnects now, resubscribing every channel and firing onReconnect as the timer would. A second call never opens a second socket. With reconnect: false, or after a fatal close, a later connect() opens a fresh socket and its handshake still resubscribes every channel the app holds.

'connections': {
  'reverb': {
    'driver': 'reverb',
    // ... other options
    'connection_timeout': 15, // seconds; increase on high-latency links
  },
},

Auth-Scoped Subscriptions with AuthChannelSubscription#

AuthChannelSubscription keeps a single private channel subscription in sync with a caller-supplied name, re-read on every sync() call. It is the seam behind a channel whose name depends on the signed-in user or team: which channel is currently subscribed, leaving the old one and standing up the replacement when the name changes, and firing onReconnect once the channel is confirmed again after a drop so a caller can refetch whatever the socket missed while it was down (Reverb does not replay).

late final subscription = AuthChannelSubscription(
  channelName: () {
    final teamId = Auth.user<User>()?.teamId;
    return teamId == null ? null : 'teams.$teamId';
  },
  listeners: {
    'incident.opened': (event) => refetchIncidents(),
  },
  onReconnect: refetchIncidents,
);

// Wire it to whatever changes the channel name, typically the guard's
// own state notifier:
Auth.stateNotifier.addListener(subscription.sync);

// Reconcile once at startup too, since the listener only fires on a change.
subscription.sync();

sync() is serialised: a call arriving while another is in flight defers and re-runs once more after the current one settles, rather than risking two live subscriptions across an await. When channelName() still answers the name it is already subscribed to, the channel is never registered again (the driver holds it and subscribes it on every handshake); if the connection is down, sync() only calls Echo.connect(), which a driver configured with reconnect: false needs to come back at all. A name change leaves the old channel by its fully-qualified (prefixed) name, calls Echo.connect() when the connection is not live, then subscribes to the new channel and wires every entry of listeners. That connect is safe during a pending reconnect because the Reverb driver's connect() is idempotent (see Connection Timeout).

onReconnect fires each time the server confirms the channel again after losing it (onSubscriptionChanged turning true), not when the socket comes back: after a Reverb restart the socket is up well before a channel whose auth is still being retried, and a refetch any earlier would miss what is published in between. It also fires on the first confirmation when the channel was registered while the connect failed, since that can come long after the screen loaded; on a live socket the first confirmation does not refetch. If the connect fails, the channel is still registered and its listeners attached, and the driver subscribes it on its next handshake. dispose() cancels only the reconnect-listening subscriptions; it does not leave the channel or disconnect, which stay live until the next sync() resolves a null channel name. A null channel name disconnects the whole default connection via Echo.disconnect(), not just this channel, dropping any other channel the app subscribed elsewhere through Echo: deliberate, since a signed-out app has no business staying on the socket.

@override
void onClose() {
  Auth.stateNotifier.removeListener(subscription.sync);
  subscription.dispose();
}

One Subscription Per Alias with BroadcastListeners#

AuthChannelSubscription keeps ONE channel's name in sync with auth state; it says nothing about what happens when two controllers both want events off that same channel. Left to Echo directly, a second controller's Echo.private(name).listen(event, ...) REPLACES the first controller's handler for that event rather than adding a second one, and either controller closing would tear the channel down under the other. BroadcastListeners is the process-wide registry that fixes both: exactly one AuthChannelSubscription per declared alias, with every handler registered against an event fanned out from one stable callback.

Declare the channel an alias resolves to once, typically from a provider's boot():

BroadcastListeners.channel(
  'team',
  () => currentTeamId == null ? null : 'teams.$currentTeamId',
  onReconnect: refetchAll,
);
await BroadcastListeners.sync(); // wired to Auth.stateNotifier, same as AuthChannelSubscription

A controller then mixes in ListensToBroadcasts and declares what it wants, keyed '<alias>:<event>':

class MonitorController extends MagicController with ListensToBroadcasts {
  @override
  Map<String, void Function(BroadcastEvent)> get listeners => {
    'team:check.recorded': (event) => MonitorRepository.instance.patch(
      event.data['monitor_id'] as String,
      {'last_status': event.data['status']},
    ),
  };
}

startListening() runs automatically from onInit() for any controller mounted in a view; a controller read only through .instance and never mounted must call startListening() from its own constructor. onClose() removes every token this mixin registered and leaves any other controller listening on the same alias untouched: the underlying subscription and channel belong to BroadcastListeners, never to one controller.

The aliases share the default Echo connection. A null name on one alias leaves only that alias's channel, and BroadcastListeners.sync() disconnects once no alias resolves a channel any more, so a user leaving their last team keeps receiving on the alias that still names them. Each alias's subscription is built with disconnectOnTeardown: false; a standalone AuthChannelSubscription keeps the default true and disconnects on its own null name.

Testing Broadcasting#

Echo.fake() replaces the real BroadcastManager with an in-memory FakeBroadcastManager. No WebSocket connection is opened; all channel operations are recorded for assertion.

import 'package:flutter_test/flutter_test.dart';
import 'package:magic/magic.dart';
import 'package:magic/testing.dart';

void main() {
  MagicTest.init();

  test('subscribes to the orders channel on init', () async {
    final fake = Echo.fake();

    OrderController().onInit(); // onInit() is synchronous (void); do not await it

    // channel() records the subscription; the fake stays disconnected until
    // Echo.connect() is awaited.
    fake.assertSubscribed('orders');
    fake.assertDisconnected();

    await Echo.connect();
    fake.assertConnected();
  });
}

The full assertion API (assertSubscribed, assertNotSubscribed, assertConnected, assertDisconnected, assertInterceptorAdded, reset, and low-level fake.driver access) is covered in the Testing section above and in doc/testing/facades.md.