Subtypes filtering from Supertype stream.

main.dart

///
/// [Author] Alex (https://github.com/AlexV525)
/// [Date] 2021/10/22 18:38
///
import 'dart:async';

import 'package:meta/meta.dart';

void main() {
  _initObservers();
  addListener((SubTypeA data) {
    print('Listener for A');
  });
  addListener((SubTypeB data) {
    print('Listener for B');
  });
  addListener((SubTypeC data) {
    print('Listener for C');
  });
  _streamController.add(SubTypeC());
  _streamController.add(SubTypeA());
  _streamController.add(SubTypeB());
}

@immutable
class _Listener<T extends SuperType> {
  const _Listener(this._listener);

  final Listener<T> _listener;

  Type get type => T;

  void call(T data) => _listener(data);

  @override
  bool operator ==(Object other) =>
      identical(this, other) ||
      other is _Listener &&
          runtimeType == other.runtimeType &&
          _listener == other._listener &&
          type == other.type;

  @override
  int get hashCode => runtimeType.hashCode ^ _listener.hashCode ^ type.hashCode;
}

typedef Listener<T extends SuperType> = void Function(T data);

final List<_Listener<SuperType>> _listeners = <_Listener<SuperType>>[];

void addListener<T extends SuperType>(Listener<T> listener) {
  _listeners.add(_Listener<T>(listener));
}

void removeListener<T extends SuperType>(Listener<T> listener) {
  _listeners.remove(_Listener<T>(listener));
}

StreamSubscription<SuperType>? responses;

late final StreamController<SuperType> _streamController =
      StreamController<SuperType>.broadcast();

void _initObservers() {
  responses = _streamController.stream.listen((SuperType data) {
    for (final _Listener<SuperType> listener in _listeners) {
      if (listener.type == data.runtimeType) {
        listener(data);
      }
    }
  });
}

abstract class SuperType {
  void printRuntimeType() => print(runtimeType);
}

class SubTypeA extends SuperType {}

class SubTypeB extends SuperType {}

class SubTypeC extends SuperType {}