Improve WebSocket interface by making it implement StreamSink.

sdk/lib/io/io_sink.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.
 
 part of dart.io;
 
 /**
  * Helper class to wrap a [StreamConsumer<List<int>>] and provide
  * utility functions for writing to the StreamConsumer directly. The
  * [IOSink] buffers the input given by all [StringSink] methods and will delay
@@ skipping 34 matching lines @@
    */
   Future close();
 
   /**
    * Get a future that will complete when all synchronous have completed, or an
    * error happened. This future is identical to the future returned from close.
    */
   Future get done;
 }
 
-
-class _IOSinkImpl implements IOSink {
-  final StreamConsumer<List<int>> _target;
+class _StreamSinkImpl<T> implements StreamSink<T> {

[Søren Gjesse, 2013/04/16 12:41:31]

Should this class be public in dart:async to enable StreamSink implementations
for users?

[Anders Johnsen, 2013/04/16 13:46:22]

Maybe. I'll consider moving it later on, after talking with Florian.

+  final StreamConsumer<T> _target;
   Completer _doneCompleter = new Completer();
   Future _doneFuture;
-  StreamController<List<int>> _controllerInstance;
+  StreamController<T> _controllerInstance;
   Completer _controllerCompleter;
-  Encoding _encoding;
   bool _isClosed = false;
   bool _isBound = false;
-  bool _encodingMutable = true;
 
-  _IOSinkImpl(StreamConsumer<List<int>> this._target, this._encoding) {
+  _StreamSinkImpl(StreamConsumer<T> this._target) {
     _doneFuture = _doneCompleter.future;
   }
 
-  Encoding get encoding => _encoding;
-
-  void set encoding(Encoding value) {
-    if (!_encodingMutable) {
-      throw new StateError("IOSink encoding is not mutable");
-    }
-    _encoding = value;
-  }
-
-  void write(Object obj) {
-    // This comment is copied from runtime/lib/string_buffer_patch.dart.
-    // TODO(srdjan): The following four lines could be replaced by
-    // '$obj', but apparently this is too slow on the Dart VM.
-    String string;
-    if (obj is String) {
-      string = obj;
-    } else {
-      string = obj.toString();
-      if (string is! String) {
-        throw new ArgumentError('toString() did not return a string');
-      }
-    }
-    if (string.isEmpty) return;
-    add(_encodeString(string, _encoding));
-  }
-
-  void writeAll(Iterable objects, [String separator = ""]) {
-    Iterator iterator = objects.iterator;
-    if (!iterator.moveNext()) return;
-    if (separator.isEmpty) {
-      do {
-        write(iterator.current);
-      } while (iterator.moveNext());
-    } else {
-      write(iterator.current);
-      while (iterator.moveNext()) {
-        write(separator);
-        write(iterator.current);
-      }
-    }
-  }
-
-  void writeln([Object obj = ""]) {
-    write(obj);
-    write("\n");
-  }
-
-  void writeCharCode(int charCode) {
-    write(new String.fromCharCode(charCode));
-  }
-
-  void add(List<int> data) {
+  void add(T data) {
     _controller.add(data);
   }
 
   void addError(error) {
     _controller.addError(error);
   }
 
-  Future addStream(Stream<List<int>> stream) {
+  Future addStream(Stream<T> stream) {
     if (_isBound) {
-      throw new StateError("IOSink is already bound to a stream");
+      throw new StateError("StreamSink is already bound to a stream");
     }
     _isBound = true;
     // Wait for any sync operations to complete.
     Future targetAddStream() {
       return _target.addStream(stream)
           .whenComplete(() {
             _isBound = false;
           });
     }
     if (_controllerInstance == null) return targetAddStream();
     var future = _controllerCompleter.future;
     _controllerInstance.close();
     return future.then((_) => targetAddStream());
   }
 
   Future close() {
     if (_isBound) {
-      throw new StateError("IOSink is bound to a stream");
+      throw new StateError("StreamSink is bound to a stream");
     }
     if (!_isClosed) {
       _isClosed = true;
       if (_controllerInstance != null) {
         _controllerInstance.close();
       } else {
         _closeTarget();
       }
     }
     return done;
@@ skipping 11 matching lines @@
     if (_doneCompleter == null) return;
     var tmp = _doneCompleter;
     _doneCompleter = null;
     if (error == null) {
       tmp.complete(value);
     } else {
       tmp.completeError(error);
     }
   }
 
-  StreamController<List<int>> get _controller {
+  StreamController<T> get _controller {
     if (_isBound) {
-      throw new StateError("IOSink is bound to a stream");
+      throw new StateError("StreamSink is bound to a stream");
     }
     if (_isClosed) {
-      throw new StateError("IOSink is closed");
+      throw new StateError("StreamSink is closed");
     }
     if (_controllerInstance == null) {
-      _controllerInstance = new StreamController<List<int>>();
+      _controllerInstance = new StreamController<T>();
       _controllerCompleter = new Completer();
       _target.addStream(_controller.stream)
           .then(
               (_) {
                 if (_isBound) {
                   // A new stream takes over - forward values to that stream.
                   var completer = _controllerCompleter;
                   _controllerCompleter = null;
                   _controllerInstance = null;
                   completer.complete();
@@ skipping 12 matching lines @@
                 } else {
                   // No new stream. No need to close target, as it have already
                   // failed.
                   _completeDone(error: error);
                 }
               });
     }
     return _controllerInstance;
   }
 }
+
+
+class _IOSinkImpl extends _StreamSinkImpl<List<int>> implements IOSink {
+  Encoding _encoding;
+  bool _encodingMutable = true;
+
+  _IOSinkImpl(StreamConsumer<List<int>> target, this._encoding)
+      : super(target);
+
+  Encoding get encoding => _encoding;
+
+  void set encoding(Encoding value) {
+    if (!_encodingMutable) {
+      throw new StateError("IOSink encoding is not mutable");
+    }
+    _encoding = value;
+  }
+
+  void write(Object obj) {
+    // This comment is copied from runtime/lib/string_buffer_patch.dart.
+    // TODO(srdjan): The following four lines could be replaced by
+    // '$obj', but apparently this is too slow on the Dart VM.
+    String string;
+    if (obj is String) {
+      string = obj;
+    } else {
+      string = obj.toString();
+      if (string is! String) {
+        throw new ArgumentError('toString() did not return a string');
+      }
+    }
+    if (string.isEmpty) return;
+    add(_encodeString(string, _encoding));
+  }
+
+  void writeAll(Iterable objects, [String separator = ""]) {
+    Iterator iterator = objects.iterator;
+    if (!iterator.moveNext()) return;
+    if (separator.isEmpty) {
+      do {
+        write(iterator.current);
+      } while (iterator.moveNext());
+    } else {
+      write(iterator.current);
+      while (iterator.moveNext()) {
+        write(separator);
+        write(iterator.current);
+      }
+    }
+  }
+
+  void writeln([Object obj = ""]) {
+    write(obj);
+    write("\n");
+  }
+
+  void writeCharCode(int charCode) {
+    write(new String.fromCharCode(charCode));
+  }
+}