Event-driven architecture using Flutter Hooks, Rust, and StreamController

[vincent-herlemont]

I am implementing an event-driven architecture for a Flutter application, which consists of three parts:

1. An embedded backend in Rust with flutter_rust_bridge using feature/stream.
2. A class event in Dart which broadcasts all events with StreamController.broadcast().
3. Each component which should react to the events uses the custom hooks useEventStream<T>

Schema example:

graph BT
A[Rust Backend] -->|Events| B[Flutter Rust Bridge]
B --> C[Event class StreamController.broadcast]
C --> D1[Component 1 <br> useEventStream T1]
C --> D2[Component 2 <br> useEventStream T2]
C --> D3[Component 3 <br> useEventStream T3]
D1 -->|Cast to T1| E1[State and UI Updates]
D2 -->|Cast to T2| E2[State and UI Updates]
D3 -->|Cast to T3| E3[State and UI Updates]

class A RustBackend;
class B FlutterRustBridge;
class C EventClass;
class D1,D2,D3 Components;
class E1,E2,E3 UIUpdates;

classDef RustBackend fill:#f9d142,stroke:#333,stroke-width:2px;
classDef FlutterRustBridge fill:#9fdaf3,stroke:#333,stroke-width:2px;
classDef EventClass fill:#96c9a8,stroke:#333,stroke-width:2px;
classDef Components fill:#fea8c2,stroke:#333,stroke-width:2px;
classDef UIUpdates fill:#e9b3c3,stroke:#333,stroke-width:2px;

Here is the part 2 which receives events from Rust:

class Event {

  static final StreamController<FEvent> _global_event_stream = StreamController.broadcast();

  static init() async {
    final Stream<FEvent> stream = api.eventSetup();
    _global_event_stream.addStream(stream);
  }

  static StreamSubscription<FEvent> stream<T>(StreamController<T> streamController, [bool Function(T)? callback]) {
    callback ??= (e) {return true;};
    StreamSubscription<FEvent> streamSubscription = _global_event_stream.stream.listen((event) {
      if (event is T) {
        final _event = event as T;
        if (callback!(_event)) {
          streamController.add(_event);
        }
      }
    });
    return streamSubscription;
  }
}

Here is the custom hooks useEventStream<T>:

StreamSubscription<FEvent> useEventStream<T>({
  required StreamController<T> controller,
  bool Function(T)? callback,
}) {
  return use(
    _EventStreamHook(
      controller: controller,
      callback: callback
    )
  );
}

class _EventStreamHook<T> extends Hook<StreamSubscription<FEvent>> {
  const _EventStreamHook({
    required this.controller,
    this.callback
  });

  final StreamController<T> controller;
  final bool Function(T)? callback;

  @override
  HookState<StreamSubscription<FEvent>, Hook<StreamSubscription<FEvent>>> createState() =>
    _EventStreamHookState<T>();
}

class _EventStreamHookState<T>
    extends HookState<StreamSubscription<FEvent>, _EventStreamHook<T>> {

  late final StreamSubscription<FEvent>? _streamSubscription;

  @override
  void initHook() {
    _streamSubscription = Event.stream<T>(hook.controller, hook.callback);
  }

  @override
  build(BuildContext context) {
    return _streamSubscription!;
  }

  @override
  void dispose() {
    _streamSubscription!.cancel();
  }

  @override
  String? get debugLabel => "useEventStream";
}

And in each component, we have:

class Component1 extends HookWidget {
  const Component1({super.key});

  @override
  Widget build(BuildContext context) {

    final controller = useStreamController<MyEvent>();
    useEventStream<TE>(controller: controller, callback: callback);
    final stream = useStream(controller.stream, preserveState: true);

    if (stream.hasData) {
      final MyEvent eventData = stream.data!;
      asyncSnapshotData.setEvent(eventData);
    }
    ...
  }
}

I have a few questions regarding this architecture:

1. Does this type of architecture seem viable?
2. Is Flutter Hooks used correctly, or should Riverpod be used instead?

👉 It is worth noting that the event stream could be significant, with several hundred events per second during peak times.

Any feedback or suggestions for improvements are welcome :)

[kulame]

Why don't you use Riverpod's StreamProvider to handle Rust's stream input?