Reactive Extensions: how to sample the latest value from IGroupedObservable

Viewed 152

I have the following unit test where I am trying to group objects from a source observable by their identifier and sample them every 50 ticks. However, the code with the GroupBy, SelectMany -> Sample gives a different output than the expected output. It returns OnNext@280, OnNext@380 and OnCompleted@380.

When I run the test with only the Sample method the output meets the expectation as can be seen in the AssertEqual call.

What am I doing wrong here? I expect that both versions should produce the same output since the identifiers of all products are, in this case, the same.

Ps: I have already checked this question about the same kind of operation and this question on grouped throttling + the reactive source unit tests.

public class ObservableTests : ReactiveTest
    {
        [Fact]
        public void sample_test()
        {
            // Arrange
            var productPrice = new ProductPrice("1", 10);
            var productPrice2 = new ProductPrice("1", 20);
            var productPrice3 = new ProductPrice("1", 30);
            var productPrice4 = new ProductPrice("1", 40);
            var productPrice5 = new ProductPrice("1", 50);
            var productPrice6 = new ProductPrice("1", 60);

            var testScheduler = new TestScheduler();
            var observable = testScheduler.CreateHotObservable(
                OnNext(230, productPrice),
                OnNext(260, productPrice2),
                OnNext(280, productPrice3),
                OnNext(340, productPrice4),
                OnNext(360, productPrice5),
                OnNext(380, productPrice6),
                OnCompleted<ProductPrice>(380));

            // Act

            // WORKS
            //var result = testScheduler.Start(
            //    () =>
            //        observable
            //            .Sample(TimeSpan.FromTicks(50), testScheduler));

            // DOESNT WORK
            var result = testScheduler.Start(
                () =>
                    observable
                        .GroupBy(value => value.Identifier)
                        .SelectMany(groupedObservable => groupedObservable.Sample(TimeSpan.FromTicks(50), testScheduler)));

            result.Messages.AssertEqual(
                OnNext(250, productPrice),
                OnNext(300, productPrice3),
                OnNext(350, productPrice4),
                OnNext(400, productPrice6),
                OnCompleted<ProductPrice>(400));
        }

        private class ProductPrice
        {
            public ProductPrice(string identifier, decimal price)
            {
                this.Identifier = identifier;
                this.Price = price;
            }

            public string Identifier { get; }

            public decimal Price { get; }
        }
0 Answers
Related