EDIT: @Asti fixed his bug, and I fixed mine based on his answer. Our answers are now largely similar. I have an idea how to do a purely reactive one, if I have time I'll post that later.
Fixed code:
public static IObservable<T> RoundRobin2<T>(this IObservable<T> source)
{
var subscribers = new BehaviorSubject<ImmutableList<IObserver<T>>>(ImmutableList<IObserver<T>>.Empty);
ImmutableList<IObserver<T>> latest = ImmutableList<IObserver<T>>.Empty;
subscribers.Subscribe(l => latest = l);
var shared = source
.Select((v, i) => (v, i))
.WithLatestFrom(subscribers, (t, s) => (t.v, t.i, s))
.Publish()
.RefCount();
return Observable.Create<T>(observer =>
{
subscribers.OnNext(latest.Add(observer));
var dispose = Disposable.Create(() => subscribers.OnNext(latest.Remove(observer)));
var sub = shared
.Where(t => t.i % t.s.Count == t.s.FindIndex(o => o == observer))
.Select(t => t.v)
.Subscribe(observer);
return new CompositeDisposable(dispose, sub);
});
}
Original answer:
I upvoted @Asti's answer, because he's largely correct: Just because you can, doesn't mean you should. And his answer largely works, but it's subject to a bug:
This works fine:
var source = Observable.Range(1, 20).Publish();
var dist = source.RoundRobin();
dist.Subscribe(i => Console.WriteLine($"One sees {i}"));
dist.Take(1).Subscribe(i => Console.WriteLine($"Two sees {i}"));
This doesn't:
var source = Observable.Range(1, 20).Publish();
var dist = source.RoundRobin();
dist.Take(1).Subscribe(i => Console.WriteLine($"One sees {i}"));
dist.Subscribe(i => Console.WriteLine($"Two sees {i}"));
Output is:
One sees 1
Two sees 1
Two sees 2
Two sees 3
Two sees 4
...
I first thought the bug is Halloween related, but now I'm not sure. The .ToArray() in Repeat should take care of that. I also wrote a pure-ish observable implementation which has the same bug. This implementation doesn't guarantee a perfect Round Robin, but that wasn't in the question:
public static IObservable<T> RoundRobin2<T>(this IObservable<T> source)
{
var subscribers = new BehaviorSubject<ImmutableList<IObserver<T>>>(ImmutableList<IObserver<T>>.Empty);
ImmutableList<IObserver<T>> latest = ImmutableList<IObserver<T>>.Empty;
subscribers.Subscribe(l => latest = l);
var shared = source
.Select((v, i) => (v, i))
.WithLatestFrom(subscribers, (t, s) => (t.v, t.i, s))
.Publish()
.RefCount();
return Observable.Create<T>(observer =>
{
subscribers.OnNext(latest.Add(observer));
var dispose = Disposable.Create(() => subscribers.OnNext(latest.Remove(observer)));
var sub = shared
.Where(t => t.i % t.s.Count == t.s.FindIndex(o => o == observer))
.Select(t => t.v)
.Subscribe(observer);
return new CompositeDisposable(dispose, sub);
});
}