Skip to content

Commit 0eaab5c

Browse files
TurnerjTurnerj
authored
Support bifurcation with results (#34)
* Add results to bifurcation * Update dependencies * Update build targets
1 parent bcdd850 commit 0eaab5c

7 files changed

Lines changed: 200 additions & 25 deletions

File tree

‎.github/workflows/build.yml‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,7 @@ jobs:
3131
uses: actions/setup-dotnet@v4
3232
with:
3333
dotnet-version: |
34-
6.0.x
35-
7.0.x
34+
8.0.x
3635
- name: Install dependencies
3736
run: dotnet restore
3837
- name: Build

‎src/TurnerSoftware.Aqueduct/BifurcationExtensionMethods.cs‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,45 @@ namespace TurnerSoftware.Aqueduct;
77
/// </summary>
88
public static class BifurcationExtensionMethods
99
{
10+
/// <summary>
11+
/// Performs bifurcation with <paramref name="sourceReader"/>, splitting the resulting data into multiple <paramref name="targetConfigs"/>.
12+
/// </summary>
13+
/// <param name="sourceReader">The source to read from.</param>
14+
/// <param name="targetConfigs">The targets to provide the bifurcated data to.</param>
15+
/// <returns>A list of results from the targets, in the order the targets were provided.</returns>
16+
public static Task<IReadOnlyList<TResult?>> BifurcatedReadAsync<TResult>(this PipeReader sourceReader, params BifurcationTargetConfig<TResult>[] targetConfigs)
17+
=> BifurcatedReadAsync(sourceReader, BifurcationSourceConfig.DefaultConfig, targetConfigs);
18+
/// <summary>
19+
/// Performs bifurcation with <paramref name="sourceReader"/>, splitting the resulting data into multiple <paramref name="targetConfigs"/>.
20+
/// </summary>
21+
/// <param name="sourceReader">The source to read from.</param>
22+
/// <param name="sourceConfig">Source-specific configuration for reading.</param>
23+
/// <param name="targetConfigs">The targets to provide the bifurcated data to.</param>
24+
/// <returns>A list of results from the targets, in the order the targets were provided.</returns>
25+
public static Task<IReadOnlyList<TResult?>> BifurcatedReadAsync<TResult>(this PipeReader sourceReader, BifurcationSourceConfig sourceConfig, params BifurcationTargetConfig<TResult>[] targetConfigs)
26+
=> PipeBifurcation.BifurcatedReadAsync(sourceReader, sourceConfig, targetConfigs);
27+
28+
/// <summary>
29+
/// Performs bifurcation with <paramref name="sourceStream"/>, splitting the resulting data into multiple <paramref name="targetConfigs"/>.
30+
/// </summary>
31+
/// <param name="sourceStream">The source to read from.</param>
32+
/// <param name="targetConfigs">The targets to provide the bifurcated data to.</param>
33+
/// <returns>A list of results from the targets, in the order the targets were provided.</returns>
34+
public static Task<IReadOnlyList<TResult?>> BifurcatedReadAsync<TResult>(this Stream sourceStream, params BifurcationTargetConfig<TResult>[] targetConfigs)
35+
=> BifurcatedReadAsync(sourceStream, StreamBifurcationSourceConfig.DefaultStreamConfig, targetConfigs);
36+
/// <summary>
37+
/// Performs bifurcation with <paramref name="sourceStream"/>, splitting the resulting data into multiple <paramref name="targetConfigs"/>.
38+
/// </summary>
39+
/// <param name="sourceStream">The source to read from.</param>
40+
/// <param name="sourceConfig">Source-specific configuration for reading.</param>
41+
/// <param name="targetConfigs">The targets to provide the bifurcated data to.</param>
42+
/// <returns>A list of results from the targets, in the order the targets were provided.</returns>
43+
public static Task<IReadOnlyList<TResult?>> BifurcatedReadAsync<TResult>(this Stream sourceStream, StreamBifurcationSourceConfig sourceConfig, params BifurcationTargetConfig<TResult>[] targetConfigs)
44+
{
45+
var sourceReader = PipeReader.Create(sourceStream, new StreamPipeReaderOptions(leaveOpen: sourceConfig.LeaveOpen));
46+
return PipeBifurcation.BifurcatedReadAsync(sourceReader, sourceConfig, targetConfigs);
47+
}
48+
1049
/// <summary>
1150
/// Performs bifurcation with <paramref name="sourceReader"/>, splitting the resulting data into multiple <paramref name="targetConfigs"/>.
1251
/// </summary>

‎src/TurnerSoftware.Aqueduct/BifurcationTargetConfig.cs‎

Lines changed: 66 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ namespace TurnerSoftware.Aqueduct;
55
/// <summary>
66
/// Manages the configuration for a specific bifurcation target.
77
/// </summary>
8-
public class BifurcationTargetConfig
8+
public class BifurcationTargetConfig<TResult>
99
{
1010
internal const int DefaultBlockAfter = 32768;
1111
internal const int DefaultResumeAfter = 16384;
@@ -14,7 +14,7 @@ public class BifurcationTargetConfig
1414
/// <summary>
1515
/// The reader function that will handle this specific bifurcation target.
1616
/// </summary>
17-
public Func<PipeReader, CancellationToken, Task> Reader { get; }
17+
public Func<PipeReader, CancellationToken, Task<TResult>> Reader { get; }
1818
/// <summary>
1919
/// The individual exception handler for this bifurcation target when exceptions occur in any target during bifurcation.
2020
/// </summary>
@@ -42,7 +42,7 @@ public class BifurcationTargetConfig
4242
/// <param name="maxTotalBytes">The max number of bytes to write to the bifurcation target. Use -1 to specify no limit.</param>
4343
/// <exception cref="ArgumentException"></exception>
4444
public BifurcationTargetConfig(
45-
Func<Stream, CancellationToken, Task> reader,
45+
Func<Stream, CancellationToken, Task<TResult>> reader,
4646
Func<Exception, Task>? exceptionHandler = null,
4747
int blockAfter = DefaultBlockAfter,
4848
int resumeAfter = DefaultResumeAfter,
@@ -66,7 +66,7 @@ public BifurcationTargetConfig(
6666
/// <param name="maxTotalBytes">The max number of bytes to write to the bifurcation target. Use -1 to specify no limit.</param>
6767
/// <exception cref="ArgumentException"></exception>
6868
public BifurcationTargetConfig(
69-
Func<PipeReader, CancellationToken, Task> reader,
69+
Func<PipeReader, CancellationToken, Task<TResult>> reader,
7070
Func<Exception, Task>? exceptionHandler = null,
7171
int blockAfter = DefaultBlockAfter,
7272
int resumeAfter = DefaultResumeAfter,
@@ -90,3 +90,65 @@ public BifurcationTargetConfig(
9090
MaxTotalBytes = maxTotalBytes;
9191
}
9292
}
93+
94+
/// <summary>
95+
/// Manages the configuration for a specific bifurcation target.
96+
/// </summary>
97+
public class BifurcationTargetConfig : BifurcationTargetConfig<object?>
98+
{
99+
/// <summary>
100+
/// Creates a new <see cref="BifurcationTargetConfig"/> for <see cref="Stream"/>-based readers.
101+
/// </summary>
102+
/// <param name="reader">The reader function that will handle this specific bifurcation target.</param>
103+
/// <param name="exceptionHandler">The individual exception handler for this bifurcation target when exceptions occur in any target during bifurcation.</param>
104+
/// <param name="blockAfter">The number of unread bytes before writing will block to the bifurcation target. This must be the same or greater than <paramref name="resumeAfter"/>.</param>
105+
/// <param name="resumeAfter">The number of unread bytes before resuming writing to the bifurcation target. This must be the same or lower than <paramref name="blockAfter"/>.</param>
106+
/// <param name="maxTotalBytes">The max number of bytes to write to the bifurcation target. Use -1 to specify no limit.</param>
107+
/// <exception cref="ArgumentException"></exception>
108+
public BifurcationTargetConfig(
109+
Func<Stream, CancellationToken, Task> reader,
110+
Func<Exception, Task>? exceptionHandler = null,
111+
int blockAfter = DefaultBlockAfter,
112+
int resumeAfter = DefaultResumeAfter,
113+
int maxTotalBytes = DefaultMaxTotalBytes
114+
) : base(
115+
async (pipeReader, cancellationToken) =>
116+
{
117+
await reader(pipeReader.AsStream(), cancellationToken);
118+
return null;
119+
},
120+
exceptionHandler,
121+
blockAfter,
122+
resumeAfter,
123+
maxTotalBytes
124+
)
125+
{ }
126+
127+
/// <summary>
128+
/// Creates a new <see cref="BifurcationTargetConfig"/> for <see cref="PipeReader"/>-based readers.
129+
/// </summary>
130+
/// <param name="reader">The reader function that will handle this specific bifurcation target.</param>
131+
/// <param name="exceptionHandler">The individual exception handler for this bifurcation target when exceptions occur during bifurcation.</param>
132+
/// <param name="blockAfter">The number of unread bytes before writing will block to the bifurcation target. This must be the same or greater than <paramref name="resumeAfter"/>.</param>
133+
/// <param name="resumeAfter">The number of unread bytes before resuming writing to the bifurcation target. This must be the same or lower than <paramref name="blockAfter"/>.</param>
134+
/// <param name="maxTotalBytes">The max number of bytes to write to the bifurcation target. Use -1 to specify no limit.</param>
135+
/// <exception cref="ArgumentException"></exception>
136+
public BifurcationTargetConfig(
137+
Func<PipeReader, CancellationToken, Task> reader,
138+
Func<Exception, Task>? exceptionHandler = null,
139+
int blockAfter = DefaultBlockAfter,
140+
int resumeAfter = DefaultResumeAfter,
141+
int maxTotalBytes = DefaultMaxTotalBytes
142+
) : base(
143+
async (pipeReader, cancellationToken) =>
144+
{
145+
await reader(pipeReader, cancellationToken);
146+
return null;
147+
},
148+
exceptionHandler,
149+
blockAfter,
150+
resumeAfter,
151+
maxTotalBytes
152+
)
153+
{ }
154+
}

‎src/TurnerSoftware.Aqueduct/PipeBifurcation.cs‎

Lines changed: 33 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -5,16 +5,18 @@ namespace TurnerSoftware.Aqueduct;
55

66
internal static class PipeBifurcation
77
{
8-
private class BifurcationState
8+
private class BifurcationState<TResult>
99
{
1010
private readonly Pipe Pipe;
1111

12-
private Task? ReaderTask;
12+
private Task<TResult>? ReaderTask;
1313
private int RemainingBytes;
1414

15-
public readonly BifurcationTargetConfig Config;
15+
public readonly BifurcationTargetConfig<TResult> Config;
1616

17-
public BifurcationState(BifurcationTargetConfig config)
17+
public TResult? Result { get; private set; }
18+
19+
public BifurcationState(BifurcationTargetConfig<TResult> config)
1820
{
1921
Config = config;
2022

@@ -30,7 +32,14 @@ public BifurcationState(BifurcationTargetConfig config)
3032

3133
public void StartReader(CancellationToken cancellationToken)
3234
{
33-
ReaderTask = Config.Reader(Pipe.Reader, cancellationToken);
35+
try
36+
{
37+
ReaderTask = Config.Reader(Pipe.Reader, cancellationToken);
38+
}
39+
catch (Exception ex)
40+
{
41+
ReaderTask = Task.FromException<TResult>(ex);
42+
}
3443
}
3544

3645
/// <summary>
@@ -81,11 +90,11 @@ public async ValueTask<bool> WriteAsync(ReadOnlySequence<byte> buffer, Cancellat
8190
/// Completes the bifurcation target and awaits the reader. Any exceptions the reader throws will bubble out.
8291
/// </summary>
8392
/// <returns></returns>
84-
public async Task CompleteAsync()
93+
public async Task<TResult?> CompleteAsync()
8594
{
8695
if (IsCompleted)
8796
{
88-
return;
97+
return Result;
8998
}
9099

91100
IsCompleted = true;
@@ -94,8 +103,10 @@ public async Task CompleteAsync()
94103
//Run the task to completion
95104
if (ReaderTask is not null)
96105
{
97-
await ReaderTask;
106+
Result = await ReaderTask;
98107
}
108+
109+
return Result;
99110
}
100111

101112
/// <summary>
@@ -104,17 +115,17 @@ public async Task CompleteAsync()
104115
/// </summary>
105116
/// <param name="exception">The exception used to trigger the faulted state.</param>
106117
/// <returns></returns>
107-
public async Task CompleteWithExceptionAsync(Exception exception)
118+
public async Task<TResult?> CompleteWithExceptionAsync(Exception exception)
108119
{
109120
await Pipe.Writer.CompleteAsync(exception);
110121
if (ReaderTask is not null)
111122
{
112123
//Ensure that the reader task has completed execution (faulted or not)
113-
if (!ReaderTask.IsCompleted && !ReaderTask.IsFaulted)
124+
if (!ReaderTask.IsFaulted)
114125
{
115126
try
116127
{
117-
await ReaderTask;
128+
Result = await ReaderTask;
118129
}
119130
catch
120131
{
@@ -135,18 +146,21 @@ public async Task CompleteWithExceptionAsync(Exception exception)
135146
}
136147
}
137148
}
149+
150+
return Result;
138151
}
139152
}
140153

141-
public static async Task BifurcatedReadAsync(PipeReader sourceReader, BifurcationSourceConfig sourceConfig, params BifurcationTargetConfig[] targetConfigs)
154+
public static async Task<IReadOnlyList<TResult?>> BifurcatedReadAsync<TResult>(PipeReader sourceReader, BifurcationSourceConfig sourceConfig, params BifurcationTargetConfig<TResult>[] targetConfigs)
142155
{
143156
if (targetConfigs.Length == 0)
144157
{
145158
throw new ArgumentException("No target configurations to bifurcate the source reader to", nameof(targetConfigs));
146159
}
147160

148161
var earlyCompletedTargets = 0;
149-
var targets = new BifurcationState[targetConfigs.Length];
162+
var targets = new BifurcationState<TResult>[targetConfigs.Length];
163+
var results = new TResult?[targetConfigs.Length];
150164

151165
for (var i = 0; i < targetConfigs.Length; i++)
152166
{
@@ -203,8 +217,10 @@ public static async Task BifurcatedReadAsync(PipeReader sourceReader, Bifurcatio
203217
for (var i = 0; i < targets.Length; i++)
204218
{
205219
var target = targets[i];
206-
await target.CompleteAsync();
220+
results[i] = await target.CompleteAsync();
207221
}
222+
223+
return results;
208224
}
209225
catch (Exception innerException)
210226
{
@@ -214,13 +230,15 @@ public static async Task BifurcatedReadAsync(PipeReader sourceReader, Bifurcatio
214230
for (var i = 0; i < targets.Length; i++)
215231
{
216232
var target = targets[i];
217-
await target.CompleteWithExceptionAsync(exception);
233+
results[i] = await target.CompleteWithExceptionAsync(exception);
218234
}
219235

220236
if (sourceConfig.BubbleExceptions)
221237
{
222238
throw exception;
223239
}
240+
241+
return results;
224242
}
225243
}
226244
}

‎src/TurnerSoftware.Aqueduct/TurnerSoftware.Aqueduct.csproj‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
<Project Sdk="Microsoft.NET.Sdk">
22

33
<PropertyGroup>
4-
<TargetFramework>net6.0</TargetFramework>
4+
<TargetFramework>net8.0</TargetFramework>
55
<AssemblyName>TurnerSoftware.Aqueduct</AssemblyName>
66
<Title>TurnerSoftware.Aqueduct</Title>
77
<Description>Utilities and extension methods for working with streams and pipes</Description>
@@ -16,7 +16,7 @@
1616
</ItemGroup>
1717

1818
<ItemGroup>
19-
<PackageReference Include="System.IO.Pipelines" Version="9.0.1" />
19+
<PackageReference Include="System.IO.Pipelines" Version="8.0.0" />
2020
</ItemGroup>
2121

2222
</Project>

‎tests/TurnerSoftware.Aqueduct.Tests/PipeBifurcationTests.cs‎

Lines changed: 58 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ public async Task NoTargets_ThrowsException()
2727
{
2828
var source = CreateSource("Test Value");
2929

30-
await FluentActions.Awaiting(() => PipeBifurcation.BifurcatedReadAsync(source, BifurcationSourceConfig.DefaultConfig))
30+
await FluentActions.Awaiting(() => PipeBifurcation.BifurcatedReadAsync<object>(source, BifurcationSourceConfig.DefaultConfig))
3131
.Should().ThrowAsync<ArgumentException>()
3232
.WithMessage("No target configurations*");
3333
}
@@ -204,6 +204,63 @@ public async Task MultiTarget_CompletedTargets_AreNotWrittenToFurther()
204204
secondTargetReadBufferLength.Should().Be(8);
205205
}
206206

207+
private record TestResultData(long TargetOneValue, long TargetTwoValue);
208+
209+
[TestMethod]
210+
public async Task MultiTarget_ResultsAreReturnedInOrderFromTargets()
211+
{
212+
var source = CreateSource("Test Value");
213+
214+
var bifurcationTask = PipeBifurcation.BifurcatedReadAsync(
215+
source,
216+
new BifurcationSourceConfig(),
217+
new BifurcationTargetConfig<TestResultData>(
218+
async (reader, cancellationToken) =>
219+
{
220+
var result = await reader.ReadAsync(cancellationToken);
221+
return new(result.Buffer.Length, 0);
222+
}
223+
),
224+
new BifurcationTargetConfig<TestResultData>(
225+
async (reader, cancellationToken) =>
226+
{
227+
var result = await reader.ReadAsync(cancellationToken);
228+
return new(0, result.Buffer.Length);
229+
}
230+
)
231+
);
232+
233+
var result = await bifurcationTask;
234+
result.Should().NotBeNull().And.HaveCount(2);
235+
result[0].Should().BeEquivalentTo(new TestResultData(10, 0));
236+
result[1].Should().BeEquivalentTo(new TestResultData(0, 10));
237+
}
238+
239+
[TestMethod]
240+
public async Task MultiTarget_NonBubblingExceptionsStillReturnResultsThatCompleted()
241+
{
242+
var source = CreateSource("Test Value");
243+
244+
var bifurcationTask = PipeBifurcation.BifurcatedReadAsync(
245+
source,
246+
new BifurcationSourceConfig(bubbleExceptions: false),
247+
new BifurcationTargetConfig<bool>(
248+
(PipeReader reader, CancellationToken cancellationToken) =>
249+
{
250+
throw new Exception("Whoops");
251+
}
252+
),
253+
new BifurcationTargetConfig<bool>(
254+
(PipeReader reader, CancellationToken cancellationToken) => Task.FromResult(true)
255+
)
256+
);
257+
258+
var result = await bifurcationTask;
259+
result.Should().NotBeNull().And.HaveCount(2);
260+
result[0].Should().BeFalse();
261+
result[1].Should().BeTrue();
262+
}
263+
207264
[TestMethod]
208265
public async Task MultiTarget_ExceptionsFromOtherTargets_PushesExceptionsToTargets()
209266
{

‎tests/TurnerSoftware.Aqueduct.Tests/TurnerSoftware.Aqueduct.Tests.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
<Project Sdk="Microsoft.NET.Sdk">
22

33
<PropertyGroup>
4-
<TargetFrameworks>net6.0;net7.0</TargetFrameworks>
4+
<TargetFrameworks>net8.0</TargetFrameworks>
55
<ImplicitUsings>enable</ImplicitUsings>
66
<Nullable>enable</Nullable>
77

0 commit comments

Comments
 (0)