Add a 30s timeout for all HTTP requests in Pub.

utils/pub/io.dart

Index: utils/pub/io.dart
diff --git a/utils/pub/io.dart b/utils/pub/io.dart
index c622b9505ddc0b81ca2d1a5ff8d0526f0b02d519..005916d8b5cbd4f145ca2b897e6a3784cd66d31a 100644
--- a/utils/pub/io.dart
+++ b/utils/pub/io.dart
@@ -8,6 +8,7 @@
 #library('io');
 
 #import('dart:io');
+#import('dart:isolate');
 #import('dart:uri');
 
 /** Gets the current working directory. */
@@ -285,41 +286,27 @@ Future<File> createSymlink(from, to) {
 // TODO(rnystrom): Should this be async?
 String getFullPath(entry) => new File(_getPath(entry)).fullPathSync();
 
+// TODO(nweiz): make this configurable
 /**
- * Opens an input stream for a HTTP GET request to [uri], which may be a
- * [String] or [Uri].
+ * The amount of time in milliseconds to allow HTTP requests before assuming
+ * they've failed.
  */
-InputStream httpGet(uri) {
-  var resultStream = new ListInputStream();
-  var client = new HttpClient();
-  var connection = client.getUrl(_getUri(uri));
-
-  // TODO(nweiz): propagate this error to the return value. See issue 3657.
-  connection.onError = (e) { throw e; };
-  connection.onResponse = (response) {
-    if (response.statusCode >= 400) {
-      // TODO(nweiz): propagate this error to the return value. See issue 3657.
-      throw new Exception(
-          "HTTP request for $uri failed with status ${response.statusCode}");
-    }
-
-    pipeInputToInput(response.inputStream, resultStream, client.shutdown);
-  };
-
-  return resultStream;
-}
+final HTTP_TIMEOUT = 30 * 1000;
 
 /**
  * Opens an input stream for a HTTP GET request to [uri], which may be a
- * [String] or [Uri]. Completes with the result of the request as a String.
+ * [String] or [Uri].
+ *
+ * Callers should be sure to use [timeout] to make sure that the HTTP request
+ * doesn't last indefinitely
  */
-Future<String> httpGetString(uri) {
-  // TODO(rnystrom): This code is very similar to httpGet() and
-  // consumeInputStream() (but this one propagates errors). Merge them when
-  // #3657 is fixed.
+Future<InputStream> httpGet(uri) {
+  // TODO(nweiz): This could return an InputStream synchronously if issue 3657
+  // were fixed and errors could be propagated through it. Then we could also
+  // automatically attach a timeout to that stream.
   uri = _getUri(uri);
 
-  var completer = new Completer<String>();
+  var completer = new Completer<InputStream>();
   var client = new HttpClient();
   var connection = client.getUrl(uri);
 
@@ -344,28 +331,23 @@ Future<String> httpGetString(uri) {
       return;
     }
 
-    var resultStream = new ListInputStream();
-    var buffer = <int>[];
-
-    response.inputStream.onClosed = () {
-      client.shutdown();
-      completer.complete(new String.fromCharCodes(buffer));
-    };
-
-    response.inputStream.onData = () {
-      buffer.addAll(response.inputStream.read());
-    };
-
-    response.inputStream.onError = (e) {
-      client.shutdown();
-      completer.completeException(e);
-    };
+    completer.complete(response.inputStream);
   };
 
   return completer.future;
 }
 
 /**
+ * Opens an input stream for a HTTP GET request to [uri], which may be a
+ * [String] or [Uri]. Completes with the result of the request as a String.
+ */
+Future<String> httpGetString(uri) {
+  var future = httpGet(uri).chain((stream) => consumeInputStream(stream))
+    .transform((bytes) => new String.fromCharCodes(bytes));

[Bob Nystrom, 2012/09/06 22:52:05]

Indent +2

[nweiz, 2012/09/06 23:33:34]

Done.

+  return timeout(future, HTTP_TIMEOUT, 'Timed out while fetching URL "$uri".');
+}
+
+/**
  * Takes all input from [source] and writes it to [sink].
  *
  * [onClosed] is called when [source] is closed.
@@ -460,6 +442,32 @@ Future<PubProcessResult> runProcess(String executable, List<String> args,
 }
 
 /**
+ * Wraps [input] to provide a timeout. If [input] completes before
+ * [milliSeconds] have passed, then the return value completes in the same way.
+ * However, if [milliSeconds] pass before [input] has completed, it completes
+ * with [exception].
+ */
+Future timeout(Future input, int milliSeconds, exception) {

[Bob Nystrom, 2012/09/06 22:52:05]

Can you create a TimeoutException class and then take a message string here? I'd
like to do something a little more structured that throwing a raw string, and
that lets us do better error-reporting/handling later at a higher level.

[nweiz, 2012/09/06 23:33:34]

Done.

+  var completer = new Completer();
+  var timer = new Timer(milliSeconds, (_) {
+    if (completer.future.isComplete) return;
+    completer.completeException(exception);

[Bob Nystrom, 2012/09/06 22:52:05]

I like the idea of timeout() being a separate orthogonal function, but the
problem here is that it doesn't actually do anything to try to cancel the
request that timed out.

It probably more or less works here because we explicitly exit() now, but if we
didn't, this would leave the app still running until dart:io's internal timeout
triggered. Can we do something better here?

[nweiz, 2012/09/06 23:33:34]

Given that we do explicitly exit, I don't see anything wrong with letting the
original future dangle. I could change this to take a callback that could cancel
and then throw, but it feels like that just adds complexity without any
practical benefit.

[Bob Nystrom, 2012/09/07 00:15:34]

OK. Can you update the doc comment to note that this doesn't cancel the async
operation when a timeout occurs?

+  });
+  input.handleException((e) {
+    if (completer.future.isComplete) return false;
+    timer.cancel();
+    completer.completeException(e);
+    return true;
+  });
+  input.then((value) {
+    if (completer.future.isComplete) return;
+    timer.cancel();
+    completer.complete(value);
+  });
+  return completer.future;
+}
+
+/**
  * Tests whether or not the git command-line app is available for use.
  */
 Future<bool> get isGitInstalled {