Handle large number of http requests at once in flutter

Question

Requirement:

I've 1000 URLs of PDF, from which I want to create the zip.

Function:

static Future<bool> downloadZip(List<String> urls) {
    int counter = 0
    for(String url in urls) {
        bytesForFileAtUrl(url).then((bytes){
            counter++;
            if(bytes != null) {
               //Add bytes to the zip encoder
            }
            if(counter == urls.length) {
               //close the zip encoder
               //download the zip file
            }
        }
      }
  }

  static Future<Uint8List> bytesForFileAtUrl(String url) async {
      try {
          http.Response response = await http.get(url);
          return response?.bodyBytes;
      } catch (e) {
          print('Error getting bytes: ${e.toString()}');
          return null;
      }
  }

Problem:

When there is small number of requests its working fine. But in case large number of requests I'm getting the exception: SocketException: Connection failed (OS Error: Too many open files, errno = 24) as reported here. This might because of memory issue on my mobile device, as its working fine on the web platform.

I've tried other solutions like the one using Future.wait() suggested here, but its not working and device is hanged.

Whats the best solution for this scenario ?

Worst Solution:

Using await keyword. No exception at all, but then its taking too long for the overall process to complete.

for(String url in urls) {
    final bytes = await bytesForFileAtUrl(url);
    ...
    ...
}

Answer 1

static Future < bool > downloadZip(List < String > urls) {
  int counter = 0;
  urls.forEach(async() {
    await bytesForFileAtUrl(url).then((bytes){
      counter++;
      if (bytes != null) {
        //Add bytes to the zip encoder
      }
      if (counter == urls.length) {
        //close the zip encoder
        //download the zip file
      }
    });
  });
}

  static Future < Uint8List > bytesForFileAtUrl(String url) async {
  try {
    http.Response response = await http.get(url);
    return response?.bodyBytes;
  } catch (e) {
    print('Error getting bytes: ${e.toString()}');
    return null;
  }
}

maybe try this ?

Answer 2

I had a similar issue managing file uploads.

Firstly, see this post. As you are experiencing SocketException's, it's likely because too many connections are open.

To manage this, I built a class to manage the number of active operations while allowing multiple http calls to run concurrently. If you need these to finish in a particular order, you may want to combine this method with Future.wait().

This method relies on a StreamController to handle the http responses. Review the StreamController class documentation as there are several events (not used below) that you can listen to which may be helpful.

To use the class below (in simple terms):

1. Declare the DownloadManager() in your main class/Widget
2. Call the download() function
3. download() creates a StreamController and starts to loop through the url's
4. The number of concurrent downloads is controlled via _maxActiveConnections
5. The loop will skip when _activeConnections == _maxActiveConnections
6. Once the loop ends, the function will wait for any remaining http requests to complete
7. After the await downloadStream.done is complete, return your result or do whatever you need to with the result.

import 'dart:async';
import 'dart:math' as math;
import 'package:http/http.dart' as http;
import 'package:flutter/services.dart';

class MyWidgetClass {
  DownloadManager downloadManager = DownloadManager();

  Future<bool> downloadZip(List<String> urls) async {
    final result = await downloadManager.download(urls);
    return result;
  }
}

class DownloadManager {
  /// Max number of http calls allowed at any time
  final _maxActiveConnections = 5;

  /// Time before retries when waiting for a free connection
  final _waitDuration = const Duration(milliseconds: 250);

  /// Current index to be uploaded when the downloader is [_enabled]
  int _index = 0;

  /// The number of connections currently active
  int _activeConnections = 0;

  /// Allow additional http calls to be added to the queue
  bool get _enabled => _activeConnections < _maxActiveConnections;

  /// Track the number of successful downloads (optional, useful if you need to
  /// know if all downloads are completed successfully)
  int _successCount = 0;

  /// the urls.length value
  int _totalDownloads = 0;

  /// number of http operations that have completed
  int _downloadsCompleted = 0;

  Future<bool> download(List<String> urls) async {
    _totalDownloads = urls.length;
    _downloadsCompleted = 0;
    _activeConnections = 0;
    _index = 0;
    _successCount = 0;

    // declare your encoder here
    // dynamic exampleZipEncoder = ...;

    // declare a Uint8List stream
    final downloadStream = StreamController<Uint8List?>();

    // listens to the http responses as they return
    downloadStream.stream.listen(
      // Fired when a download is finished
      (Uint8List? event) {
        print('Success!');

        if (event != null) {
          // Add bytes to the zip encoder
          //  exampleZipEncoder.add(event);
          _successCount++;
        }

        _setActiveConnections(-1);
      },
      onError: (error) {
        print('Error! $error');

        // Handle case when http fails

        _setActiveConnections(-1);
      },
    );

    // Run the downloads when _enabled = true
    while (_index < urls.length) {
      if (_enabled) {
        _bytesForFileAtUrl(urls[_index], downloadStream);
        _index++;
        _setActiveConnections(1);
      }
      await Future.delayed(_waitDuration);
    }

    // waits for all/any remaining http's to complete
    await downloadStream.done;

    // Close the zip encoder here, or you can add it to the StreamController
    // done event listener

    // await exampleZipEncoder.close();

    // return exampleZipEncoder;
    return _successCount == urls.length;
  }

  /// Handle download of a file
  Future<void> _bytesForFileAtUrl(String url, StreamController stream) async {
    try {
      http.Response response = await http.get(Uri(path: url));
      if (response.bodyBytes.isNotEmpty) {
        stream.add(response.bodyBytes);
      } else {
        stream.addError('Error: ${response.statusCode}');
      }
    } catch (e) {
      stream.addError('$e');
    }

    _downloadsCompleted++;

    // once all files have been downloaded, close the stream
    if (_downloadsCompleted == _totalDownloads) {
      stream.close();
    }
  }

  /// Increments/Decrements the number of [_activeConnections]
  void _setActiveConnections(int change) =>
      _activeConnections = math.max(0, _activeConnections + change);
}