@@ -25,8 +25,7 @@ namespace Elasticsearch.Net
25
25
internal class RequestDataContent : HttpContent
26
26
{
27
27
private readonly RequestData _requestData ;
28
- private readonly Func < PostData , CompleteTaskOnCloseStream , RequestDataContent , TransportContext , Task > _onStreamAvailable ;
29
-
28
+ private readonly Func < RequestData , CompleteTaskOnCloseStream , RequestDataContent , TransportContext , Task > _onStreamAvailable ;
30
29
31
30
public RequestDataContent ( RequestData requestData )
32
31
{
@@ -35,14 +34,17 @@ public RequestDataContent(RequestData requestData)
35
34
if ( requestData . HttpCompression )
36
35
Headers . ContentEncoding . Add ( "gzip" ) ;
37
36
38
- Task OnStreamAvailable ( PostData data , Stream stream , HttpContent content , TransportContext context )
37
+ Task OnStreamAvailable ( RequestData data , Stream stream , HttpContent content , TransportContext context )
39
38
{
40
- if ( _requestData . HttpCompression )
39
+ if ( data . HttpCompression )
41
40
stream = new GZipStream ( stream , CompressionMode . Compress , false ) ;
41
+
42
42
using ( stream )
43
- data . Write ( stream , requestData . ConnectionSettings ) ;
43
+ data . PostData . Write ( stream , data . ConnectionSettings ) ;
44
+
44
45
return Task . CompletedTask ;
45
46
}
47
+
46
48
_onStreamAvailable = OnStreamAvailable ;
47
49
}
48
50
public RequestDataContent ( RequestData requestData , CancellationToken token )
@@ -52,13 +54,15 @@ public RequestDataContent(RequestData requestData, CancellationToken token)
52
54
if ( requestData . HttpCompression )
53
55
Headers . ContentEncoding . Add ( "gzip" ) ;
54
56
55
- async Task OnStreamAvailable ( PostData data , Stream stream , HttpContent content , TransportContext context )
57
+ async Task OnStreamAvailable ( RequestData data , Stream stream , HttpContent content , TransportContext context )
56
58
{
57
- if ( _requestData . HttpCompression )
59
+ if ( data . HttpCompression )
58
60
stream = new GZipStream ( stream , CompressionMode . Compress , false ) ;
61
+
59
62
using ( stream )
60
- await data . WriteAsync ( stream , requestData . ConnectionSettings , token ) . ConfigureAwait ( false ) ;
63
+ await data . PostData . WriteAsync ( stream , data . ConnectionSettings , token ) . ConfigureAwait ( false ) ;
61
64
}
65
+
62
66
_onStreamAvailable = OnStreamAvailable ;
63
67
}
64
68
@@ -73,14 +77,9 @@ async Task OnStreamAvailable(PostData data, Stream stream, HttpContent content,
73
77
[ SuppressMessage ( "Microsoft.Design" , "CA1031:DoNotCatchGeneralExceptionTypes" , Justification = "Exception is passed as task result." ) ]
74
78
protected override async Task SerializeToStreamAsync ( Stream stream , TransportContext context )
75
79
{
76
-
77
- var data = _requestData . PostData ;
78
- if ( data == null ) return ;
79
-
80
80
var serializeToStreamTask = new TaskCompletionSource < bool > ( ) ;
81
-
82
81
var wrappedStream = new CompleteTaskOnCloseStream ( stream , serializeToStreamTask ) ;
83
- await _onStreamAvailable ( data , wrappedStream , this , context ) . ConfigureAwait ( false ) ;
82
+ await _onStreamAvailable ( _requestData , wrappedStream , this , context ) . ConfigureAwait ( false ) ;
84
83
await serializeToStreamTask . Task . ConfigureAwait ( false ) ;
85
84
}
86
85
@@ -113,7 +112,6 @@ protected override void Dispose(bool disposing)
113
112
base . Dispose ( ) ;
114
113
}
115
114
116
-
117
115
public override void Close ( ) => _serializeToStreamTask . TrySetResult ( true ) ;
118
116
}
119
117
@@ -195,6 +193,8 @@ public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, As
195
193
public override void EndWrite ( IAsyncResult asyncResult ) => _innerStream . EndWrite ( asyncResult ) ;
196
194
197
195
public override void WriteByte ( byte value ) => _innerStream . WriteByte ( value ) ;
196
+
197
+ public override void Close ( ) => _innerStream . Close ( ) ;
198
198
}
199
199
}
200
200
}
0 commit comments