Repository: dart-lang/http (dart)

Issue:
StreamedRequest.sink should be a StreamSink
Right now `StreamedRequest.sink` is declared as an `EventSink<List<int>>`:

```dart
  /// The sink to which to write data that will be sent as the request body.
  ///
  /// This may be safely written to before the request is sent; the data will be
  /// buffered.
  ///
  /// Closing this signals the end of the request.
  EventSink<List<int>> get sink => _controller.sink;
```

However, the sink of a `StreamController` is a child class `StreamSink`, which has useful methods like `addStream`: https://api.dart.dev/stable/2.17.5/dart-async/StreamConsumer/addStream.html

Being able to add a stream using this method is essential if you need the stream to be paused when the send buffer is full, e.g. if you are reading from disk faster than you can send it over the network.

Here's a simple example showing how a StreamController is notified when the send buffer is full:

```dart
void main() async {
  final server = await HttpServer.bind('127.0.0.1', 8456);
  server.listen((event) {
    print('server got client');
    final sub = event.listen((event) {});
    // Pause it on the server end, Dart will no longer read from the socket
    // causing new data to be buffered by the OS
    sub.pause();
  });

  final request = StreamedRequest('post', Uri.parse('http://127.0.0.1:8456'));

  final controller = StreamController<List<int>>(
    onPause: () {
      print('paused');
    },
    onResume: () {
      print('resumed');
    },
  );

  Timer.periodic(
    const Duration(seconds: 1),
    (timer) {
      if (controller.isPaused) {
        print('not sending (paused)');
        return;
      }
      print('sending 512K');
      controller.add(Uint8List(512 * 1024));
    },
  );

  // Cast request.sink to a StreamSink so we can call addStream
  (request.sink as StreamSink<List<int>>).addStream(controller.stream);

  await request.send();
}
```

Outputs:

```dart
paused
resumed
sending 512K
paused
server got client
resumed
sending 512K
paused
resumed
sending 512K
paused
resumed
sending 512K
paused
resumed
sending 512K
paused
resumed
sending 512K
paused
not sending - paused
not sending - paused
not sending - paused
not sending - paused
not sending - paused
```

My proposal is just a 1 line change to the getter:

```diff
-  EventSink<List<int>> get sink => _controller.sink;
+  StreamSink<List<int>> get sink => _controller.sink;
```

This would also allow us to get upload progress events without adding any callbacks to post / send:

```dart
final progress = ValueNotifier<int>(0);
request.sink.addStream(
  file.openRead().map((e) {
    progress.value += e.length;
    return e;
  }),
);
```

Pull request description:
Closes #727

This class was not designed for extension, but there is at least one
extending implementation. I don't see any overrides of this field, so no
existing usage should be broken.

Patch:
diff --git a/pkgs/http/CHANGELOG.md b/pkgs/http/CHANGELOG.md
index 29d7761fc6..a23d4c9b22 100644
--- a/pkgs/http/CHANGELOG.md
+++ b/pkgs/http/CHANGELOG.md
@@ -1,7 +1,11 @@
-## 1.0.1
+## 1.1.0-wip
+
+* Add better error messages for `SocketException`s when using `IOClient`.
+* Make `StreamedRequest.sink` a `StreamSink`. This makes `request.sink.close()`
+  return a `Future` instead of `void`, but the returned future should _not_ be
+  awaited. The Future returned from `sink.close()` may only complete after the
+  request has been sent.
 
-* Add better error messages for `SocketException`s when using `IOClient`. 
- 
 ## 1.0.0
 
 * Requires Dart 3.0 or later.
diff --git a/pkgs/http/lib/src/streamed_request.dart b/pkgs/http/lib/src/streamed_request.dart
index 46519567b9..d10386e263 100644
--- a/pkgs/http/lib/src/streamed_request.dart
+++ b/pkgs/http/lib/src/streamed_request.dart
@@ -20,9 +20,12 @@ import 'byte_stream.dart';
 /// ```dart
 /// final request = http.StreamedRequest('POST', Uri.http('example.com', ''))
 ///     ..contentLength = 10
-///     ..sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
-///     ..sink.close();  // The sink must be closed to end the request.
+///     ..sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
 ///
+/// // The sink must be closed to end the request.
+/// // The Future returned from `close()` may not complete until after the
+/// // request is sent, and it should not be awaited.
+/// unawaited(request.sink.close());
 /// final response = await request.send();
 /// ```
 class StreamedRequest extends BaseRequest {
@@ -32,7 +35,7 @@ class StreamedRequest extends BaseRequest {
   /// buffered.
   ///
   /// Closing this signals the end of the request.
-  EventSink<List<int>> get sink => _controller.sink;
+  StreamSink<List<int>> get sink => _controller.sink;
 
   /// The controller for [sink], from which [BaseRequest] will read data for
   /// [finalize].
diff --git a/pkgs/http/pubspec.yaml b/pkgs/http/pubspec.yaml
index 30d8878ee0..ac6f66fda2 100644
--- a/pkgs/http/pubspec.yaml
+++ b/pkgs/http/pubspec.yaml
@@ -1,5 +1,5 @@
 name: http
-version: 1.0.1-wip
+version: 1.1.0-wip
 description: A composable, multi-platform, Future-based API for HTTP requests.
 repository: https://github.com/dart-lang/http/tree/master/pkgs/http
 


Test patch:
diff --git a/pkgs/http/test/html/client_test.dart b/pkgs/http/test/html/client_test.dart
index b56b228518..1a6c634362 100644
--- a/pkgs/http/test/html/client_test.dart
+++ b/pkgs/http/test/html/client_test.dart
@@ -5,6 +5,8 @@
 @TestOn('browser')
 library;
 
+import 'dart:async';
+
 import 'package:http/browser_client.dart';
 import 'package:http/http.dart' as http;
 import 'package:test/test.dart';
@@ -17,9 +19,8 @@ void main() {
     var request = http.StreamedRequest('POST', echoUrl);
 
     var responseFuture = client.send(request);
-    request.sink
-      ..add('{"hello": "world"}'.codeUnits)
-      ..close();
+    request.sink.add('{"hello": "world"}'.codeUnits);
+    unawaited(request.sink.close());
 
     var response = await responseFuture;
     var bytesString = await response.stream.bytesToString();
diff --git a/pkgs/http/test/html/streamed_request_test.dart b/pkgs/http/test/html/streamed_request_test.dart
index c7671734b6..1668656046 100644
--- a/pkgs/http/test/html/streamed_request_test.dart
+++ b/pkgs/http/test/html/streamed_request_test.dart
@@ -5,6 +5,8 @@
 @TestOn('browser')
 library;
 
+import 'dart:async';
+
 import 'package:http/browser_client.dart';
 import 'package:http/http.dart' as http;
 import 'package:test/test.dart';
@@ -16,8 +18,8 @@ void main() {
     test("works when it's set", () async {
       var request = http.StreamedRequest('POST', echoUrl)
         ..contentLength = 10
-        ..sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
-        ..sink.close();
+        ..sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
+      unawaited(request.sink.close());
 
       final response = await BrowserClient().send(request);
 
@@ -28,7 +30,7 @@ void main() {
     test("works when it's not set", () async {
       var request = http.StreamedRequest('POST', echoUrl);
       request.sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
-      request.sink.close();
+      unawaited(request.sink.close());
 
       final response = await BrowserClient().send(request);
       expect(await response.stream.toBytes(),
diff --git a/pkgs/http/test/io/client_test.dart b/pkgs/http/test/io/client_test.dart
index 9476f054ad..fd426a8b3f 100644
--- a/pkgs/http/test/io/client_test.dart
+++ b/pkgs/http/test/io/client_test.dart
@@ -5,6 +5,7 @@
 @TestOn('vm')
 library;
 
+import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
@@ -42,9 +43,8 @@ void main() {
       ..headers[HttpHeaders.userAgentHeader] = 'Dart';
 
     var responseFuture = client.send(request);
-    request
-      ..sink.add('{"hello": "world"}'.codeUnits)
-      ..sink.close();
+    request.sink.add('{"hello": "world"}'.codeUnits);
+    unawaited(request.sink.close());
 
     var response = await responseFuture;
 
@@ -81,9 +81,8 @@ void main() {
       ..headers[HttpHeaders.userAgentHeader] = 'Dart';
 
     var responseFuture = client.send(request);
-    request
-      ..sink.add('{"hello": "world"}'.codeUnits)
-      ..sink.close();
+    request.sink.add('{"hello": "world"}'.codeUnits);
+    unawaited(request.sink.close());
 
     var response = await responseFuture;
 
diff --git a/pkgs/http/test/io/streamed_request_test.dart b/pkgs/http/test/io/streamed_request_test.dart
index 76efd6fe56..f0e990c767 100644
--- a/pkgs/http/test/io/streamed_request_test.dart
+++ b/pkgs/http/test/io/streamed_request_test.dart
@@ -5,6 +5,7 @@
 @TestOn('vm')
 library;
 
+import 'dart:async';
 import 'dart:convert';
 
 import 'package:http/http.dart' as http;
@@ -22,9 +23,8 @@ void main() {
     test('controls the Content-Length header', () async {
       var request = http.StreamedRequest('POST', serverUrl)
         ..contentLength = 10
-        ..sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
-        ..sink.close();
-
+        ..sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
+      unawaited(request.sink.close());
       var response = await request.send();
       expect(
           await utf8.decodeStream(response.stream),
@@ -35,8 +35,7 @@ void main() {
     test('defaults to sending no Content-Length', () async {
       var request = http.StreamedRequest('POST', serverUrl);
       request.sink.add([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
-      request.sink.close();
-
+      unawaited(request.sink.close());
       var response = await request.send();
       expect(await utf8.decodeStream(response.stream),
           parse(containsPair('headers', isNot(contains('content-length')))));
@@ -47,7 +46,7 @@ void main() {
   test('.send() with a response with no content length', () async {
     var request =
         http.StreamedRequest('GET', serverUrl.resolve('/no-content-length'));
-    request.sink.close();
+    unawaited(request.sink.close());
     var response = await request.send();
     expect(await utf8.decodeStream(response.stream), equals('body'));
   });