Skip to content

Commit ae162dc

Browse files
Jac21Jac21
committed
Modernize BlockingCollection APIs and packaging
1 parent 727f205 commit ae162dc

12 files changed

Lines changed: 543 additions & 394 deletions

‎.github/workflows/ci.yml‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
name: CI
2+
3+
on:
4+
push:
5+
branches:
6+
- main
7+
- master
8+
pull_request:
9+
10+
jobs:
11+
build-test-pack:
12+
runs-on: ubuntu-latest
13+
14+
steps:
15+
- name: Checkout
16+
uses: actions/checkout@v4
17+
18+
- name: Setup .NET
19+
uses: actions/setup-dotnet@v4
20+
with:
21+
dotnet-version: |
22+
8.0.x
23+
24+
- name: Restore
25+
run: dotnet restore BlockingCollectionExtensions.sln
26+
27+
- name: Build
28+
run: dotnet build BlockingCollectionExtensions.sln --configuration Release --no-restore
29+
30+
- name: Test
31+
run: dotnet test BlockingCollectionExtensions.sln --configuration Release --no-build
32+
33+
- name: Pack
34+
run: dotnet pack BlockingCollectionExtensions/BlockingCollectionExtensions.csproj --configuration Release --no-build -o artifacts

‎.travis.yml‎

Lines changed: 0 additions & 16 deletions
This file was deleted.

‎BlockingCollectionExtensions/AdditiveUtilities.cs‎

Lines changed: 106 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -1,63 +1,119 @@
1-
using System;
2-
using System.Collections.Concurrent;
3-
using System.Collections.Generic;
1+
using System.Collections.Concurrent;
42
using BlockingCollectionExtensions.Structures;
53

6-
namespace BlockingCollectionExtensions
4+
namespace BlockingCollectionExtensions;
5+
6+
/// <summary>
7+
/// Utility methods for adding data to <see cref="BlockingCollection{T}"/> instances.
8+
/// </summary>
9+
public static class AdditiveUtilities
710
{
8-
public static class AdditiveUtilities
11+
/// <summary>
12+
/// Transfer contents of an enumerable into a target blocking collection.
13+
/// </summary>
14+
public static void AddFromEnumerable<T>(
15+
this BlockingCollection<T> target,
16+
IEnumerable<T> source,
17+
bool completeAddingWhenDone = false)
918
{
10-
/// <summary>
11-
/// Transfer contents of a generic enumerable into a target blocking collection,
12-
/// determine whether blocking collection should complete adding when enumerable has finished being added
13-
/// </summary>
14-
/// <typeparam name="T"></typeparam>
15-
/// <param name="target"></param>
16-
/// <param name="source"></param>
17-
/// <param name="completeAddingWhenDone"></param>
18-
public static void AddFromEnumerable<T>(this BlockingCollection<T> target, IEnumerable<T> source,
19-
bool completeAddingWhenDone)
20-
{
21-
try
19+
AddFromEnumerable(target, source, CancellationToken.None, completeAddingWhenDone);
20+
}
21+
22+
/// <summary>
23+
/// Transfer contents of an enumerable into a target blocking collection with cancellation support.
24+
/// </summary>
25+
public static void AddFromEnumerable<T>(
26+
this BlockingCollection<T> target,
27+
IEnumerable<T> source,
28+
CancellationToken cancellationToken,
29+
bool completeAddingWhenDone = false)
30+
{
31+
if (target is null)
32+
{
33+
throw new ArgumentNullException(nameof(target));
34+
}
35+
36+
if (source is null)
37+
{
38+
throw new ArgumentNullException(nameof(source));
39+
}
40+
41+
try
42+
{
43+
foreach (var item in source)
2244
{
23-
foreach (var item in source)
24-
{
25-
target.Add(item);
26-
}
45+
cancellationToken.ThrowIfCancellationRequested();
46+
target.Add(item, cancellationToken);
2747
}
28-
finally
48+
}
49+
finally
50+
{
51+
CompleteAddingIfRequested(target, completeAddingWhenDone);
52+
}
53+
}
54+
55+
/// <summary>
56+
/// Transfer contents from an async enumerable into a target blocking collection.
57+
/// </summary>
58+
public static async Task AddFromAsyncEnumerable<T>(
59+
this BlockingCollection<T> target,
60+
IAsyncEnumerable<T> source,
61+
CancellationToken cancellationToken = default,
62+
bool completeAddingWhenDone = false)
63+
{
64+
if (target is null)
65+
{
66+
throw new ArgumentNullException(nameof(target));
67+
}
68+
69+
if (source is null)
70+
{
71+
throw new ArgumentNullException(nameof(source));
72+
}
73+
74+
try
75+
{
76+
await foreach (var item in source.WithCancellation(cancellationToken).ConfigureAwait(false))
2977
{
30-
if (completeAddingWhenDone)
31-
{
32-
target.CompleteAdding();
33-
}
78+
target.Add(item, cancellationToken);
3479
}
3580
}
81+
finally
82+
{
83+
CompleteAddingIfRequested(target, completeAddingWhenDone);
84+
}
85+
}
3686

37-
/// <summary>
38-
/// Transfer contents arriving asynchronously in the form of an IObservable, subscribe a delegate that takes any data from the observable
39-
/// and adds it to the BlockingCollection
40-
/// </summary>
41-
/// <typeparam name="T"></typeparam>
42-
/// <param name="target"></param>
43-
/// <param name="source"></param>
44-
/// <param name="completeAddingWhenDone"></param>
45-
/// <returns></returns>
46-
public static IDisposable AddFromObservable<T>(this BlockingCollection<T> target, IObservable<T> source,
47-
bool completeAddingWhenDone)
48-
{
49-
return source.Subscribe(new DelegateBasedObserver<T>
50-
(
87+
/// <summary>
88+
/// Subscribe to an observable and add received values to a blocking collection.
89+
/// </summary>
90+
public static IDisposable AddFromObservable<T>(
91+
this BlockingCollection<T> target,
92+
IObservable<T> source,
93+
bool completeAddingWhenDone = false)
94+
{
95+
if (target is null)
96+
{
97+
throw new ArgumentNullException(nameof(target));
98+
}
99+
100+
if (source is null)
101+
{
102+
throw new ArgumentNullException(nameof(source));
103+
}
104+
105+
return source.Subscribe(
106+
new DelegateBasedObserver<T>(
51107
target.Add,
52-
error =>
53-
{
54-
if (completeAddingWhenDone) target.CompleteAdding();
55-
},
56-
() =>
57-
{
58-
if (completeAddingWhenDone) target.CompleteAdding();
59-
}
60-
));
108+
_ => CompleteAddingIfRequested(target, completeAddingWhenDone),
109+
() => CompleteAddingIfRequested(target, completeAddingWhenDone)));
110+
}
111+
112+
private static void CompleteAddingIfRequested<T>(BlockingCollection<T> target, bool completeAddingWhenDone)
113+
{
114+
if (completeAddingWhenDone && !target.IsAddingCompleted)
115+
{
116+
target.CompleteAdding();
61117
}
62118
}
63-
}
119+
}

‎BlockingCollectionExtensions/BlockingCollectionExtensions.csproj‎

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -3,28 +3,37 @@
33
<PropertyGroup>
44
<Authors>Jeremy Cantu</Authors>
55
<Company>Jeremy Cantu</Company>
6-
<Description>📎 Utilities to aid in utilizing the ever-useful .NET structure BlockingCollection&lt;T&gt;</Description>
6+
<Description>Helpers and adapters for working with BlockingCollection&lt;T&gt; in modern .NET applications.</Description>
77
<PackageProjectUrl>https://github.com/Jac21/BlockingCollectionExtensions</PackageProjectUrl>
88
<RepositoryUrl>https://github.com/Jac21/BlockingCollectionExtensions</RepositoryUrl>
9-
<PackageLicenseUrl>https://raw.githubusercontent.com/Jac21/BlockingCollectionExtensions/master/LICENSE</PackageLicenseUrl>
10-
<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
11-
<PackOnBuild>true</PackOnBuild>
12-
<Version>1.0.1</Version>
9+
<PackageLicenseExpression>MIT</PackageLicenseExpression>
10+
<PackageReadmeFile>README.md</PackageReadmeFile>
11+
<PackageTags>blockingcollection;producer-consumer;collections;async;threading</PackageTags>
12+
<GeneratePackageOnBuild>false</GeneratePackageOnBuild>
1313
<PackageIcon>favicon.png</PackageIcon>
14-
<LangVersion>default</LangVersion>
15-
<PackageVersion>6.0.1</PackageVersion>
16-
<TargetFrameworks>net6.0;netstandard2.1</TargetFrameworks>
17-
14+
<PackageVersion>7.0.0</PackageVersion>
15+
<Version>7.0.0</Version>
16+
<TargetFrameworks>net8.0;netstandard2.1</TargetFrameworks>
17+
<GenerateDocumentationFile>true</GenerateDocumentationFile>
18+
<IncludeSymbols>true</IncludeSymbols>
19+
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
20+
<PublishRepositoryUrl>true</PublishRepositoryUrl>
21+
<EmbedUntrackedSources>true</EmbedUntrackedSources>
22+
<PackageReleaseNotes>Modernized packaging, improved APIs, async enumerable support, and updated CI/test coverage.</PackageReleaseNotes>
1823
<DebugType>embedded</DebugType>
1924
<EnablePackageValidation>true</EnablePackageValidation>
20-
<PackageValidationBaselineVersion>6.0.0</PackageValidationBaselineVersion>
25+
<PackageValidationBaselineVersion>6.0.1</PackageValidationBaselineVersion>
2126
</PropertyGroup>
2227

2328
<ItemGroup>
2429
<None Include="..\media\favicon.png">
2530
<Pack>True</Pack>
2631
<PackagePath></PackagePath>
2732
</None>
33+
<None Include="..\README.md">
34+
<Pack>True</Pack>
35+
<PackagePath></PackagePath>
36+
</None>
2837
</ItemGroup>
2938

3039
</Project>

‎BlockingCollectionExtensions/Structures/DelegateBasedObserver.cs‎

Lines changed: 23 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,35 +1,30 @@
1-
using System;
1+
namespace BlockingCollectionExtensions.Structures;
22

3-
namespace BlockingCollectionExtensions.Structures
3+
internal sealed class DelegateBasedObserver<T> : IObserver<T>
44
{
5-
internal class DelegateBasedObserver<T> : IObserver<T>
6-
{
7-
private readonly Action<T> _onNext;
8-
private readonly Action<Exception> _onError;
9-
private readonly Action _onCompleted;
10-
11-
internal DelegateBasedObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted)
12-
{
13-
if (onCompleted != null) throw new ArgumentNullException(nameof(onCompleted));
14-
_onNext = onNext ?? throw new ArgumentNullException(nameof(onNext));
15-
_onError = onError ?? throw new ArgumentNullException(nameof(onError));
5+
private readonly Action<T> _onNext;
6+
private readonly Action<Exception> _onError;
7+
private readonly Action _onCompleted;
168

17-
_onCompleted = null;
18-
}
9+
internal DelegateBasedObserver(Action<T> onNext, Action<Exception> onError, Action onCompleted)
10+
{
11+
_onNext = onNext ?? throw new ArgumentNullException(nameof(onNext));
12+
_onError = onError ?? throw new ArgumentNullException(nameof(onError));
13+
_onCompleted = onCompleted ?? throw new ArgumentNullException(nameof(onCompleted));
14+
}
1915

20-
public void OnCompleted()
21-
{
22-
_onCompleted();
23-
}
16+
public void OnCompleted()
17+
{
18+
_onCompleted();
19+
}
2420

25-
public void OnError(Exception error)
26-
{
27-
_onError(error);
28-
}
21+
public void OnError(Exception error)
22+
{
23+
_onError(error);
24+
}
2925

30-
public void OnNext(T value)
31-
{
32-
_onNext(value);
33-
}
26+
public void OnNext(T value)
27+
{
28+
_onNext(value);
3429
}
35-
}
30+
}

0 commit comments

Comments
 (0)