Load Splitting?

[shinayser]

Hey Guys!

I was wondering, is it possible to make load balancing with rxdart?
Imagine the following situation: I have one stream producer and many listeners but I want that the data produced are split between the listeners.

Like, listeners are A, B and C:

Emits [0], A receives but not B and C
Then it emits [1], B receives but not A and C
Then it emits [2], C receives but not A and B

[shinayser]

I just elaborated a nice solution for that and created a gist. Perhaps you guys could add it to the library itself?

https://gist.github.com/shinayser/519dedda4626e0c80203bc0673880963

[hoc081098]

Since Dart 2.9, we can use Stream.multi constructor. Here is my implementation:

Dartpad

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,
    );
  }
}

[shinayser]

Your implementation is great, much better then mine actually! I just think that the errors should be emitted to EVERY listener and not only 1, since this can cause some side efects.