import 'dart:async'; import 'package:meta/meta.dart'; import 'package:synchronized/synchronized.dart' as sync; import '../exceptions.dart'; import '../extensions.dart'; import '../logger.dart'; import '../support/disposable.dart'; import '../types.dart'; mixin EventsEmittable { final events = EventsEmitter(); EventsListener createListener({bool synchronized = false}) => EventsListener(events, synchronized: synchronized); } // Type-safe, multi-listenable, dispose safe event handling // TODO: Move to a separate package class EventsEmitter extends EventsListenable { // suppport for multiple event listeners final streamCtrl = StreamController.broadcast(sync: false); EventsEmitter({ bool listenSynchronized = false, }) : super(synchronized: listenSynchronized) { onDispose(() async => await streamCtrl.close()); } @override EventsEmitter get emitter => this; @internal void emit(T event) { // do nothing if already closed if (streamCtrl.isClosed) { logger.warning('failed to emit event ${event} on a disposed emitter'); return; } // emit the event streamCtrl.add(event); } } // for listening only class EventsListener extends EventsListenable { @override final EventsEmitter emitter; EventsListener( this.emitter, { bool synchronized = false, }) : super( synchronized: synchronized, ); } // ensures all listeners will close on dispose abstract class EventsListenable extends Disposable { // the emitter to listen to EventsEmitter get emitter; final bool synchronized; // keep track of listeners to cancel later final _listeners = >[]; final _syncLock = sync.Lock(); List> get listeners => _listeners; EventsListenable({ required this.synchronized, }) { onDispose(() async { await cancelAll(); }); } Future cancelAll() async { if (_listeners.isNotEmpty) { // Stop listening to all events logger.fine('${objectId} cancelling ${_listeners.length} listeners(s)'); for (final listener in _listeners) { await listener.cancel(); } } } // @override // @mustCallSuper // Future dispose() async { // // mark as disposed // final didDispose = await super.dispose(); // if (didDispose && _listeners.isNotEmpty) { // // Stop listening to all events // logger.fine('${objectId} dispose() cancelling ${_listeners.length} event(s)'); // for (final listener in _listeners) { // await listener.cancel(); // } // } // return didDispose; // } // listens to all events, guaranteed to be cancelled on dispose CancelListenFunc listen(FutureOr Function(T) onEvent) { // FutureOr Function(T) _func = onEvent; if (synchronized) { // ensure `onEvent` will trigger one by one (waits for previous `onEvent` to complete) _func = (event) async { await _syncLock.synchronized(() async { await onEvent(event); }); }; } final listener = emitter.streamCtrl.stream.listen(_func); _listeners.add(listener); // make a cancel func to cancel listening and remove from list in 1 call _cancelFunc() async { await listener.cancel(); _listeners.remove(listener); logger.fine('${objectId} event was cancelled by func'); } return _cancelFunc; } // convenience method to listen & filter a specific event type CancelListenFunc on( FutureOr Function(E) then, { bool Function(E)? filter, }) => listen((event) async { // event must be E if (event is! E) return; // filter must be true (if filter is used) if (filter != null && !filter(event)) return; // cast to E await then(event); }); // waits for a specific event type Future waitFor({ required Duration duration, bool Function(E)? filter, FutureOr Function()? onTimeout, }) async { final completer = Completer(); final _cancelFunc = on( (event) => completer.complete(event), filter: filter, ); try { // wait to complete with timeout return await completer.future.timeout( duration, onTimeout: onTimeout ?? () => throw TimeoutException(), ); // do not catch exceptions and pass it up } finally { // always clean-up listener await _cancelFunc.call(); } } }