Remove .writeStream, .consume and rewrite IOSink to correctly implement a (sane) well-defined behav…

runtime/bin/socket_patch.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 patch class RawServerSocket  {
   /* patch */ static Future<RawServerSocket> bind([String address = "127.0.0.1",
                                                    int port = 0,
                                                    int backlog = 0]) {
     return _RawServerSocket.bind(address, port, backlog);
   }
@@ skipping 647 matching lines @@
       new _SecureSocket(rawSocket);
 }
 
 
 class _SocketStreamConsumer extends StreamConsumer<List<int>> {
   StreamSubscription subscription;
   final _Socket socket;
   int offset;
   List<int> buffer;
   bool paused = false;
+  Completer streamCompleter;
 
   _SocketStreamConsumer(this.socket);
 
-  Future<Socket> consume(Stream<List<int>> stream) {
+  Future<Socket> addStream(Stream<List<int>> stream) {
+    socket._ensureRawSocketSubscription();
+    streamCompleter = new Completer<Socket>();
     if (socket._raw != null) {
       subscription = stream.listen(
           (data) {
-            assert(!paused);
-            assert(buffer == null);
-            buffer = data;
-            offset = 0;
-            write();
-          },
-          onError: (error) {
-            socket._consumerDone(error);
-          },
-          onDone: () {
-            socket._consumerDone();
-          },
-          unsubscribeOnError: true);
-    }
-    return socket._doneFuture;
-  }
-
-  Future<Socket> addStream(Stream<List<int>> stream) {
-    Completer completer = new Completer<Socket>();
-    if (socket._raw != null) {
-      subscription = stream.listen(
-          (data) {
             assert(!paused);
             assert(buffer == null);
             buffer = data;
             offset = 0;
             write();
           },
           onError: (error) {
-            socket._consumerDone(error);
-            completer.completeError(error.error, error.stackTrace);
+            socket._consumerDone();
+            done(error);
           },
           onDone: () {
-            completer.complete(socket);
+            done();
           },
           unsubscribeOnError: true);
     }
-    return completer.future;
+    return streamCompleter.future;
   }
 
   Future<Socket> close() {
     socket._consumerDone();
-    return completer.future;
+    return new Future.immediate(socket);
   }
 
   void write() {
     try {
       if (subscription == null) return;
       assert(buffer != null);
       // Write as much as possible.
       offset += socket._write(buffer, offset, buffer.length - offset);
       if (offset < buffer.length) {
         if (!paused) {
           paused = true;
           // TODO(ajohnsen): It would be nice to avoid this check.
           // Some info: socket._write can emit an event, if it fails to write.
           // If the user closes the socket in that event, stop() will be called
           // before we get a change to pause.
           if (subscription == null) return;
           subscription.pause();
         }
         socket._enableWriteEvent();
       } else {
         buffer = null;
         if (paused) {
           paused = false;
           subscription.resume();
         }
       }
     } catch (e) {
       stop();
-      socket._consumerDone(e);
+      socket._consumerDone();
+      done(e);
     }
   }
 
+  void done([error]) {
+    if (streamCompleter != null) {
+      var tmp = streamCompleter;
+      streamCompleter = null;
+      if (error != null) {
+        tmp.completeError(error);
+      } else {
+        tmp.complete(socket);
+      }
+    }
+  }
+
   void stop() {
     if (subscription == null) return;
     subscription.cancel();
     subscription = null;
     socket._disableWriteEvent();
   }
 }
 
 
 class _Socket extends Stream<List<int>> implements Socket {
   RawSocket _raw;  // Set to null when the raw socket is closed.
   bool _closed = false;  // Set to true when the raw socket is closed.
   StreamController _controller;
   bool _controllerClosed = false;
   _SocketStreamConsumer _consumer;
-  IOSink<Socket> _sink;
-  Completer _doneCompleter;
+  IOSink _sink;
   var _subscription;
 
   _Socket(RawSocket this._raw) {
     _controller = new StreamController<List<int>>(
         onSubscriptionStateChange: _onSubscriptionStateChange,
         onPauseStateChange: _onPauseStateChange);
     _consumer = new _SocketStreamConsumer(this);
     _sink = new IOSink(_consumer);
 
     // Disable read events until there is a subscription.
@@ skipping 33 matching lines @@
   void write(Object obj) => _sink.write(obj);
 
   void writeln([Object obj = ""]) => _sink.writeln(obj);
 
   void writeCharCode(int charCode) => _sink.writeCharCode(charCode);
 
   void writeAll(Iterable objects, [sep = ""]) => _sink.writeAll(objects, sep);
 
   void add(List<int> bytes) => _sink.add(bytes);
 
-  Future<Socket> consume(Stream<List<int>> stream) {
-    return _sink.consume(stream);
+  Future<Socket> addStream(Stream<List<int>> stream) {
+    return _sink.addStream(stream);
   }
 
-  Future<Socket> writeStream(Stream<List<int>> stream) {
-    return _sink.writeStream(stream);
-  }
-
-  close() => _sink.close();
+  Future<Socket> close() => _sink.close();
 
   Future<Socket> get done => _sink.done;
 
   void destroy() {
     // Destroy can always be called to get rid of a socket.
     if (_raw == null) return;
     _consumer.stop();
     _closeRawSocket();
     _controllerClosed = true;
     _controller.close();
   }
 
   bool setOption(SocketOption option, bool enabled) {
     if (_raw == null) return false;
     return _raw.setOption(option, enabled);
   }
 
   int get port => _raw.port;
   String get remoteHost => _raw.remoteHost;
   int get remotePort => _raw.remotePort;
 
   // Ensure a subscription on the raw socket. Both the stream and the
   // consumer needs a subscription as they share the error and done
   // events from the raw socket.
   void _ensureRawSocketSubscription() {
-    if (_subscription == null) {
+    if (_subscription == null && _raw != null) {
       _subscription = _raw.listen(_onData,
                                   onError: _onError,
                                   onDone: _onDone,
                                   unsubscribeOnError: true);
     }
   }
 
   _closeRawSocket() {
     var tmp = _raw;
     _raw = null;
@@ skipping 36 matching lines @@
         _controller.close();
         break;
     }
   }
 
   void _onDone() {
     if (!_controllerClosed) {
       _controllerClosed = true;
       _controller.close();
     }
-    _done();
+    _consumer.done();
   }
 
   void _onError(error) {
     if (!_controllerClosed) {
       _controllerClosed = true;
       _controller.addError(error);
       _controller.close();
     }
-    _done(error);
-  }
-
-  get _doneFuture {
-    if (_doneCompleter == null) {
-      _ensureRawSocketSubscription();
-      _doneCompleter = new Completer();
-    }
-    return _doneCompleter.future;
-  }
-
-  void _done([error]) {
-    if (_doneCompleter != null) {
-      var tmp = _doneCompleter;
-      _doneCompleter = null;
-      if (error != null) {
-        tmp.completeError(error);
-      } else {
-        tmp.complete(this);
-      }
-    }
+    _consumer.done(error);
   }
 
   int _write(List<int> data, int offset, int length) =>
       _raw.write(data, offset, length);
 
   void _enableWriteEvent() {
     _raw.writeEventsEnabled = true;
   }
 
   void _disableWriteEvent() {
     if (_raw != null) {
       _raw.writeEventsEnabled = false;
     }
   }
 
-  void _consumerDone([error]) {
-    _done(error);
+  void _consumerDone() {
     if (_raw != null) {
       _raw.shutdown(SocketDirection.SEND);
       _disableWriteEvent();
     }
   }
 }
 
 
 class _SecureSocket extends _Socket implements SecureSocket {
   _SecureSocket(RawSecureSocket raw) : super(raw);
 
   void set onBadCertificate(bool callback(X509Certificate certificate)) {
     if (_raw == null) {
       throw new StateError("onBadCertificate called on destroyed SecureSocket");
     }
     _raw.onBadCertificate = callback;
   }
 
   X509Certificate get peerCertificate {
     if (_raw == null) {
      throw new StateError("peerCertificate called on destroyed SecureSocket");
     }
     return _raw.peerCertificate;
   }
 }