Work around issue 6984.

This ensures that before an HTTP response is closed, its corresponding
request is drained of input.

Review URL: https://codereview.chromium.org//11348285

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@15476 260f80e4-7a28-3924-810f-c04153c831b5

pkg/http/test/utils.dart

@@ -29,7 +29,7 @@ void startServer() {
       (request, response) {
     response.statusCode = 400;
     response.contentLength = 0;
-    response.outputStream.close();
+    closeResponse(request, response);
   });
 
   _server.addRequestHandler((request) => request.path == '/loop',
@@ -39,7 +39,7 @@ void startServer() {
     response.headers.set('location',
         serverUrl.resolve('/loop?${n + 1}').toString());
     response.contentLength = 0;
-    response.outputStream.close();
+    closeResponse(request, response);
   });
 
   _server.addRequestHandler((request) => request.path == '/redirect',
@@ -47,7 +47,7 @@ void startServer() {
     response.statusCode = 302;
     response.headers.set('location', serverUrl.resolve('/').toString());
     response.contentLength = 0;
-    response.outputStream.close();
+    closeResponse(request, response);
   });
 
   _server.defaultRequestHandler = (request, response) {
@@ -104,6 +104,15 @@ void stopServer() {
   _server = null;
 }
 
+/// Closes [response] while ignoring the body of [request].
+///
+/// Due to issue 6984, it's necessary to drain the request body before closing
+/// the response.
+void closeResponse(HttpRequest request, HttpResponse response) {
+  request.inputStream.onData = request.inputStream.read;
+  request.inputStream.onClosed = response.outputStream.close;
+}
+
 /// A matcher that matches JSON that parses to a value that matches the inner
 /// matcher.
 Matcher parse(matcher) => new _Parse(matcher);

utils/pub/io.dart

@@ -576,6 +576,22 @@ Future _doProcess(Function fn, String executable, List<String> args, workingDir,
   return fn(executable, args, options);
 }
 
+/// Closes [response] while ignoring the body of [request]. Returns a Future
+/// that completes once the response is closed.
+///
+/// Due to issue 6984, it's necessary to drain the request body before closing
+/// the response.
+Future closeHttpResponse(HttpRequest request, HttpResponse response) {
+  var completer = new Completer();
+  request.inputStream.onError = completer.completeException;
+  request.inputStream.onData = request.inputStream.read;
+  request.inputStream.onClosed = () {
+    response.outputStream.close();
+    completer.complete(null);
+  };
+  return completer.future;
+}
+
 /**
  * Wraps [input] to provide a timeout. If [input] completes before
  * [milliseconds] have passed, then the return value completes in the same way.

utils/pub/oauth2.dart

@@ -157,9 +157,13 @@ Future<Client> _authorize() {
       if (queryString == null) queryString = '';
       response.statusCode = 302;
       response.headers.set('location', 'http://pub.dartlang.org/authorized');
-      response.outputStream.close();
+      return Futures.wait([
+        closeHttpResponse(request, response),
+        grant.handleAuthorizationResponse(queryToMap(queryString))
+      ]);
+    }).transform((results) {
       server.close();
-      return grant.handleAuthorizationResponse(queryToMap(queryString));
+      return results[1];
     }), completer);
   });
   server.listen('127.0.0.1', 0);

utils/tests/pub/oauth2_test.dart

@@ -27,7 +27,7 @@ main() {
       expect(request.headers.value('authorization'),
           equals('Bearer access token'));
 
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     pub.kill();
@@ -46,7 +46,7 @@ main() {
       expect(request.headers.value('authorization'),
           equals('Bearer access token'));
 
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     pub.kill();
@@ -83,7 +83,7 @@ main() {
       expect(request.headers.value('authorization'),
           equals('Bearer new access token'));
 
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     pub.shouldExit();
@@ -111,7 +111,7 @@ main() {
       expect(request.headers.value('authorization'),
           equals('Bearer new access token'));
 
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     pub.kill();
@@ -135,7 +135,7 @@ main() {
       expect(request.headers.value('authorization'),
           equals('Bearer new access token'));
 
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     pub.kill();

utils/tests/pub/pub_lish_test.dart

@@ -13,7 +13,7 @@ import '../../pub/io.dart';
 
 void handleUploadForm(ScheduledServer server, [Map body]) {
   server.handle('GET', '/packages/versions/new.json', (request, response) {
-    return server.url.transform((url) {
+    return server.url.chain((url) {
       expect(request.headers.value('authorization'),
           equals('Bearer access token'));
 
@@ -29,7 +29,7 @@ void handleUploadForm(ScheduledServer server, [Map body]) {
 
       response.headers.contentType = new ContentType("application", "json");
       response.outputStream.writeString(JSON.stringify(body));
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
   });
 }
@@ -38,10 +38,10 @@ void handleUpload(ScheduledServer server) {
   server.handle('POST', '/upload', (request, response) {
     // TODO(nweiz): Once a multipart/form-data parser in Dart exists, validate
     // that the request body is correctly formatted. See issue 6952.
-    return server.url.transform((url) {
+    return server.url.chain((url) {
       response.statusCode = 302;
       response.headers.set('location', url.resolve('/create').toString());
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
   });
 }
@@ -60,7 +60,7 @@ main() {
       response.outputStream.writeString(JSON.stringify({
         'success': {'message': 'Package test_pkg 1.0.0 uploaded!'}
       }));
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextLine(), equals('Package test_pkg 1.0.0 uploaded!'));
@@ -85,7 +85,7 @@ main() {
       response.outputStream.writeString(JSON.stringify({
         'error': {'message': 'your request sucked'}
       }));
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextErrLine(), equals('your request sucked'));
@@ -101,7 +101,7 @@ main() {
 
     server.handle('GET', '/packages/versions/new.json', (request, response) {
       response.outputStream.writeString('{not json');
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextErrLine(), equals('Invalid server response:'));
@@ -208,7 +208,7 @@ main() {
       response.headers.contentType = new ContentType('application', 'xml');
       response.outputStream.writeString('<Error><Message>Your request sucked.'
           '</Message></Error>');
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     // TODO(nweiz): This should use the server's error message once the client
@@ -227,7 +227,7 @@ main() {
 
     server.handle('POST', '/upload', (request, response) {
       // don't set the location header
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextErrLine(), equals('Failed to upload the package.'));
@@ -247,7 +247,7 @@ main() {
       response.outputStream.writeString(JSON.stringify({
         'error': {'message': 'Your package was too boring.'}
       }));
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextErrLine(), equals('Your package was too boring.'));
@@ -265,7 +265,7 @@ main() {
 
     server.handle('GET', '/create', (request, response) {
       response.outputStream.writeString('{not json');
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextErrLine(), equals('Invalid server response:'));
@@ -286,7 +286,7 @@ main() {
     server.handle('GET', '/create', (request, response) {
       response.statusCode = 400;
       response.outputStream.writeString(JSON.stringify(body));
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextErrLine(), equals('Invalid server response:'));
@@ -306,7 +306,7 @@ main() {
     var body = {'success': 'Your package was awesome.'};
     server.handle('GET', '/create', (request, response) {
       response.outputStream.writeString(JSON.stringify(body));
-      response.outputStream.close();
+      return closeHttpResponse(request, response);
     });
 
     expectLater(pub.nextErrLine(), equals('Invalid server response:'));

utils/tests/pub/test_pub.dart

@@ -105,7 +105,7 @@ void serve([List<Descriptor> contents]) {
         } catch (e) {
           response.statusCode = 404;
           response.contentLength = 0;
-          response.outputStream.close();
+          closeHttpResponse(request, response);
           return;
         }
 
@@ -114,14 +114,14 @@ void serve([List<Descriptor> contents]) {
           response.statusCode = 200;
           response.contentLength = data.length;
           response.outputStream.write(data);
-          response.outputStream.close();
+          closeHttpResponse(request, response);
         });
 
         future.handleException((e) {
           print("Exception while handling ${request.uri}: $e");
           response.statusCode = 500;
           response.reasonPhrase = e.message;
-          response.outputStream.close();
+          closeHttpResponse(request, response);
         });
       };
       _server.listen("127.0.0.1", 0);