dotnet / dotnet/reactive

How to sample the latest value from multiple IGroupedObservable<T> and merge back to a single Observable<T>

Open
#1,646 3 comments 0 reactions 0 assignees View on GitHub
[area] Rx
Dominant language
C#
Stars
7.2k
Forks
798
PR merge metrics
No merged PRs in 30d

Description

Hi!

I have asked the following question on [StackOverflow](https://stackoverflow.com/questions/70003565/reactive-extensions-how-to-sample-the-latest-value-from-igroupedobservable), but no luck so far on getting an answer.

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.

```csharp
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(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(400));
}

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

public string Identifier { get; }

public decimal Price { get; }
}
```

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.