@@ -59,6 +59,118 @@ void testResponseBodyStreamed(Client client,
5959 expect (response.statusCode, 200 );
6060 });
6161
62+ test ('pausing response stream after events' , () async {
63+ final response = await client.send (Request ('GET' , Uri .http (host, '' )));
64+ expect (response.reasonPhrase, 'OK' );
65+ expect (response.statusCode, 200 );
66+
67+ // The server responds with a streamed response of lines containing
68+ // incrementing integers. Verify that pausing the stream after each one
69+ // does not cause any missed lines.
70+ final stream = response.stream
71+ .transform (const Utf8Decoder ())
72+ .transform (const LineSplitter ())
73+ .map (int .parse);
74+ var expectedLine = 0 ;
75+ final cancelCompleter = Completer <void >();
76+ late StreamSubscription <int > subscription;
77+
78+ subscription = stream.listen ((line) async {
79+ expect (line, expectedLine);
80+ expectedLine++ ;
81+
82+ if (expectedLine == 10 ) {
83+ subscription.pause ();
84+ Future .delayed (
85+ const Duration (seconds: 1 ), () => subscription.resume ());
86+ }
87+
88+ if (expectedLine == 100 ) {
89+ cancelCompleter.complete (subscription.cancel ());
90+ }
91+ await pumpEventQueue ();
92+ });
93+
94+ await cancelCompleter.future;
95+ expect (expectedLine, 100 );
96+ });
97+
98+ test ('pausing response stream asynchronously' , () async {
99+ final response = await client.send (Request ('GET' , Uri .http (host, '' )));
100+ expect (response.reasonPhrase, 'OK' );
101+ expect (response.statusCode, 200 );
102+
103+ final originalSubscription = response.stream
104+ .transform (const Utf8Decoder ())
105+ .transform (const LineSplitter ())
106+ .map (int .parse)
107+ .listen (null );
108+ var expectedLine = 0 ;
109+ await for (final line in SubscriptionStream (originalSubscription)) {
110+ expect (line, expectedLine);
111+ expectedLine++ ;
112+ if (expectedLine == 100 ) {
113+ break ;
114+ }
115+
116+ // Instead of pausing the subscription in response to an event, pause it
117+ // after the event has already been delivered.
118+ Timer .run (() {
119+ originalSubscription.pause (Future (pumpEventQueue));
120+ });
121+ }
122+ });
123+
124+ test ('cancel paused stream' , () async {
125+ final response = await client.send (Request ('GET' , Uri .http (host, '' )));
126+ expect (response.reasonPhrase, 'OK' );
127+ expect (response.statusCode, 200 );
128+
129+ final completer = Completer <void >();
130+ late StreamSubscription <String > subscription;
131+ subscription = response.stream
132+ .transform (const Utf8Decoder ())
133+ .transform (const LineSplitter ())
134+ .listen ((line) async {
135+ subscription.pause ();
136+
137+ completer.complete (Future (() async {
138+ await pumpEventQueue ();
139+ await subscription.cancel ();
140+ }));
141+ });
142+
143+ await completer.future;
144+ });
145+
146+ test ('cancel paused stream via abortable request' , () async {
147+ final abortTrigger = Completer <void >();
148+ final response = await client.send (AbortableRequest (
149+ 'GET' , Uri .http (host, '' ),
150+ abortTrigger: abortTrigger.future));
151+ expect (response.reasonPhrase, 'OK' );
152+ expect (response.statusCode, 200 );
153+
154+ late StreamSubscription <String > subscription;
155+ subscription = response.stream
156+ .transform (const Utf8Decoder ())
157+ .transform (const LineSplitter ())
158+ .listen ((line) {
159+ if (! abortTrigger.isCompleted) {
160+ abortTrigger.complete ();
161+ }
162+ });
163+
164+ final aborted = expectLater (subscription.asFuture <void >(),
165+ throwsA (isA <RequestAbortedException >()));
166+ await abortTrigger.future;
167+
168+ // We need to resume the subscription after the response has been
169+ // cancelled to record that error event.
170+ subscription.resume ();
171+ await aborted;
172+ });
173+
62174 test ('cancel streamed response' , () async {
63175 final request = Request ('GET' , Uri .http (host, '' ));
64176 final response = await client.send (request);
@@ -77,5 +189,63 @@ void testResponseBodyStreamed(Client client,
77189 });
78190 await cancelled.future;
79191 });
192+
193+ test ('cancelling stream subscription after chunk' , () async {
194+ // Request a 10s delay between subsequent lines.
195+ const delayMillis = 10000 ;
196+ final request = Request ('GET' , Uri .http (host, '$delayMillis ' ));
197+ final response = await client.send (request);
198+ expect (response.reasonPhrase, 'OK' );
199+ expect (response.statusCode, 200 );
200+
201+ final cancelled = Completer <void >();
202+ var stopwatch = Stopwatch ();
203+ final subscription = response.stream
204+ .transform (const Utf8Decoder ())
205+ .transform (const LineSplitter ())
206+ .listen (null );
207+ subscription.onData ((line) {
208+ stopwatch.start ();
209+ cancelled.complete (subscription.cancel ());
210+ expect (line, '0' );
211+ });
212+
213+ await cancelled.future;
214+ stopwatch.stop ();
215+
216+ // Receiving the first line and cancelling the stream should not wait for
217+ // the second line, which is sent much later.
218+ expect (stopwatch.elapsed.inMilliseconds, lessThan (delayMillis));
219+ });
220+
221+ test ('cancelling stream subscription after chunk with delay' , () async {
222+ // Request a 10s delay between subsequent lines.
223+ const delayMillis = 10000 ;
224+ final request = Request ('GET' , Uri .http (host, '$delayMillis ' ));
225+ final response = await client.send (request);
226+ expect (response.reasonPhrase, 'OK' );
227+ expect (response.statusCode, 200 );
228+
229+ var stopwatch = Stopwatch ()..start ();
230+ final done = Completer <void >();
231+ late StreamSubscription <String > sub;
232+ sub = response.stream
233+ .transform (utf8.decoder)
234+ .transform (const LineSplitter ())
235+ .listen ((line) {
236+ // Don't cancel in direct response to event, we want to test cancelling
237+ // while the client is actively waiting for data.
238+ Timer .run (() {
239+ stopwatch.start ();
240+ done.complete (sub.cancel ());
241+ });
242+ });
243+
244+ await done.future;
245+ stopwatch.stop ();
246+ // Receiving the first line and cancelling the stream should not wait for
247+ // the second line, which is sent much later.
248+ expect (stopwatch.elapsed.inMilliseconds, lessThan (delayMillis));
249+ });
80250 }, skip: canStreamResponseBody ? false : 'does not stream response bodies' );
81251}
0 commit comments