huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add support for creating a bounded channel in helper

This commit is contained in:
David Fowler 2018-04-07 16:53:56 -07:00
parent 0d9c3d8898
commit 767f2900f9
1 changed files with 4 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
samples/SignalRSamples/ObservableExtensions.cs
View File

@ -9,18 +9,18 @@ namespace SignalRSamples
{ {
public static class ObservableExtensions public static class ObservableExtensions
{ {
public static ChannelReader<T> AsChannelReader<T>(this IObservable<T> observable) public static ChannelReader<T> AsChannelReader<T>(this IObservable<T> observable, int? maxBufferSize = null)
{ {
// This sample shows adapting an observable to a ChannelReader without // This sample shows adapting an observable to a ChannelReader without
// back pressure, if the connection is slower than the producer, memory will // back pressure, if the connection is slower than the producer, memory will
// start to increase. // start to increase.
// If the channel is unbounded, TryWrite will return false and effectively // If the channel is bounded, TryWrite will return false and effectively
// drop items. // drop items.
// The other alternative is to use a bounded channel, and when the limit is reached // The other alternative is to use a bounded channel, and when the limit is reached
// block on WaitToWriteAsync. This will block a thread pool thread and isn't recommended // block on WaitToWriteAsync. This will block a thread pool thread and isn't recommended and isn't shown here.
var channel = Channel.CreateUnbounded<T>(); var channel = maxBufferSize != null ? Channel.CreateBounded<T>(maxBufferSize.Value) : Channel.CreateUnbounded<T>();
var disposable = observable.Subscribe( var disposable = observable.Subscribe(
value => channel.Writer.TryWrite(value), value => channel.Writer.TryWrite(value),