33using System . ComponentModel ;
44using System . IO . Pipelines ;
55using System . Linq ;
6+ using System . Reactive ;
67using System . Reactive . Concurrency ;
78using System . Reactive . Disposables ;
89using System . Reactive . Linq ;
910using System . Reactive . Subjects ;
1011using System . Text ;
1112using System . Threading ;
13+ using System . Threading . Channels ;
1214using System . Threading . Tasks ;
1315using Microsoft . Extensions . Logging ;
1416
@@ -20,11 +22,16 @@ public class OutputHandler : IOutputHandler
2022 private readonly ISerializer _serializer ;
2123 private readonly IEnumerable < IOutputFilter > _outputFilters ;
2224 private readonly ILogger < OutputHandler > _logger ;
23- private readonly Subject < object > _queue ;
24- private readonly ReplaySubject < object > _delayedQueue ;
25+
26+
27+ private readonly ChannelReader < object > _queue ;
28+ private readonly Queue < object > _delayedQueue ;
2529 private readonly TaskCompletionSource < object ? > _outputIsFinished ;
2630 private readonly CompositeDisposable _disposable ;
2731 private bool _delayComplete ;
32+ private readonly CancellationTokenSource _stopProcessing ;
33+ private readonly Channel < object > _channel ;
34+ private readonly ChannelWriter < object > _writer ;
2835
2936 public OutputHandler (
3037 PipeWriter pipeWriter ,
@@ -38,22 +45,26 @@ ILogger<OutputHandler> logger
3845 _serializer = serializer ;
3946 _outputFilters = outputFilters . ToArray ( ) ;
4047 _logger = logger ;
41- _queue = new Subject < object > ( ) ;
42- _delayedQueue = new ReplaySubject < object > ( ) ;
48+ _delayedQueue = new Queue < object > ( ) ;
4349 _outputIsFinished = new TaskCompletionSource < object ? > ( ) ;
4450
51+ _channel = Channel . CreateUnbounded < object > (
52+ new UnboundedChannelOptions ( ) {
53+ AllowSynchronousContinuations = true ,
54+ SingleReader = true ,
55+ SingleWriter = false
56+ }
57+ ) ;
58+ _queue = _channel . Reader ;
59+ _writer = _channel . Writer ;
60+
61+ _stopProcessing = new CancellationTokenSource ( ) ;
4562 _disposable = new CompositeDisposable {
46- _queue
47- . ObserveOn ( scheduler )
48- . Select ( value => Observable . FromAsync ( ct => ProcessOutputStream ( value , ct ) ) )
49- . Concat ( )
50- . Subscribe ( ) ,
51- _delayedQueue
52- . ToArray ( )
53- . SelectMany ( z => z )
54- . Subscribe ( _queue . OnNext ) ,
55- _queue ,
56- _delayedQueue
63+ Disposable . Create ( ( ) => _stopProcessing . Cancel ( ) ) ,
64+ _stopProcessing ,
65+ Observable . FromAsync ( ( ) => ProcessOutputStream ( _stopProcessing . Token ) )
66+ . Do ( _ => { } , e => _logger . LogCritical ( e , "unhandled exception" ) )
67+ . Subscribe ( )
5768 } ;
5869 }
5970
@@ -64,34 +75,39 @@ private bool ShouldSend(object value)
6475
6576 public void Send ( object ? value )
6677 {
67- // _logger.LogTrace("Writing out value {@Value} ({Type})", value, value?.GetType().FullName);
68-
6978 try
7079 {
71- if ( _queue . IsDisposed || _disposable . IsDisposed || value == null ) return ;
72- if ( ! ShouldSend ( value ) )
80+ if ( _disposable . IsDisposed || value == null ) return ;
81+ if ( ! ShouldSend ( value ) && ! _delayComplete )
7382 {
74- if ( _delayComplete || _delayedQueue . IsDisposed || ! _delayedQueue . HasObservers ) return ;
75- _delayedQueue . OnNext ( value ) ;
83+ _delayedQueue . Enqueue ( value ) ;
7684 }
7785 else
7886 {
79- _queue . OnNext ( value ) ;
87+ _writer . TryWrite ( value ) ;
8088 }
8189 }
82- catch ( ObjectDisposedException ) { }
90+ catch ( ObjectDisposedException )
91+ {
92+ }
8393 }
8494
8595 public void Initialized ( )
8696 {
87- if ( _delayComplete || _delayedQueue . IsDisposed || ! _delayedQueue . HasObservers ) return ;
88- _delayedQueue . OnCompleted ( ) ;
97+ if ( _delayComplete ) return ;
98+ while ( _delayedQueue . Count > 0 )
99+ {
100+ var item = _delayedQueue . Dequeue ( ) ;
101+ _writer . TryWrite ( item ) ;
102+ }
103+
89104 _delayComplete = true ;
90- _delayedQueue . Dispose ( ) ;
105+ _delayedQueue . Clear ( ) ;
91106 }
92107
93108 public async Task StopAsync ( )
94109 {
110+ _channel . Writer . TryComplete ( ) ;
95111 await _pipeWriter . CompleteAsync ( ) . ConfigureAwait ( false ) ;
96112 _disposable . Dispose ( ) ;
97113 }
@@ -107,17 +123,21 @@ internal async Task WriteAndFlush()
107123 await _pipeWriter . CompleteAsync ( ) . ConfigureAwait ( false ) ;
108124 }
109125
110- private async Task ProcessOutputStream ( object value , CancellationToken cancellationToken )
126+ private async Task ProcessOutputStream ( CancellationToken cancellationToken )
111127 {
112128 try
113129 {
130+ do
131+ {
132+ var value = await _queue . ReadAsync ( cancellationToken ) ;
114133// _logger.LogTrace("Writing out {@Value}", value);
115- // TODO: this will be part of the serialization refactor to make streaming first class
116- var content = _serializer . SerializeObject ( value ) ;
117- var contentBytes = Encoding . UTF8 . GetBytes ( content ) . AsMemory ( ) ;
118- await _pipeWriter . WriteAsync ( Encoding . UTF8 . GetBytes ( $ "Content-Length: { contentBytes . Length } \r \n \r \n ") , cancellationToken ) . ConfigureAwait ( false ) ;
119- await _pipeWriter . WriteAsync ( contentBytes , cancellationToken ) . ConfigureAwait ( false ) ;
120- await _pipeWriter . FlushAsync ( cancellationToken ) . ConfigureAwait ( false ) ;
134+ // TODO: this will be part of the serialization refactor to make streaming first class
135+ var content = _serializer . SerializeObject ( value ) ;
136+ var contentBytes = Encoding . UTF8 . GetBytes ( content ) . AsMemory ( ) ;
137+ await _pipeWriter . WriteAsync ( Encoding . UTF8 . GetBytes ( $ "Content-Length: { contentBytes . Length } \r \n \r \n ") , cancellationToken ) . ConfigureAwait ( false ) ;
138+ await _pipeWriter . WriteAsync ( contentBytes , cancellationToken ) . ConfigureAwait ( false ) ;
139+ await _pipeWriter . FlushAsync ( cancellationToken ) . ConfigureAwait ( false ) ;
140+ } while ( true ) ;
121141 }
122142 catch ( OperationCanceledException ex ) when ( ex . CancellationToken != cancellationToken )
123143 {
@@ -136,12 +156,14 @@ private async Task ProcessOutputStream(object value, CancellationToken cancellat
136156 private void Error ( Exception ex )
137157 {
138158 _outputIsFinished . TrySetResult ( ex ) ;
159+ _writer . TryComplete ( ) ;
139160 _disposable . Dispose ( ) ;
140161 }
141162
142163 public void Dispose ( )
143164 {
144165 _outputIsFinished . TrySetResult ( null ) ;
166+ _writer . TryComplete ( ) ;
145167 _disposable . Dispose ( ) ;
146168 }
147169 }
0 commit comments