added onUploadProgress to MultipartRequest.new and Client.send. (adjusted file movements and from new fork)

Mehmetyaz · Mehmetyaz · commit a4d001df73f7 · 2022-07-02T01:42:28.000+03:00
diff --git a/.gitignore b/.gitignore
@@ -2,3 +2,4 @@
 .dart_tool
 .packages
 pubspec.lock
+.idea
diff --git a/.idea/.gitignore b/.idea/.gitignore
diff --git a/pkgs/http/lib/retry.dart b/pkgs/http/lib/retry.dart
@@ -8,6 +8,7 @@ import 'dart:math' as math;
 import 'package:async/async.dart';
 
 import 'http.dart';
+import 'src/utils.dart';
 
 /// An HTTP client wrapper that automatically retries failing requests.
 class RetryClient extends BaseClient {
@@ -101,7 +102,8 @@ class RetryClient extends BaseClient {
         );
 
   @override
-  Future<StreamedResponse> send(BaseRequest request) async {
+  Future<StreamedResponse> send(BaseRequest request,
+      {OnUploadProgress? onUploadProgress}) async {
     final splitter = StreamSplitter(request.finalize());
 
     var i = 0;
diff --git a/pkgs/http/lib/src/base_client.dart b/pkgs/http/lib/src/base_client.dart
@@ -12,6 +12,7 @@ import 'exception.dart';
 import 'request.dart';
 import 'response.dart';
 import 'streamed_response.dart';
+import 'utils.dart';
 
 /// The abstract base class for an HTTP client.
 ///
@@ -67,8 +68,16 @@ abstract class BaseClient implements Client {
   /// state of the stream; it could have data written to it asynchronously at a
   /// later point, or it could already be closed when it's returned. Any
   /// internal HTTP errors should be wrapped as [ClientException]s.
+  ///
+  /// If [onUploadProgress] callback is provided and length is computable,
+  /// [onUploadProgress] will execute for each chunk was sent.
+  ///
+  /// lengthComputable :
+  ///     library.html : xhr.lengthComputable
+  ///     library.io : content-length is provided (MultipartRequest provide)
   @override
-  Future<StreamedResponse> send(BaseRequest request);
+  Future<StreamedResponse> send(BaseRequest request,
+      {OnUploadProgress? onUploadProgress});
 
   /// Sends a non-streaming [Request] and returns a non-streaming [Response].
   Future<Response> _sendUnstreamed(
diff --git a/pkgs/http/lib/src/base_request.dart b/pkgs/http/lib/src/base_request.dart
@@ -89,14 +89,26 @@ abstract class BaseRequest {
   bool _finalized = false;
 
   static final _tokenRE = RegExp(r"^[\w!#%&'*+\-.^`|~]+$");
+
   static String _validateMethod(String method) {
     if (!_tokenRE.hasMatch(method)) {
       throw ArgumentError.value(method, 'method', 'Not a valid method');
     }
     return method;
   }
 
-  BaseRequest(String method, this.url)
+  /// On upload progress callback.
+  ///
+  /// If defined, this callback will be called when the upload progress changes.
+  ///
+  /// In browser, uses XMLHttpRequest's "xhr.upload.onLoad" event.
+  ///
+  /// In IO, uses the yield length of the stream. The total length of the bytes
+  /// yielded by the stream at any given moment is "uploaded" and the total
+  /// length of the stream is "total"
+  final OnUploadProgress? onUploadProgress;
+
+  BaseRequest(String method, this.url, {this.onUploadProgress})
       : method = _validateMethod(method),
         headers = LinkedHashMap(
             equals: (key1, key2) => key1.toLowerCase() == key2.toLowerCase(),
@@ -130,7 +142,8 @@ abstract class BaseRequest {
     var client = Client();
 
     try {
-      var response = await client.send(this);
+      var response =
+          await client.send(this, onUploadProgress: onUploadProgress);
       var stream = onDone(response.stream, client.close);
       return StreamedResponse(ByteStream(stream), response.statusCode,
           contentLength: response.contentLength,
diff --git a/pkgs/http/lib/src/browser_client.dart b/pkgs/http/lib/src/browser_client.dart
@@ -11,6 +11,7 @@ import 'base_request.dart';
 import 'byte_stream.dart';
 import 'exception.dart';
 import 'streamed_response.dart';
+import 'utils.dart';
 
 /// Create a [BrowserClient].
 ///
@@ -38,8 +39,16 @@ class BrowserClient extends BaseClient {
   bool withCredentials = false;
 
   /// Sends an HTTP request and asynchronously returns the response.
+  ///
+  /// If [onUploadProgress] callback is provided and length is computable,
+  /// [onUploadProgress] will execute for each chunk was sent.
+  ///
+  /// lengthComputable :
+  ///     library.html : xhr.lengthComputable
+  ///     library.io : content-length is provided (MultipartRequest provide)
   @override
-  Future<StreamedResponse> send(BaseRequest request) async {
+  Future<StreamedResponse> send(BaseRequest request,
+      {OnUploadProgress? onUploadProgress}) async {
     var bytes = await request.finalize().toBytes();
     var xhr = HttpRequest();
     _xhrs.add(xhr);
@@ -51,6 +60,14 @@ class BrowserClient extends BaseClient {
 
     var completer = Completer<StreamedResponse>();
 
+    if (onUploadProgress != null) {
+      xhr.upload.onLoad.listen((event) {
+        if (event.lengthComputable) {
+          onUploadProgress(event.total!, event.loaded!);
+        }
+      });
+    }
+
     unawaited(xhr.onLoad.first.then((_) {
       var body = (xhr.response as ByteBuffer).asUint8List();
       completer.complete(StreamedResponse(
diff --git a/pkgs/http/lib/src/client.dart b/pkgs/http/lib/src/client.dart
@@ -17,6 +17,7 @@ import 'client_stub.dart'
 import 'exception.dart';
 import 'response.dart';
 import 'streamed_response.dart';
+import 'utils.dart';
 
 /// The interface for HTTP clients that take care of maintaining persistent
 /// connections across multiple requests to the same server.
@@ -139,7 +140,11 @@ abstract class Client {
   Future<Uint8List> readBytes(Uri url, {Map<String, String>? headers});
 
   /// Sends an HTTP request and asynchronously returns the response.
-  Future<StreamedResponse> send(BaseRequest request);
+  ///
+  /// If [onUploadProgress] defined, the callback will be called when the
+  /// upload progress changes.
+  Future<StreamedResponse> send(BaseRequest request,
+      {OnUploadProgress? onUploadProgress});
 
   /// Closes the client and cleans up any resources associated with it.
   ///
diff --git a/pkgs/http/lib/src/io_client.dart b/pkgs/http/lib/src/io_client.dart
@@ -2,12 +2,15 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
 import 'dart:io';
 
+import '../http.dart' show ByteStream;
 import 'base_client.dart';
 import 'base_request.dart';
 import 'exception.dart';
 import 'io_streamed_response.dart';
+import 'utils.dart';
 
 /// Create an [IOClient].
 ///
@@ -22,6 +25,7 @@ BaseClient createClient() => IOClient();
 class _ClientSocketException extends ClientException
     implements SocketException {
   final SocketException cause;
+
   _ClientSocketException(SocketException e, Uri url)
       : cause = e,
         super(e.message, url);
@@ -44,26 +48,50 @@ class IOClient extends BaseClient {
   IOClient([HttpClient? inner]) : _inner = inner ?? HttpClient();
 
   /// Sends an HTTP request and asynchronously returns the response.
+  ///
+  /// If [onUploadProgress] callback is provided and length is computable,
+  /// [onUploadProgress] will execute for each chunk was sent.
+  ///
+  /// lengthComputable :
+  ///     library.html : xhr.lengthComputable
+  ///     library.io : content-length is provided (MultipartRequest provide)
   @override
-  Future<IOStreamedResponse> send(BaseRequest request) async {
+  Future<IOStreamedResponse> send(BaseRequest request,
+      {OnUploadProgress? onUploadProgress}) async {
     if (_inner == null) {
       throw ClientException(
           'HTTP request failed. Client is already closed.', request.url);
     }
 
     var stream = request.finalize();
 
+    ByteStream? handledStream;
+
+    var contentLength = request.contentLength;
+    if (onUploadProgress != null && contentLength != null) {
+      var load = 0;
+      handledStream =
+          ByteStream(stream.transform(StreamTransformer.fromBind((d) async* {
+        await for (var data in d) {
+          load += data.length;
+          onUploadProgress(contentLength, load);
+          yield data;
+        }
+      })));
+    }
+
     try {
       var ioRequest = (await _inner!.openUrl(request.method, request.url))
         ..followRedirects = request.followRedirects
         ..maxRedirects = request.maxRedirects
-        ..contentLength = (request.contentLength ?? -1)
+        ..contentLength = (contentLength ?? -1)
         ..persistentConnection = request.persistentConnection;
       request.headers.forEach((name, value) {
         ioRequest.headers.set(name, value);
       });
 
-      var response = await stream.pipe(ioRequest) as HttpClientResponse;
+      var response =
+          await (handledStream ?? stream).pipe(ioRequest) as HttpClientResponse;
 
       var headers = <String, String>{};
       response.headers.forEach((key, values) {
diff --git a/pkgs/http/lib/src/mock_client.dart b/pkgs/http/lib/src/mock_client.dart
@@ -2,13 +2,16 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
+
 import 'base_client.dart';
 import 'base_request.dart';
 import 'byte_stream.dart';
 import 'request.dart';
 import 'response.dart';
 import 'streamed_request.dart';
 import 'streamed_response.dart';
+import 'utils.dart';
 
 // TODO(nweiz): once Dart has some sort of Rack- or WSGI-like standard for
 // server APIs, MockClient should conform to it.
@@ -65,9 +68,26 @@ class MockClient extends BaseClient {
         });
 
   @override
-  Future<StreamedResponse> send(BaseRequest request) async {
+  Future<StreamedResponse> send(BaseRequest request,
+      {OnUploadProgress? onUploadProgress}) async {
     var bodyStream = request.finalize();
-    return await _handler(request, bodyStream);
+    ByteStream? handledStream;
+
+    var contentLength = request.contentLength;
+
+    if (onUploadProgress != null && contentLength != null) {
+      var load = 0;
+      handledStream = ByteStream(
+          bodyStream.transform(StreamTransformer.fromBind((d) async* {
+        await for (var data in d) {
+          load += data.length;
+          onUploadProgress(contentLength, load);
+          yield data;
+        }
+      })));
+    }
+
+    return await _handler(request, handledStream ?? bodyStream);
   }
 }
 
diff --git a/pkgs/http/lib/src/multipart_request.dart b/pkgs/http/lib/src/multipart_request.dart
@@ -45,7 +45,14 @@ class MultipartRequest extends BaseRequest {
   /// The list of files to upload for this request.
   final files = <MultipartFile>[];
 
-  MultipartRequest(String method, Uri url) : super(method, url);
+  /// If [onUploadProgress] callback is provided and length is computable,
+  /// [onUploadProgress] will execute for each chunk was sent.
+  ///
+  /// lengthComputable :
+  ///     library.html : xhr.lengthComputable
+  ///     library.io : content-length is provided (MultipartRequest provide)
+  MultipartRequest(String method, Uri url, {OnUploadProgress? onUploadProgress})
+      : super(method, url, onUploadProgress: onUploadProgress);
 
   /// The total length of the request body, in bytes.
   ///
diff --git a/pkgs/http/lib/src/utils.dart b/pkgs/http/lib/src/utils.dart
@@ -74,3 +74,8 @@ Stream<T> onDone<T>(Stream<T> stream, void Function() onDone) =>
       sink.close();
       onDone();
     }));
+
+/// On upload progress callback.
+///
+/// See : "MultipartRequest"
+typedef OnUploadProgress = void Function(int uploaded, int total);
diff --git a/pkgs/http/test/io/client_test.dart b/pkgs/http/test/io/client_test.dart
@@ -8,20 +8,23 @@ import 'dart:io';
 
 import 'package:http/http.dart' as http;
 import 'package:http/io_client.dart' as http_io;
+import 'package:http/src/utils.dart';
 import 'package:test/test.dart';
 
 import '../utils.dart';
 
 class TestClient extends http.BaseClient {
   @override
-  Future<http.StreamedResponse> send(http.BaseRequest request) {
+  Future<http.StreamedResponse> send(http.BaseRequest request,
+      {OnUploadProgress? onUploadProgress}) {
     throw UnimplementedError();
   }
 }
 
 class TestClient2 extends http.BaseClient {
   @override
-  Future<http.StreamedResponse> send(http.BaseRequest request) {
+  Future<http.StreamedResponse> send(http.BaseRequest request,
+      {OnUploadProgress? onUploadProgress}) {
     throw UnimplementedError();
   }
 }
diff --git a/pkgs/http/test/io/http_test.dart b/pkgs/http/test/io/http_test.dart
@@ -3,15 +3,16 @@
 // BSD-style license that can be found in the LICENSE file.
 
 @TestOn('vm')
-
 import 'package:http/http.dart' as http;
+import 'package:http/src/utils.dart';
 import 'package:test/test.dart';
 
 import '../utils.dart';
 
 class TestClient extends http.BaseClient {
   @override
-  Future<http.StreamedResponse> send(http.BaseRequest request) {
+  Future<http.StreamedResponse> send(http.BaseRequest request,
+      {OnUploadProgress? onUploadProgress}) {
     throw UnimplementedError();
   }
 }
diff --git a/pkgs/http/test/io/streamed_request_test.dart b/pkgs/http/test/io/streamed_request_test.dart
@@ -50,4 +50,32 @@ void main() {
     var response = await request.send();
     expect(await utf8.decodeStream(response.stream), equals('body'));
   });
+
+  test('sends a MultipartRequest with onUploadProgress', () async {
+    var totalLoaded = 0;
+
+    var loadedNotifications = <int>[];
+
+    int? contentLength;
+
+    final request =
+        http.MultipartRequest('GET', serverUrl.resolve('/multipart'),
+            onUploadProgress: (total, loaded) {
+      contentLength = total;
+      totalLoaded = loaded;
+      loadedNotifications.add(loaded);
+    });
+    request.files.add(http.MultipartFile.fromBytes(
+        'file', List.generate(1500, (index) => 100)));
+
+    var response = await (await request.send()).stream.bytesToString();
+
+    print(response);
+
+    expect(response, '1739');
+    expect(contentLength, 1739);
+    expect(totalLoaded, 1739);
+    expect(loadedNotifications.length, 5);
+    expect(loadedNotifications, [74, 161, 1661, 1663, 1739]);
+  });
 }
diff --git a/pkgs/http/test/stub_server.dart b/pkgs/http/test/stub_server.dart
