Make CancelToken be able to cancel response streams from ResponseType.stream requests

[tomekit]

Package

dio

Version

5.3.3

Operating-System

Linux

Output of flutter doctor -v

[✓] Flutter (Channel stable, 3.13.9, on Ubuntu 23.04 6.2.0-36-generic, locale en_US.UTF-8)
    • Flutter version 3.13.9 on channel stable at /opt/flutter
    • Upstream repository https://github.com/flutter/flutter.git
    • Framework revision d211f42860 (12 days ago), 2023-10-25 13:42:25 -0700
    • Engine revision 0545f8705d
    • Dart version 3.1.5
    • DevTools version 2.25.0

[✓] Android toolchain - develop for Android devices (Android SDK version 33.0.1)
    • Android SDK at /home/tomek/Android/Sdk
    • Platform android-34, build-tools 33.0.1
    • ANDROID_HOME = /home/tomek/Android/Sdk
    • Java binary at: /opt/android-studio/jbr/bin/java
    • Java version OpenJDK Runtime Environment (build 17.0.6+0-17.0.6b829.9-10027231)
    • All Android licenses accepted.

[✓] Chrome - develop for the web
    • Chrome at google-chrome

[✓] Linux toolchain - develop for Linux desktop
    • Ubuntu clang version 15.0.7
    • cmake version 3.25.1
    • ninja version 1.11.1
    • pkg-config version 1.8.1

[✓] Android Studio (version 2022.3)
    • Android Studio at /opt/android-studio
    • Flutter plugin version 75.1.2
    • Dart plugin can be installed from:
      🔨 https://plugins.jetbrains.com/plugin/6351-dart
    • Java version OpenJDK Runtime Environment (build 17.0.6+0-17.0.6b829.9-10027231)

[✓] VS Code (version 1.78.2)
    • VS Code at /usr/share/code
    • Flutter extension can be installed from:
      🔨 https://marketplace.visualstudio.com/items?itemName=Dart-Code.flutter

[✓] Connected device (2 available)
    • Linux (desktop) • linux  • linux-x64      • Ubuntu 23.04 6.2.0-36-generic
    • Chrome (web)    • chrome • web-javascript • Google Chrome 114.0.5735.198

[✓] Network resources
    • All expected network resources are available.

• No issues found!

Dart Version

3.1.5

Steps to Reproduce

Run: https://github.com/tomekit/dio_cancel_issue_stream_get

or use below main.dart file:

main.dart

import 'package:dio/dio.dart';
import 'package:flutter/material.dart';

void main() {
  runApp(const MyApp());
}

class MyApp extends StatelessWidget {
  const MyApp({super.key});
  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Dio cancellation issue',
      theme: ThemeData(
        colorScheme: ColorScheme.fromSeed(seedColor: Colors.deepPurpleAccent),
        useMaterial3: true,
      ),
      home: const MyHomePage(title: 'Dio cancellation issue'),
    );
  }
}

class MyHomePage extends StatefulWidget {
  const MyHomePage({super.key, required this.title});
  final String title;

  @override
  State<MyHomePage> createState() => _MyHomePageState();
}

class _MyHomePageState extends State<MyHomePage> {

  int _downloaded = 0;
  bool _running = false;
  Exception? _failed;
  CancelToken cancelToken = CancelToken();

  @override
  Widget build(BuildContext context) {
    return Scaffold(
        appBar: AppBar(
          backgroundColor: Theme.of(context).colorScheme.inversePrimary,
          title: Text(widget.title),
        ),
        body: Center(
          child: Column(
            mainAxisAlignment: MainAxisAlignment.center,
            children: <Widget>[
              Text('_downloaded: ${_downloaded}, _running: ${_running}, _failed: ${_failed?.toString()},')
            ],
          ),
        ),
        floatingActionButton:
        Stack(
          children: <Widget>[
            Align(
              alignment: Alignment.bottomLeft,
              child: FloatingActionButton(
                onPressed: () async {
                  fetch(ResponseType.bytes);
                },
                tooltip: 'Start bytes',
                child: const Icon(Icons.start),
              ),
            ),
            Align(
              alignment: Alignment.bottomCenter,
              child: FloatingActionButton(
                onPressed: () async {
                  fetch(ResponseType.stream);
                },
                tooltip: 'Start stream',
                child: const Icon(Icons.start),
              ),
            ),
            Align(
              alignment: Alignment.bottomRight,
              child: FloatingActionButton(
                onPressed: () => cancelToken.cancel(),
                tooltip: 'Cancel',
                child: const Icon(Icons.cancel),
              ),
            ),
          ],
        )
    );
  }

  fetch(ResponseType responseType) async {
    final url = "http://ash-speed.hetzner.com/100MB.bin";

    final dio = Dio();

    final options = Options(
      // responseType: ResponseType.bytes, // Works fine
      // responseType: ResponseType.stream, // No effect
      responseType: responseType
    );

    // For ResponseType.bytes purposes
    onReceiveProgress(int sent, int __) {
      setState(() {
        _downloaded = sent;
      });
      return;
    }

    try {

      setState(() {
        _running = true;
        _failed = null;
        cancelToken = CancelToken();
      });

      final response = await dio.get(url, options: options, cancelToken: cancelToken, onReceiveProgress: onReceiveProgress);
      final status = response.statusCode;

      var downloadedTotalStream = 0;
      if (status == 200) {
        await for (final List<int> dataChunk in response.data.stream) {
          downloadedTotalStream += dataChunk.length;
          setState(() {
            _downloaded = downloadedTotalStream;
          });
        }
      } else {
        setState(() {
          _failed = Exception("Status code: $status");
        });
      }

    } catch(e) {
      setState(() {
        _failed = e as Exception;
      });
    }

    setState(() {
      _running = false;
    });
  }
}

Click middle "Start stream" button to start streamed download, then click bottom right: "Cancel" button. Nothing happens in the release mode. In the debug mode your IDE (in my case Android Studio) may (it doesn't happen consistently) capture uncaught exception.

That's unexpected behavior, request doesn't get cancelled.

Click bottom left: "Start bytes" button to start bytes download, then click bottom right: "Cancel" button. DioException is thrown. That's expected behavior, request cancelling takes place.

Expected Result

DioException shall be thrown with type set to DioExceptionType.cancel even if: responseType: ResponseType.stream is used.

Actual Result

DioException isn't thrown when request with responseType: ResponseType.stream gets cancelled.

Test app:

[kuhnroyal]

I agree, thanks for the reproduction. I think I have a solution but testing this is kinda hard.
Do you think #1953 is the same problem?

[AlexV525]

I've been investigated about the cause of the issue.

When we request stream responses, the response body returns immediately.

dio/dio/lib/src/transformers/sync_transformer.dart

Lines 60 to 63 in bedcc54

	// Do not handled the body for streams.
	if (responseType == ResponseType.stream) {
	return responseBody;
	}

When you get the response, the request flow is finished. Nothing in the queue was still awaiting, thus the cancel token is no longer effective (because the request is, by a specific aspect, ended). From the flow, we can say the behavior is as desired.

Given the above fact, because you have already obtained the stream, you can use other mechanisms rather than cancel tokens to stop the stream reading.

The issue is nothing related to #1953 and #2028. I understand everyone's issue is urgent to themselves, but referencing unrelated issues will cause redundant information for people who subscribed to those issues, and increase the complexibility for maintainers to find the root cause.

Still, I've tried to improve this and managed to throw the cancel exception for the stream.

But it doesn't look like a proper way to solve it because it depends on implementation in adapters. So I gonna leave this open and continue to investigate to see if we can have a proper solution.

[kuhnroyal]

@AlexV525 I already had some changes and tests for this - see #2032