Skip to content

Commit bdac03d

Browse files
JaffaKetchupJaffaKetchup
committed
Make IOClient emit an error on the response stream when aborted
Integrated abort tests
1 parent 8563f19 commit bdac03d

File tree

3 files changed

+127
-56
lines changed

3 files changed

+127
-56
lines changed

‎pkgs/http/lib/src/io_client.dart‎

Lines changed: 51 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
// for details. All rights reserved. Use of this source code is governed by a
33
// BSD-style license that can be found in the LICENSE file.
44

5+
import 'dart:async';
56
import 'dart:io';
67

78
import 'abortable.dart';
@@ -124,15 +125,45 @@ class IOClient extends BaseClient {
124125
ioRequest.headers.set(name, value);
125126
});
126127

127-
Future<void>? canceller;
128+
// We can only abort the actual connection up until the point we obtain
129+
// the response.
130+
// After that point, the full response bytes are always available.
131+
// However, we instead inject an error into the response stream to match
132+
// the behaviour of `BrowserClient`.
133+
134+
StreamSubscription<List<int>>? subscription;
135+
final controller = StreamController<List<int>>(sync: true);
136+
128137
if (request case Abortable(:final abortTrigger?)) {
129-
canceller = abortTrigger
130-
.whenComplete(() => ioRequest.abort(const AbortedRequest()));
138+
abortTrigger.whenComplete(() async {
139+
if (subscription == null) {
140+
ioRequest.abort(const AbortedRequest());
141+
} else {
142+
if (!controller.isClosed) {
143+
controller.addError(const AbortedRequest());
144+
}
145+
await subscription.cancel();
146+
}
147+
await controller.close();
148+
});
131149
}
132150

133-
var response = await stream.pipe(ioRequest) as HttpClientResponse;
134-
135-
canceller?.ignore();
151+
final response = await stream.pipe(ioRequest) as HttpClientResponse;
152+
153+
subscription = response.listen(
154+
controller.add,
155+
onDone: controller.close,
156+
onError: (Object err, StackTrace stackTrace) {
157+
if (err is HttpException) {
158+
controller.addError(
159+
ClientException(err.message, err.uri),
160+
stackTrace,
161+
);
162+
} else {
163+
controller.addError(err, stackTrace);
164+
}
165+
},
166+
);
136167

137168
var headers = <String, String>{};
138169
response.headers.forEach((key, values) {
@@ -143,22 +174,20 @@ class IOClient extends BaseClient {
143174
});
144175

145176
return _IOStreamedResponseV2(
146-
response.handleError((Object error) {
147-
final httpException = error as HttpException;
148-
throw ClientException(httpException.message, httpException.uri);
149-
}, test: (error) => error is HttpException),
150-
response.statusCode,
151-
contentLength:
152-
response.contentLength == -1 ? null : response.contentLength,
153-
request: request,
154-
headers: headers,
155-
isRedirect: response.isRedirect,
156-
url: response.redirects.isNotEmpty
157-
? response.redirects.last.location
158-
: request.url,
159-
persistentConnection: response.persistentConnection,
160-
reasonPhrase: response.reasonPhrase,
161-
inner: response);
177+
controller.stream,
178+
response.statusCode,
179+
contentLength:
180+
response.contentLength == -1 ? null : response.contentLength,
181+
request: request,
182+
headers: headers,
183+
isRedirect: response.isRedirect,
184+
url: response.redirects.isNotEmpty
185+
? response.redirects.last.location
186+
: request.url,
187+
persistentConnection: response.persistentConnection,
188+
reasonPhrase: response.reasonPhrase,
189+
inner: response,
190+
);
162191
} on SocketException catch (error) {
163192
throw _ClientSocketException(error, request.url);
164193
} on HttpException catch (error) {

‎pkgs/http_client_conformance_tests/lib/src/abort_tests.dart‎

Lines changed: 71 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22
// for details. All rights reserved. Use of this source code is governed by a
33
// BSD-style license that can be found in the LICENSE file.
44

5+
import 'dart:async';
6+
57
import 'package:async/async.dart';
68
import 'package:http/http.dart';
79
import 'package:stream_channel/stream_channel.dart';
@@ -43,82 +45,119 @@ void testAbort(
4345
tearDownAll(() => httpServerChannel.sink.add(null));
4446

4547
test('before request', () async {
46-
final request = Request('GET', serverUrl);
47-
48-
// TODO: Trigger abort
48+
final abortTrigger = Completer<void>();
49+
final request = AbortableRequest(
50+
'GET',
51+
serverUrl,
52+
abortTrigger: abortTrigger.future,
53+
);
54+
abortTrigger.complete();
4955

5056
expect(
51-
client.send(request),
52-
throwsA(
53-
isA<ClientException>().having((e) => e.uri, 'uri', serverUrl)));
57+
client.send(request),
58+
throwsA(isA<AbortedRequest>()),
59+
);
5460
});
5561

5662
test('during request stream', () async {
57-
final request = StreamedRequest('POST', serverUrl);
63+
final abortTrigger = Completer<void>();
64+
65+
final request = AbortableStreamedRequest(
66+
'POST',
67+
serverUrl,
68+
abortTrigger: abortTrigger.future,
69+
);
5870

5971
final response = client.send(request);
6072
request.sink.add('Hello World'.codeUnits);
61-
// TODO: Trigger abort
73+
74+
abortTrigger.complete();
6275

6376
expect(
64-
response,
65-
throwsA(
66-
isA<ClientException>().having((e) => e.uri, 'uri', serverUrl)));
77+
response,
78+
throwsA(isA<AbortedRequest>()),
79+
);
6780
await request
6881
.sink.done; // Verify that the stream subscription was cancelled.
6982
}, skip: canStreamRequestBody ? false : 'does not stream request bodies');
7083

7184
test('after response', () async {
72-
final request = Request('GET', serverUrl);
85+
final abortTrigger = Completer<void>();
7386

87+
final request = AbortableRequest(
88+
'GET',
89+
serverUrl,
90+
abortTrigger: abortTrigger.future,
91+
);
7492
final response = await client.send(request);
7593

76-
// TODO: Trigger abort
94+
abortTrigger.complete();
7795

7896
expect(
79-
response.stream.single,
80-
throwsA(
81-
isA<ClientException>().having((e) => e.uri, 'uri', serverUrl)));
97+
response.stream.single,
98+
throwsA(isA<AbortedRequest>()),
99+
);
82100
});
83101

84102
test('while streaming response', () async {
85-
final request = Request('GET', serverUrl);
103+
final abortTrigger = Completer<void>();
86104

105+
final request = AbortableRequest(
106+
'GET',
107+
serverUrl,
108+
abortTrigger: abortTrigger.future,
109+
);
87110
final response = await client.send(request);
88111

89112
var i = 0;
90-
expect(
91-
response.stream.listen((data) {
92-
++i;
93-
if (i == 1000) {
94-
// TODO: Trigger abort
95-
}
96-
}).asFuture<void>(),
97-
throwsA(
98-
isA<ClientException>().having((e) => e.uri, 'uri', serverUrl)));
113+
final subscription = response.stream.listen((data) {
114+
++i;
115+
if (i == 1000) abortTrigger.complete();
116+
}).asFuture<void>();
117+
expect(subscription, throwsA(isA<AbortedRequest>()));
118+
await subscription.catchError((_) => null);
99119
expect(i, 1000);
100120
}, skip: canStreamResponseBody ? false : 'does not stream response bodies');
101121

102122
test('after streaming response', () async {
103-
final request = Request('GET', serverUrl);
123+
final abortTrigger = Completer<void>();
124+
125+
final request = AbortableRequest(
126+
'GET',
127+
serverUrl,
128+
abortTrigger: abortTrigger.future,
129+
);
104130

105131
final response = await client.send(request);
106132
await response.stream.drain<void>();
107-
// Trigger abort, should have no effect.
133+
134+
abortTrigger.complete();
108135
});
109136

110137
test('after response, client still useable', () async {
111-
final request = Request('GET', serverUrl);
138+
final abortTrigger = Completer<void>();
139+
140+
final request = AbortableRequest(
141+
'GET',
142+
serverUrl,
143+
abortTrigger: abortTrigger.future,
144+
);
112145

113146
final abortResponse = await client.send(request);
114-
// TODO: Trigger abort
147+
148+
abortTrigger.complete();
149+
150+
bool triggeredAbortedRequest = false;
115151
try {
116152
await abortResponse.stream.drain<void>();
117-
} on ClientException {}
153+
} on AbortedRequest {
154+
triggeredAbortedRequest = true;
155+
}
118156

119157
final response = await client.get(serverUrl);
120158
expect(response.statusCode, 200);
121-
expect(response.body, endsWith('10000\n'));
159+
expect(response.body, endsWith('9999\n'));
160+
expect(triggeredAbortedRequest, true);
122161
});
123162
});
124163
}

‎pkgs/http_client_conformance_tests/pubspec.yaml‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,18 +3,21 @@ description: >-
33
A library that tests whether implementations of package:http's `Client` class
44
behave as expected.
55
repository: https://github.com/dart-lang/http/tree/master/pkgs/http_client_conformance_tests
6-
76
publish_to: none
87

98
environment:
109
sdk: ^3.4.0
1110

1211
dependencies:
1312
async: ^2.8.2
14-
dart_style: '>=2.3.7 <4.0.0'
13+
dart_style: ">=2.3.7 <4.0.0"
1514
http: ^1.2.0
1615
stream_channel: ^2.1.1
1716
test: ^1.21.2
1817

18+
dependency_overrides:
19+
http:
20+
path: ../http
21+
1922
dev_dependencies:
2023
dart_flutter_team_lints: ^3.0.0

0 commit comments

Comments
 (0)