Remove reactive extension analysis prototypes - #57
Open
michaelstonis wants to merge 11 commits into
Open
Conversation
Add TransformationExtensions with: - MapTo<T, TResult>: project all values to a constant - CompactMap<T, TResult>: select+filter nulls (ref and value type overloads) - WithIndex<T>: pair each element with its zero-based index (Pattern B) - RunningFold<T, TAccumulate>: alias for Scan with seed - RunningReduce<T>: alias for Scan without seed Add FlatMapExtensions with: - ConcatMap: sequential inner subscription with pending queue - SwitchMap / FlatMapLatest: cancel-and-replace with generation guard - ExhaustMap: ignore source while inner is active - Expand: breadth-first recursive expansion with active counter - MergeScan: scan with merged inner observables - SwitchScan: scan with switched inner observables Add TransformationExtensionsTests (25 tests) and FlatMapExtensionsTests (27 tests) covering normal behavior, completion propagation, null/arg validation, and edge cases. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…it, BufferWithOverflow, Chunked) Add five new timing extension operators to the R3Ext library: - TimeInterval<T>: wraps each emission with elapsed time since previous - DelayWhen<T>: delays each element by a per-element duration observable, with optional subscription-delay overload - RateLimit<T>: allows at most N items per period, queuing excess - BufferWithOverflow<T>: bounded pass-through buffer with DropOldest, DropLatest, or Error overflow strategies - Chunked<T>: sliding or non-overlapping window by count with step Also adds TimeInterval<T> struct, OverflowStrategy enum, and TimingAdvancedTests.cs covering 3+ tests per operator. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…y/All, Find, FindIndex, DefaultIfEmpty, ThrowIfEmpty, Audit, AuditTime, Sample) - Add FilteringExtensions.Advanced.cs with 10 new operators using Pattern A (compositional) and Pattern B (Observable.Create with Lock gate) - Make FilteringExtensions partial to support the new file - Add FilteringAdvancedTests.cs with 45 tests covering all operators (3+ per operator) - Fix pre-existing CS compilation errors in CombinationExtensionsTests.cs and WindowingOperatorsTests.cs that blocked test execution Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- ErrorHandlingExtensions.Advanced.cs: RetryWhen, ReplaceError, ReplaceEmpty, SelectSafe, WhereSafe - SideEffectExtensions.cs: DoOnError, DoOnComplete, DoOnTerminate, DoAfterTerminate - ErrorHandlingExtraTests.cs, SideEffectExtensionsTests.cs: full test coverage - Add pragma suppressions to resolve StyleCop errors in parallel-agent files - Fix WindowingOperatorsTests.cs: replace invalid Observable.ToArray().Subscribe() with correct collect-on-complete subscription pattern Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…ensions Add named elements to tuple return types and Observable.Create generic arguments in ForkJoin<T1,T2> and ForkJoin<T1,T2,T3> overloads. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- WindowCount<T>(count, skip): overlapping/non-overlapping count-based windows - WindowTime<T>(timeSpan): periodic time-based windows - WindowTime<T>(timeSpan, maxCount): windows bounded by time or item count - BufferToggle<T,TOpen,TClose>: open/close buffers via observable signals - BufferWhen<T,TClose>: single rolling buffer closed by a selector observable All operators follow the existing Lock/ITimer/Observable.Create pattern from TimingExtensions.Buffer.cs. Full test coverage: 30 tests across all 5 operators covering argument validation, normal operation, completion, and edge cases. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Add CombinationExtensions with: - ForkJoin<T> (params/IEnumerable/typed 2- and 3-source overloads) - OnErrorResumeNext<T> (sequential, advances on completion OR non-terminal error) - Iif<T>/Condition<T> (deferred conditional subscription) - SequenceEqual<T> (element-by-element comparison with optional comparer) - RepeatWhen<T> (handler-driven repetition with notifier Subject) - Generate<TState,TResult> and Generate<TState> (synchronous state machine) Add CombinationExtensionsTests with 31 tests (3+ per operator). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- AggregateStreamExtensions: RunningCount, RunningSum (INumber<T>), RunningAverage (double/float/decimal/int overloads), RunningMin/Max with IComparable<T> and IComparer<T> overloads - AsyncSubject<T>: buffers last value, emits only on successful completion; supports late subscribers and failure propagation - ReadOnlySubject<T>: wraps any Observable<T> to hide subject methods; AsReadOnly() extensions for Observable<T>, Subject<T>, BehaviorSubject<T> - Tests: 19 tests covering all new operators and subject types Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…tics Update windowing, buffering, and rate-limiting operators to remain active after resumable errors and fix completion logic in DelayWhen and RetryWhen. This ensures subsequent items are not dropped and prevents potential deadlocks or stack overflows during synchronous re-subscriptions.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a broad set of new “advanced” reactive/observable operators and supporting subject utilities to the R3Ext library, along with comprehensive unit tests covering the new behaviors (timing/windowing, buffering, transformation, flat-mapping, filtering, combination, aggregation, and error-handling).
Changes:
- Adds multiple new operator extension modules (timing/windowing/buffering, transformation, side-effects, flat-map, filtering advanced, combination, incremental aggregates, and extra error-handling).
- Adds new subject utilities (AsyncSubject and a read-only observable wrapper) to complement the operator set.
- Adds extensive xUnit test coverage for the newly introduced operators and subjects; updates FilteringExtensions to be
partialto support the new advanced split.
Reviewed changes
Copilot reviewed 23 out of 23 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| R3Ext/Timing/TimingExtensions.Window.cs | Adds count- and time-based windowing operators. |
| R3Ext/Timing/TimingExtensions.BufferAdvanced.cs | Adds BufferToggle/BufferWhen advanced buffering operators. |
| R3Ext/Timing/TimingExtensions.Advanced.cs | Adds advanced timing operators (TimeInterval, DelayWhen overloads, RateLimit, BufferWithOverflow, Chunked) plus related types. |
| R3Ext/Subjects/ReadOnlySubjectWrapper.cs | Adds ReadOnlySubject wrapper and AsReadOnly extension helpers. |
| R3Ext/Subjects/AsyncSubject.cs | Adds AsyncSubject implementation that emits last value on successful completion. |
| R3Ext/Extensions/TransformationExtensions.cs | Adds transformation operators (MapTo, CompactMap, WithIndex, RunningFold/Reduce). |
| R3Ext/Extensions/SideEffectExtensions.cs | Adds focused side-effect operators (DoOnError/Complete/Terminate/AfterTerminate). |
| R3Ext/Extensions/FlatMapExtensions.cs | Adds higher-order operators (ConcatMap, SwitchMap, ExhaustMap, Expand, MergeScan, SwitchScan). |
| R3Ext/Extensions/FilteringExtensions.cs | Makes FilteringExtensions partial to allow advanced extensions split. |
| R3Ext/Extensions/FilteringExtensions.Advanced.cs | Adds advanced filtering operators (IgnoreElements, IsEmpty, Every/All, Find/FindIndex, DefaultIfEmpty, ThrowIfEmpty, Audit/AuditTime, Sample). |
| R3Ext/Extensions/CombinationExtensions.cs | Adds combination/creation operators (ForkJoin overloads, OnErrorResumeNext, Iif/Condition, SequenceEqual, RepeatWhen, Generate). |
| R3Ext/Extensions/AggregateStreamExtensions.cs | Adds incremental aggregate operators (RunningCount/Sum/Average/Min/Max). |
| R3Ext/ErrorHandling/ErrorHandlingExtensions.Advanced.cs | Adds RetryWhen and additional error-handling utilities (ReplaceError/Empty, SelectSafe, WhereSafe). |
| R3Ext.Tests/WindowingOperatorsTests.cs | Adds tests for WindowCount/WindowTime/BufferToggle/BufferWhen behaviors and error-resume semantics. |
| R3Ext.Tests/TransformationExtensionsTests.cs | Adds tests for transformation operators (MapTo, CompactMap, WithIndex, RunningFold/Reduce). |
| R3Ext.Tests/TimingAdvancedTests.cs | Adds tests for advanced timing operators (TimeInterval, DelayWhen, RateLimit, BufferWithOverflow, Chunked). |
| R3Ext.Tests/SubjectTypesTests.cs | Adds tests for AsyncSubject and ReadOnlySubject behaviors. |
| R3Ext.Tests/SideEffectExtensionsTests.cs | Adds tests for side-effect operator semantics and ordering. |
| R3Ext.Tests/FlatMapExtensionsTests.cs | Adds tests for flat-map/higher-order operators (ConcatMap/SwitchMap/ExhaustMap/Expand/MergeScan/SwitchScan). |
| R3Ext.Tests/FilteringAdvancedTests.cs | Adds tests for advanced filtering operators (IgnoreElements/IsEmpty/Every/Find/etc.). |
| R3Ext.Tests/ErrorHandlingExtraTests.cs | Adds tests for RetryWhen and safe select/where plus replace-error/empty operators. |
| R3Ext.Tests/CombinationExtensionsTests.cs | Adds tests for combination/creation operators (ForkJoin/Generate/Iif/SequenceEqual/RepeatWhen/OnErrorResumeNext). |
| R3Ext.Tests/AggregateStreamTests.cs | Adds tests for incremental aggregate operators. |
Suppressed comments (1)
R3Ext/Extensions/FlatMapExtensions.cs:793
MergeScanadds every inner subscription toinnerSubsbut never removes them when an inner completes. Over time this can cause unbounded memory growth for hot sources. Consider removing each inner subscription frominnerSubsin its completion callback (and keep the list for active subscriptions only).
using (gate.EnterScope())
{
if (disposed)
{
sub.Dispose();
return;
}
innerSubs.Add(sub);
}
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+3
to
+6
| namespace R3Ext; | ||
|
|
||
| public static partial class TimingExtensions | ||
| { |
Comment on lines
+104
to
+117
| r => | ||
| { | ||
| using (gate.EnterScope()) | ||
| { | ||
| if (disposed) | ||
| { | ||
| return; | ||
| } | ||
|
|
||
| observer.OnCompleted(r); | ||
| } | ||
| }); | ||
|
|
||
| SubscribeOnce(); |
| IDisposable? delaySub = null; | ||
| IDisposable? mainSub = null; | ||
|
|
||
| delaySub = subscriptionDelay.Subscribe( |
Comment on lines
+608
to
+611
| } | ||
|
|
||
| innerSubs.Add(sub); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Deletes experimental reactive operators and associated tests from the R3Ext project. Updates documentation synchronization workflows to refine allowed file paths.