dotnet / dotnet/reactive

Infinite loop from Buffer and ReplaySubject interaction

Open
#1,985 0 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

> Which library version?
Bug occurs with 5.0 and 6.0

> What is the use case or problem?
We have a continuous stream of data that we want to process in chunks. To archive this we a buffer that we can flush if needed.

```
public static IObservable> FlushableBuffer(this IObservable source, int max, ISubject manualFlushSignal)
{
var bufferOverFlowFlushSignal = source.Skip(max - 1)
.Select(_ => Unit.Default);
var mergedSignals = manualFlushSignal.Merge(bufferOverFlowFlushSignal);

return source.Buffer(() => mergedSignals);
}
```

This behavior works fine for `BehaviorSubject` and `Subject` but with `ReplaySubject` we get an infinite loop when the buffer is full.

I wrote some tests to illustrate this issue:
[ReactiveReplaySubjectFlushableBuffer.zip](https://github.com/dotnet/reactive/files/12464170/ReactiveReplaySubjectFlushableBuffer.zip)

Best Regards
Simon

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.