R3Async/src/R3Async.Tests/Operators/RefCountTest.cs at 546fe270d88fb0b157abd9e9377b498e61a9a4cf · fedeAlterio/R3Async · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
using R3Async.Subjects;
using Shouldly;

namespace R3Async.Tests.Operators;

public class RefCountTest
{
    [Fact]
    public async Task FirstSubscriberConnects_SecondShares_DisposeLastDisconnects()
    {
        var source = Subject.Create<int>();
        var connectable = source.Values.Publish();

        var results1 = new List<int>();
        var results2 = new List<int>();

        var refCounted = connectable.RefCount();

        await using var sub1 = await refCounted.SubscribeAsync(async (x, token) => results1.Add(x), CancellationToken.None);
        await using var sub2 = await refCounted.SubscribeAsync(async (x, token) => results2.Add(x), CancellationToken.None);

        await source.OnNextAsync(1, CancellationToken.None);
        await source.OnNextAsync(2, CancellationToken.None);

        results1.ShouldBe(new[] { 1, 2 });
        results2.ShouldBe(new[] { 1, 2 });

        await sub1.DisposeAsync();
        await sub2.DisposeAsync();

        // after all subscribers disposed, further values should not be delivered
        await source.OnNextAsync(3, CancellationToken.None);

        results1.ShouldBe(new[] { 1, 2 });
        results2.ShouldBe(new[] { 1, 2 });
    }

    [Fact]
    public async Task LateSubscriberGetsOnlyFutureValues()
    {
        var source = Subject.Create<int>();
        var connectable = source.Values.Publish();
        var refCounted = connectable.RefCount();

        var results1 = new List<int>();
        var results2 = new List<int>();

        await using var sub1 = await refCounted.SubscribeAsync(async (x, token) => results1.Add(x), CancellationToken.None);

        await source.OnNextAsync(1, CancellationToken.None);

        await using var sub2 = await refCounted.SubscribeAsync(async (x, token) => results2.Add(x), CancellationToken.None);

        await source.OnNextAsync(2, CancellationToken.None);

        results1.ShouldBe(new[] { 1, 2 });
        results2.ShouldBe(new[] { 2 });
    }

    [Fact]
    public async Task SourceCompletionPropagatesAndDisposesConnection()
    {
        var source = Subject.Create<int>();
        var connectable = source.Values.Publish();
        var refCounted = connectable.RefCount();

        var results = new List<int>();
        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);

        await using var sub = await refCounted.SubscribeAsync(
            async (x, token) => results.Add(x),
            async (ex, token) => { },
            async result => completedTcs.TrySetResult(result),
            CancellationToken.None);

        await source.OnNextAsync(1, CancellationToken.None);
        await source.OnCompletedAsync(Result.Success);

        var result = await completedTcs.Task;
        result.IsSuccess.ShouldBeTrue();
        results.ShouldBe(new[] { 1 });
    }

    [Fact]
    public async Task RefCountWithStatelessBehavior_SubscribeAfterReset_ReceivesInitialValue()
    {
        var subject = Subject.Create<int>();

        var refCounted = subject.Values.Share(10, ShareConfig.ResetOnCompletionAndRefCountZero);

        var results1 = new List<int>();

        await using (await refCounted.SubscribeAsync(async (x, token) => results1.Add(x), CancellationToken.None))
        {
            await subject.OnNextAsync(20, CancellationToken.None);
        }

        var results2 = new List<int>();
        await using var sub2 = await refCounted.SubscribeAsync(async (x, token) => results2.Add(x), CancellationToken.None);

        results1.ShouldBe(new[] { 10, 20 });
        results2.ShouldBe(new[] { 10 });
    }
}