Switch to new event system (#9)

* `SignalEvents` and `synchronized` mode for `EventsListenable`

* cascade syntax

* clean up

* `RoomEvents`

* `ParticipantEvents`

* all events implemented

* make it compile with M1 macs

* dispose logic

* cleaner connect logic

* ask to publish

* fix initial EngineTrackAddedEvent glitch

* fix unpublishTrack bug

* clean up

* better events docs

* organize event types

* clean up

* cleaner wait logic

* wait for event instead of using Completer

* notifyListeners

lib/src/managers/delay.dart

@@ -1,6 +1,3 @@
-//
-//
-//
 import 'package:async/async.dart';
 
 class CancelableDelayManager {

lib/src/managers/event.dart

@@ -1,19 +1,24 @@
 import 'dart:async';
 
 import 'package:flutter/material.dart';
+import 'package:synchronized/synchronized.dart' as sync;
 
 import '../errors.dart';
-import '../events.dart';
 import '../extensions.dart';
 import '../logger.dart';
 import '../types.dart';
 
 // Type-safe, multi-listenable, dispose safe event handling
+// TODO: Move to a separate package
 
-class EventsEmitter<T extends LiveKitEvent> extends EventsListenable<T> {
+class EventsEmitter<T> extends EventsListenable<T> {
   // suppport for multiple event listeners
   final streamCtrl = StreamController<T>.broadcast(sync: false);
 
+  EventsEmitter({
+    bool listenSynchronized = false,
+  }) : super(synchronized: listenSynchronized);
+
   @override
   EventsEmitter<T> get emitter => this;
 
@@ -25,6 +30,7 @@ class EventsEmitter<T extends LiveKitEvent> extends EventsListenable<T> {
   }
 
   @override
+  @mustCallSuper
   Future<void> dispose() async {
     await streamCtrl.close();
     await super.dispose();
@@ -32,21 +38,31 @@ class EventsEmitter<T extends LiveKitEvent> extends EventsListenable<T> {
 }
 
 // for listening only
-class EventsListener<T extends LiveKitEvent> extends EventsListenable<T> {
+class EventsListener<T> extends EventsListenable<T> {
   @override
   final EventsEmitter<T> emitter;
 
-  EventsListener({
-    required this.emitter,
-  });
+  EventsListener(
+    this.emitter, {
+    bool synchronized = false,
+  }) : super(
+          synchronized: synchronized,
+        );
 }
 
 // ensures all listeners will close on dispose
-abstract class EventsListenable<T extends LiveKitEvent> {
+abstract class EventsListenable<T> {
   // the emitter to listen to
   EventsEmitter<T> get emitter;
+
+  bool synchronized;
   // keep track of listeners to cancel later
   final _listeners = <StreamSubscription<T>>[];
+  final _syncLock = sync.Lock();
+
+  EventsListenable({
+    required this.synchronized,
+  });
 
   @mustCallSuper
   Future<void> dispose() async {
@@ -58,8 +74,19 @@ abstract class EventsListenable<T extends LiveKitEvent> {
   }
 
   // listens to all events, guaranteed to be cancelled on dispose
-  CancelListenFunc listen(Function(T) onEvent) {
-    final listener = emitter.streamCtrl.stream.listen(onEvent);
+  CancelListenFunc listen(FutureOr<void> Function(T) onEvent) {
+    //
+    FutureOr<void> 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
@@ -74,34 +101,34 @@ abstract class EventsListenable<T extends LiveKitEvent> {
 
   // convenience method to listen & filter a specific event type
   CancelListenFunc on<E>(
-    Function(E) then, {
+    FutureOr<void> Function(E) then, {
     bool Function(E)? filter,
   }) =>
-      listen((event) {
+      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 as E)) return;
+        if (filter != null && !filter(event)) return;
         // cast to E
-        then(event as E);
+        await then(event);
       });
 
   // waits for a specific event type
-  Future<void> waitFor<E>({
+  Future<E> waitFor<E>({
     required Duration duration,
     bool Function(E)? filter,
-    FutureOr<void> Function()? onTimeout,
+    FutureOr<E> Function()? onTimeout,
   }) async {
-    final completer = Completer<void>();
+    final completer = Completer<E>();
 
     final _cancelFunc = on<E>(
-      (event) => completer.complete(),
+      (event) => completer.complete(event),
       filter: filter,
     );
 
     try {
       // wait to complete with timeout
-      await completer.future.timeout(
+      return await completer.future.timeout(
         duration,
         onTimeout: onTimeout ?? () => throw TimeoutException(),
       );