Merge pull request dart-archive/gcloud#134 from simolus3/gcs-upload-fix

Forward pauses for GCS uploads with unknown sizes

Author: jonasfj

pkgs/gcloud/CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.8.6-dev
+
+- Throttle streams piped into `Bucket.write` when the size is not known
+  beforehand.
+
 ## 0.8.5
 
 - Support the latest version 7.0.0 of the `googleapis` package.

pkgs/gcloud/lib/src/storage_impl.dart

@@ -518,8 +518,7 @@ class _MediaUploadStreamSink implements StreamSink<List<int>> {
   final storage_api.Object _object;
   final String? _predefinedAcl;
   final int? _length;
-  int _bufferLength = 0;
-  final List<List<int>> buffer = <List<int>>[];
+  final BytesBuilder _buffer = BytesBuilder();
   final _controller = StreamController<List<int>>(sync: true);
   late StreamSubscription _subscription;
   late StreamController<List<int>> _resumableController;
@@ -577,15 +576,23 @@ class _MediaUploadStreamSink implements StreamSink<List<int>> {
   void _onData(List<int> data) {
     assert(_state != _stateLengthKnown);
     if (_state == _stateProbingLength) {
-      buffer.add(data);
-      _bufferLength += data.length;
-      if (_bufferLength > _maxNormalUploadLength) {
+      _buffer.add(data);
+      if (_buffer.length > _maxNormalUploadLength) {
         // Start resumable upload.
         // TODO: Avoid using another stream-controller.
         _resumableController = StreamController<List<int>>(sync: true);
-        buffer.forEach(_resumableController.add);
+        _resumableController.add(_buffer.takeBytes());
         _startResumableUpload(_resumableController.stream, _length);
         _state = _stateDecidedResumable;
+
+        // At this point, we're forwarding events to the synchronous controller,
+        // so let's also forward pause and resume requests.
+        _resumableController
+          ..onPause = _subscription.pause
+          ..onResume = _subscription.resume;
+        // We don't have to handle `onCancel`: The upload will only cancel the
+        // stream in case of errors, which we already handle by closing the
+        // subscription.
       }
     } else {
       assert(_state == _stateDecidedResumable);
@@ -597,7 +604,7 @@ class _MediaUploadStreamSink implements StreamSink<List<int>> {
     if (_state == _stateProbingLength) {
       // As the data is already cached don't bother to wait on somebody
       // listening on the stream before adding the data.
-      _startNormalUpload(Stream<List<int>>.fromIterable(buffer), _bufferLength);
+      _startNormalUpload(Stream.value(_buffer.takeBytes()), _buffer.length);
     } else {
       _resumableController.close();
     }

pkgs/gcloud/lib/storage.dart

@@ -51,6 +51,7 @@ library gcloud.storage;
 import 'dart:async';
 import 'dart:collection' show UnmodifiableListView, UnmodifiableMapView;
 import 'dart:convert';
+import 'dart:typed_data';
 
 import 'package:_discoveryapis_commons/_discoveryapis_commons.dart' as commons;
 import 'package:googleapis/storage/v1.dart' as storage_api;

pkgs/gcloud/pubspec.yaml

@@ -1,5 +1,5 @@
 name: gcloud
-version: 0.8.5
+version: 0.8.6-dev
 description: >-
   High level idiomatic Dart API for Google Cloud Storage, Pub-Sub and Datastore.
 homepage: https://github.com/dart-lang/gcloud