Fix string_transformer.

sdk/lib/io/string_transformer.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;
 
 /**
  * String encodings.
  */
 class Encoding {
@@ skipping 135 matching lines @@
   controller.stream
     .transform(new StringEncoder(encoding))
     .listen((data) => bytes = data);
   controller.add(string);
   controller.close();
   assert(bytes != null);
   return bytes;
 }
 
 
-class LineTransformer implements StreamTransformer<String, String> {
+class LineTransformer extends StreamEventTransformer<String, String> {
   const int _LF = 10;
   const int _CR = 13;
 
   final StringBuffer _buffer = new StringBuffer();
 
   StreamSubscription<String> _subscription;
-  StreamController<String> _controller;
   String _carry;
 
-  Stream<String> bind(Stream<String> stream) {
-    _controller = new StreamController<String>(
-        onPauseStateChange: _pauseChanged,
-        onSubscriptionStateChange: _subscriptionChanged);
-
-    void handle(String data, bool isClosing) {
-      if (_carry != null) {
-        data = _carry.concat(data);
-        _carry = null;
-      }
-      int startPos = 0;
-      int pos = 0;
-      while (pos < data.length) {
-        int skip = 0;
-        int char = data.codeUnitAt(pos);
-        if (char == _LF) {
-          skip = 1;
-        } else if (char == _CR) {
-          skip = 1;
-          if (pos + 1 < data.length) {
-            if (data.codeUnitAt(pos + 1) == _LF) {
-              skip = 2;
-            }
-          } else if (!isClosing) {
-            _carry = data.substring(startPos);
-            return;
+  void _handle(String data, StreamSink<String> sink, bool isClosing) {
+    if (_carry != null) {
+      data = _carry.concat(data);
+      _carry = null;
+    }
+    int startPos = 0;
+    int pos = 0;
+    while (pos < data.length) {
+      int skip = 0;
+      int char = data.codeUnitAt(pos);
+      if (char == _LF) {
+        skip = 1;
+      } else if (char == _CR) {
+        skip = 1;
+        if (pos + 1 < data.length) {
+          if (data.codeUnitAt(pos + 1) == _LF) {
+            skip = 2;
           }
-        }
-        if (skip > 0) {
-          _buffer.add(data.substring(startPos, pos));
-          _controller.add(_buffer.toString());
-          _buffer.clear();
-          startPos = pos = pos + skip;
-        } else {
-          pos++;
+        } else if (!isClosing) {
+          _carry = data.substring(startPos);
+          return;
         }
       }
-      if (pos != startPos) {
-        // Add remaining
+      if (skip > 0) {
         _buffer.add(data.substring(startPos, pos));
-      }
-      if (isClosing && !_buffer.isEmpty) {
-        _controller.add(_buffer.toString());
+        sink.add(_buffer.toString());
         _buffer.clear();
+        startPos = pos = pos + skip;
+      } else {
+        pos++;
       }
     }
-
-    _subscription = stream.listen(
-        (data) => handle(data, false),
-        onDone: () {
-          // Handle remaining data (mainly _carry).
-          handle("", true);
-          _controller.close();
-        },
-        onError: _controller.signalError);
-    return _controller.stream;
-  }
-
-  void _pauseChanged() {
-    if (_controller.isPaused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
+    if (pos != startPos) {
+      // Add remaining
+      _buffer.add(data.substring(startPos, pos));
+    }
+    if (isClosing && !_buffer.isEmpty) {
+      sink.add(_buffer.toString());
+      _buffer.clear();
     }
   }
 
-  void _subscriptionChanged() {
-    if (!_controller.hasSubscribers) {
-      _subscription.cancel();
-    }
+  handleData(String data, StreamSink<String> sink) {
+    _handle(data, sink, false);
+  }
+
+  handleDone(StreamSink<String> sink) {
+    _handle("", sink, true);
+    sink.close();
   }
 }
 
 
-class _SingleByteDecoder implements StreamTransformer<List<int>, String> {
-  StreamSubscription<List<int>> _subscription;
-  StreamController<String> _controller;
+abstract class _SingleByteDecoder
+    implements StreamTransformer<List<int>, String> {
+
   final int _replacementChar;
 
   _SingleByteDecoder(this._replacementChar);
 
   Stream<String> bind(Stream<List<int>> stream) {
-    _controller = new StreamController<String>(
-        onPauseStateChange: _pauseChanged,
-        onSubscriptionStateChange: _subscriptionChanged);
-    _subscription = stream.listen(
-        (data) {
-          var buffer = new List<int>.fixedLength(data.length);
-          for (int i = 0; i < data.length; i++) {
-            int char = _decodeByte(data[i]);
-            if (char < 0) char = _replacementChar;
-            buffer[i] = char;
-          }
-          _controller.add(new String.fromCharCodes(buffer));
-        },
-        onDone: _controller.close,
-        onError: _controller.signalError);
-    return _controller.stream;
+    return stream.map((List<int> data) {
+      var buffer = new List<int>.fixedLength(data.length);
+      for (int i = 0; i < data.length; i++) {
+        int char = _decodeByte(data[i]);
+        if (char < 0) char = _replacementChar;
+        buffer[i] = char;
+      }
+      return new String.fromCharCodes(buffer);
+    });
   }
 
   int _decodeByte(int byte);
-
-  void _pauseChanged() {
-    if (_controller.isPaused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
-    }
-  }
-
-  void _subscriptionChanged() {
-    if (!_controller.hasSubscribers) {
-      _subscription.cancel();
-    }
-  }
 }
 
 
 // Utility class for decoding ascii data delivered as a stream of
 // bytes.
 class _AsciiDecoder extends _SingleByteDecoder {
   _AsciiDecoder(int replacementChar) : super(replacementChar);
 
   int _decodeByte(int byte) => ((byte & 0x7f) == byte) ? byte : -1;
 }
 
 
 // Utility class for decoding Latin-1 data delivered as a stream of
 // bytes.
 class _Latin1Decoder extends _SingleByteDecoder {
   _Latin1Decoder(int replacementChar) : super(replacementChar);
 
   int _decodeByte(int byte) => ((byte & 0xFF) == byte) ? byte : -1;
 }
 
 
-class _SingleByteEncoder implements StreamTransformer<String, List<int>> {
-  StreamSubscription<String> _subscription;
-  StreamController<List<int>> _controller;
-
+abstract class _SingleByteEncoder implements StreamTransformer<String, List<int>> {
   Stream<List<int>> bind(Stream<String> stream) {
-    _controller = new StreamController<List<int>>(
-        onPauseStateChange: _pauseChanged,
-        onSubscriptionStateChange: _subscriptionChanged);
-    _subscription = stream.listen(
-        (string) {
-          var bytes = _encode(string);
-          if (bytes == null) {
-            _controller.signalError(new FormatException(
-                "Invalid character for encoding"));
-            _controller.close();
-            _subscription.cancel();
-          } else {
-            _controller.add(bytes);
-          }
-        },
-        onDone: _controller.close,
-        onError: _controller.signalError);
-    return _controller.stream;
+    return stream.map((String string) {
+      var bytes = _encode(string);
+      if (bytes == null) {
+        throw new FormatException("Invalid character for encoding");
+      }
+      return bytes;
+    });
   }
 
   List<int> _encode(String string);
-
-  void _pauseChanged() {
-    if (_controller.isPaused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
-    }
-  }
-
-  void _subscriptionChanged() {
-    if (!_controller.hasSubscribers) {
-      _subscription.cancel();
-    }
-  }
 }
 
 
 // Utility class for encoding a string into an ASCII byte stream.
 class _AsciiEncoder extends _SingleByteEncoder {
   List<int> _encode(String string) {
     var bytes = string.codeUnits;
     for (var byte in bytes) {
       if (byte > 127) return null;
     }
@@ skipping 21 matching lines @@
 class _WindowsCodePageEncoder extends _SingleByteEncoder {
   List<int> _encode(String string) => _encodeString(string);
 
   external static List<int> _encodeString(String string);
 }
 
 
 // Utility class for decoding Windows current code page data delivered
 // as a stream of bytes.
 class _WindowsCodePageDecoder implements StreamTransformer<List<int>, String> {
-  StreamSubscription<List<int>> _subscription;
-  StreamController<String> _controller;
-
   Stream<String> bind(Stream<List<int>> stream) {
-    _controller = new StreamController<String>(
-        onPauseStateChange: _pauseChanged,
-        onSubscriptionStateChange: _subscriptionChanged);
-    _subscription = stream.listen(
-        (data) {
-          _controller.add(_decodeBytes(data));
-        },
-        onDone: _controller.close,
-        onError: _controller.signalError);
-    return _controller.stream;
+    return stream.map(_decodeBytes);
   }
 
   external static String _decodeBytes(List<int> bytes);
-
-  void _pauseChanged() {
-    if (_controller.isPaused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
-    }
-  }
-
-  void _subscriptionChanged() {
-    if (!_controller.hasSubscribers) {
-      _subscription.cancel();
-    }
-  }
 }