Support Request hijacking in Shelf, using a similar API to Rack.

pkg/shelf/lib/src/request.dart

Index: pkg/shelf/lib/src/request.dart
diff --git a/pkg/shelf/lib/src/request.dart b/pkg/shelf/lib/src/request.dart
index f7e011dc42609321c775d6d8ecc9bc7af582e78a..a72c56c8d94e80a4c7223f609a6af7551e11af64 100644
--- a/pkg/shelf/lib/src/request.dart
+++ b/pkg/shelf/lib/src/request.dart
@@ -8,7 +8,17 @@ import 'dart:async';
 
 import 'package:http_parser/http_parser.dart';
 
+import 'hijack_exception.dart';
 import 'message.dart';
+import 'util.dart';
+
+/// A callback provided by a Shelf handler that's passed to [Request.hijack].
+typedef void HijackCallback(
+    Stream<List<int>> stream, StreamSink<List<int>> sink);
+
+/// A callback provided by a Shelf adapter that's used by [Request.hijack] to
+/// provide a [HijackCallback] with a socket.
+typedef void OnHijackCallback(HijackCallback callback);
 
 /// Represents an HTTP request to be processed by a Shelf application.
 class Request extends Message {
@@ -45,6 +55,17 @@ class Request extends Message {
   /// The original [Uri] for the request.
   final Uri requestedUri;
 
+  /// The callback wrapper for hijacking this request.
+  ///
+  /// This will be `null` if this request can't be hijacked.
+  final _OnHijack _onHijack;
+
+  /// Whether this request can be hijacked.
+  ///
+  /// This will be `false` either if the adapter doesn't support hijacking, or
+  /// if the request has already been hijacked.
+  bool get canHijack => _onHijack != null && !_onHijack.called;
+
   /// If this is non-`null` and the requested resource hasn't been modified
   /// since this date and time, the server should return a 304 Not Modified
   /// response.
@@ -68,15 +89,43 @@ class Request extends Message {
   /// [ArgumentError].
   ///
   /// The default value for [protocolVersion] is '1.1'.
+  ///
+  /// ## `onHijack`
+  ///
+  /// [onHijack] allows handlers to take control of the underlying socket for
+  /// the request. It should be passed by adapters that can provide access to
+  /// the bidirectional socket underlying the HTTP connection stream.
+  ///
+  /// The [onHijack] callback will only be called once per request. It will be
+  /// passed another callback which takes a byte stream and a byte sink.
+  /// [onHijack] must pass the stream and sink for the connection stream to this
+  /// callback, although it may do so asynchronously. Both parameters may be the
+  /// same object. If the user closes the sink, the adapter should ensure that
+  /// the stream is closed as well.
+  ///
+  /// If a request is hijacked, the adapter should expect to receive a
+  /// [HijackException] from the handler. This is a special exception used to
+  /// indicate that hijacking has occurred; the adapter should avoid either

[kevmoo, 2014/05/07 19:28:22]

occurred. The adapter...

[nweiz, 2014/05/19 20:10:29]

Done.

+  /// sending a response or notifying the user of an error if one of these is

[kevmoo, 2014/05/07 19:28:22]

"one of these is received" -> a [HijackException] is caught.

Just to be super clear.

[nweiz, 2014/05/19 20:10:29]

Done.

+  /// received.
+  ///
+  /// An adapter can check whether a request was hijacked using [canHijack],
+  /// which will be `false` for a hijacked request. The adapter may throw an
+  /// error if a [HijackException] is received for a non-hijacked request, or if
+  /// no [HijackException] is received for a hijacked request.
+  ///
+  /// See also [hijack].
   // TODO(kevmoo) finish documenting the rest of the arguments.
   Request(this.method, Uri requestedUri, {String protocolVersion,
     Map<String, String> headers, Uri url, String scriptName,
-    Stream<List<int>> body, Map<String, Object> context})
+    Stream<List<int>> body, Map<String, Object> context,
+    OnHijackCallback onHijack})
       : this.requestedUri = requestedUri,
         this.protocolVersion = protocolVersion == null ?
             '1.1' : protocolVersion,
         this.url = _computeUrl(requestedUri, url, scriptName),
         this.scriptName = _computeScriptName(requestedUri, url, scriptName),
+        this._onHijack = onHijack == null ? null : new _OnHijack(onHijack),
         super(body == null ? new Stream.fromIterable([]) : body,
             headers: headers, context: context) {
     if (method.isEmpty) throw new ArgumentError('method cannot be empty.');
@@ -107,6 +156,52 @@ class Request extends Message {
       throw new ArgumentError('scriptName and url cannot both be empty.');
     }
   }
+
+  /// Takes control of the underlying request socket.
+  ///
+  /// Synchronously, this throws a [HijackException] that indicates to the
+  /// adapter that it shouldn't emit a response itself. Asynchronously,
+  /// [callback] is called with a [Stream<List<int>>] and
+  /// [StreamSink<List<int>>], respectively, that provide access to the
+  /// underlying request socket.
+  ///
+  /// If the sink is closed, the stream will be closed as well. The stream and
+  /// sink may be the same object, as in the case of a `dart:io` `Socket`
+  /// object.
+  ///
+  /// This may only be called when using a Shelf adapter that supports
+  /// hijacking, such as the `dart:io` adapter. In addition, a given request may
+  /// only be hijacked once. [canHijack] can be used to detect whether this
+  /// request can be hijacked.
+  void hijack(HijackCallback callback) {
+    if (_onHijack == null) {
+      throw new StateError("This request can't be hijacked.");
+    }
+
+    _onHijack.run(callback);
+    throw const HijackException();
+  }
+}
+
+/// A class containing a callback for [Request.hijack] that also tracks whether
+/// the callback has been called.
+class _OnHijack {
+  /// The callback.
+  final OnHijackCallback _callback;
+
+  /// Whether [this] has been called.
+  var called = false;

[kevmoo, 2014/05/07 19:28:22]

var -> bool

[nweiz, 2014/05/19 20:10:29]

Done, under protest.

[kevmoo, 2014/05/19 20:14:18]

:-)

+
+  _OnHijack(this._callback);
+
+  /// Calls [this].
+  ///
+  /// Throws a [StateError] if [this] has already been called.
+  void run(HijackCallback callback) {
+    if (called) throw new StateError("This request has already been hijacked.");
+    called = true;
+    newFuture(() => _callback(callback));
+  }
 }
 
 /// Computes `url` from the provided [Request] constructor arguments.