Observe values not seen in other observers

Question

I have an observable that emits unique values e.g.

var source=Observable.Range(1,100).Publish();
source.Connect();

I want to observe its values from e.g. two observers but each observer to get notified only for values not seen in other observers.

So if first observer contains the value 10 the second observer should never get notified for the 10 value.

Update

I chose @Asti`s answer cause it was first and although buggy it pointed to the right direction and up-voted @Shlomo's answer. Too bad I cannot accept both answers as @Shlomo answer was more correct and I really appreciate all his help we get on this tag.

Answer 1

This is a simple distributed queue implementation using TPL Dataflow. But with respect to different observers not seeing the same value, there's little chance of it behaving incorrectly. It's not round-robin, but actually has back-pressure semantics.

    public static IObservable<T> Distribute<T>(this IObservable<T> source)
    {
        var buffer = new BufferBlock<T>();
        source.Subscribe(buffer.AsObserver());             
        return Observable.Create<T>(observer =>
            buffer.LinkTo(new ActionBlock<T>(observer.OnNext, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 })
        );
    }

Output

One sees 1
Two sees 2
One sees 3
Two sees 4
One sees 5
One sees 6
One sees 7
One sees 8
One sees 9
One sees 10

I might prefer skipping Rx entirely and just using TPL Dataflow.

Answer 2

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

Answer 3

Observables aren't supposed to behave differently for different observers; a better approach would be to give each observer its own filtered observable.

That being said, if your constraints require that you need this behavior in a single observable - we can use a Round-Robin method.

    public static IEnumerable<T> Repeat<T>(this IEnumerable<T> source)
    {
        for (; ; )
            foreach (var item in source.ToArray())
                yield return item;
    }

    public static IObservable<T> RoundRobin<T>(this IObservable<T> source)
    {
        var subscribers = new List<IObserver<T>>();
        var shared = source
            .Zip(subscribers.Repeat(), (value, observer) => (value, observer))
            .Publish()
            .RefCount();

        return Observable.Create<T>(observer =>
        {
            subscribers.Add(observer);
            var subscription = 
                shared
                .Where(pair => pair.observer == observer)
                .Select(pair => pair.value)
                .Subscribe(observer);

            var dispose = Disposable.Create(() => subscribers.Remove(observer));
            return new CompositeDisposable(subscription, dispose);
        });
    }

Usage:

var source = Observable.Range(1, 100).Publish();
var dist = source.RoundRobin();
dist.Subscribe(i => Console.WriteLine($"One sees {i}"));
dist.Subscribe(i => Console.WriteLine($"Two sees {i}"));

source.Connect();

Result:

One sees 1
Two sees 2
One sees 3
Two sees 4
One sees 5
Two sees 6
One sees 7
Two sees 8
One sees 9
Two sees 10

If you already have a list of observers, the code becomes much simpler.