rxdart_split.dart - https://github.com/ReactiveX/rxdart/issues/493

rxdart_split.dart

import 'dart:async';

extension SplitExtension<T> on Stream<T> {
  Stream<T> split() {
    final controllers = <MultiStreamController<T>>[];
    var index = -1;
    var done = false;
    StreamSubscription<T> subscription;

    return Stream.multi(
      (controller) {
        if (done) {
          return controller.closeSync();
        }

        final wasEmpty = controllers.isEmpty;
        controllers.add(controller);

        if (wasEmpty) {
          subscription = listen(
            (value) {
              if (controllers.every((s) => s.isPaused)) {
                return;
              }

              while (true) {
                index = (index + 1) % controllers.length;
                final controller = controllers[index];

                if (!controller.isPaused) {
                  controller.addSync(value);
                  return;
                }
              }
            },
            onError: (e, StackTrace st) {
              if (controllers.every((s) => s.isPaused)) {
                return;
              }

              while (true) {
                index = (index + 1) % controllers.length;
                final controller = controllers[index];

                if (!controller.isPaused) {
                  controller.addErrorSync(e, st);
                  return;
                }
              }
            },
            onDone: () {
              done = true;
              controllers.forEach((c) {
                c.onCancel = null;
                c.closeSync();
              });
              controllers.clear();
            },
          );
        }

        controller.onCancel = () {
          controllers.remove(controller);

          if (controllers.isEmpty) {
            subscription?.cancel();
            subscription = null;
            done = true;
          }
        };
      },
      isBroadcast: true,
    );
  }
}