Welcome to OGeek Q&A Community for programmer and developer-Open, Learning and Share
Welcome To Ask or Share your Answers For Others

Categories

0 votes
257 views
in Technique[技术] by (71.8m points)

c# - How to bind a non-disposable object with each subscription of a cold observable?

Sorry if this question has been asked before, but I can't find a duplicate. Also sorry for asking too many questions lately! I am probably searching for a custom Observable.Using method, that is not restricted to disposable resources. What I have is a cold IObservable that maintains some internal state, for example a Random instance. This instance should be bound not with the IObservable itself, but with each of its subscriptions. Each subscriber should use a different instance of this resource. Take a look for example to the GetRandomNumbers method below:

static IObservable<int> GetRandomNumbers()
{
    var random = new Random(0);
    return Observable
        .Interval(TimeSpan.FromMilliseconds(100))
        .Select(x => random.Next(1, 10))
        .Take(10);
}

This method generates 10 random numbers. The RNG is a Random instance initialized with a constant seed, so it should produce always the same 10 numbers. But alas it doesn't:

var stream = GetRandomNumbers();
Console.WriteLine($"Results A: {String.Join(", ", await stream.ToArray())}");
Console.WriteLine($"Results B: {String.Join(", ", await stream.ToArray())}");

Output:

Results A: 7, 8, 7, 6, 2, 6, 9, 4, 9, 3
Results B: 3, 5, 6, 5, 9, 1, 8, 9, 7, 3

Each subscriber of the stream observable gets a different set of numbers! What happens is that the same Random instance is used by all subscribers. This is not only undesirable, but it also creates the risk of corrupting the internal state of the object, since the Random class is not thread-safe.

My attempt to solve this problem was to use the Using operator, that has a Func<TResource> resourceFactory parameter:

static IObservable<int> GetRandomNumbers()
{
    return Observable.Using(() => new Random(0), random =>
        Observable
            .Interval(TimeSpan.FromMilliseconds(100))
            .Select(x => random.Next(1, 10))
            .Take(10)
    );
}

This would be a perfect solution if the Random was disposable (I tested it with a disposable class and worked as expected), but it's not, and so the code doesn't compile:

The type 'System.Random' cannot be used as type parameter 'TResource' in the generic type or method 'Observable.Using<TResult, TResource>(Func, Func<TResource, IObservable>)'. There is no implicit reference conversion from 'System.Random' to 'System.IDisposable'.

Could you suggest a solution to this problem?

See Question&Answers more detail:os

与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…
Welcome To Ask or Share your Answers For Others

1 Reply

0 votes
by (71.8m points)

Observable.Defer is your friend if you want per-subscriber state.

Try this:

static IObservable<int> GetRandomNumbers() =>
    Observable
        .Defer(() =>
        {
            var random = new Random(0);
            return Observable
                .Interval(TimeSpan.FromMilliseconds(100))
                .Select(x => random.Next(1, 10))
                .Take(10);
        });

My results:

Results A: 7, 8, 7, 6, 2, 6, 9, 4, 9, 3
Results B: 7, 8, 7, 6, 2, 6, 9, 4, 9, 3

与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…
OGeek|极客中国-欢迎来到极客的世界,一个免费开放的程序员编程交流平台!开放,进步,分享!让技术改变生活,让极客改变未来! Welcome to OGeek Q&A Community for programmer and developer-Open, Learning and Share
Click Here to Ask a Question

...