Fix subscription handling in stream decoder

sdk/lib/utf/utf_stream.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.utf;
 
 class _HelperStreamController<T> extends StreamController<T> {
   final Function onPauseChanged;
 
   _HelperStreamController(this.onPauseChanged);
 
   void onPauseStateChange() {
     onPauseChanged();
   }
 }
 
-abstract class _StringDecoder implements StreamTransformer<List<int>, String> {
-  _HelperStreamController<String> _controller;
-  StreamSubscription<List<int>> _subscription;
+abstract class _StringDecoder
+    extends StreamEventTransformer<List<int>, String> {
   List<int> _carry;
   List<int> _buffer;
   int _replacementChar;
-  bool _paused = false;
 
-  _StringDecoder(int this._replacementChar) {
-    _controller = new _HelperStreamController<String>(_onPauseChanged);
-  }
+  _StringDecoder(int this._replacementChar);
 
-  void _onPauseChanged() {
-    _paused = _controller.isPaused;
-    if (_subscription == null) return;
-    if (_paused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
-    }
-  }
-
-  Stream<String> bind(Stream<List<int>> stream) {
-    _subscription = stream.listen(
-        _onData,
-        onError: _controller.signalError,
-        onDone: () {
-          if (_carry != null) {
-            _controller.add(new String.fromCharCodes(
-                new List.fixedLength(_carry.length, fill: _replacementChar)));
-          }
-          _controller.close();
-        },
-        unsubscribeOnError: false);
-    if (_paused) _subscription.pause();
-    return _controller.stream;
-  }
-
-  void _onData(List<int> bytes) {
+  void handleData(List<int> bytes, StreamSink<String> sink) {
     _buffer = <int>[];
     List<int> carry = _carry;
     _carry = null;
     int pos = 0;
     int available = bytes.length;
     // If we have carry-over data, start from negative index, indicating carry
     // index.
     int goodChars = 0;
     if (carry != null) pos = -carry.length;
     while (pos < available) {
@@ skipping 21 matching lines @@
         break;
       } else {
         // Invalid byte at position pos - 1
         _buffer.length = goodChars;
         _addChar(-1);
         goodChars = _buffer.length;
       }
     }
     if (_buffer.length > 0) {
       // Limit to 'goodChars', if lower than actual charCodes in the buffer.
-      _controller.add(new String.fromCharCodes(_buffer));
+      sink.add(new String.fromCharCodes(_buffer));
     }
     _buffer = null;
   }
 
+  void handleDone(StreamSink<String> sink) {
+    if (_carry != null) {
+      sink.add(new String.fromCharCodes(
+          new List.fixedLength(_carry.length, fill: _replacementChar)));
+    }
+    sink.close();
+  }
+
   int _processBytes(int getNext());
 
   void _addChar(int char) {
     if (char > 0x10FFFF || char < 0) char = _replacementChar;
     _buffer.add(char);
   }
 }
 
 /**
  * StringTransformer that decodes a stream of UTF-8 encoded bytes.
@@ skipping 42 matching lines @@
       if (value < min) value = -1;
       _addChar(value);
       return 1 + additionalBytes;
     }
     _addChar(value);
     return 1;
   }
 }
 
 
-abstract class _StringEncoder implements StreamTransformer<String, List<int>> {
-  _HelperStreamController<List<int>> _controller;
-  StreamSubscription<String> _subscription;
+abstract class _StringEncoder
+    extends StreamEventTransformer<String, List<int>> {
 
-  void _onPauseChanged() {
-    if (_controller.isPaused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
-    }
-  }
-  Stream<List<int>> bind(Stream<String> stream) {
-    _controller = new _HelperStreamController(_onPauseChanged);
-    _subscription = stream.listen(
-        (string) => _controller.add(_processString(string)),
-        onError: _controller.signalError,
-        onDone: _controller.close,
-        unsubscribeOnError: false);
-    return _controller.stream;
+  void handleData(String data, StreamSink<List<int>> sink) {
+    sink.add(_processString(data));
   }
 
   List<int> _processString(String string);
 }
 
 /**
  * StringTransformer that UTF-8 encodes a stream of strings.
  */
 class Utf8EncoderTransformer extends _StringEncoder {
   List<int> _processString(String string) {
@@ skipping 22 matching lines @@
       }
       for (int i = additionalBytes; i > 0; i--) {
         // 10xxxxxx (xxxxxx is next 6 bits from the top).
         bytes.add(((charCode >> (6 * (i - 1))) & 0x3F) | 0x80);
       }
       pos += additionalBytes + 1;
     }
     return bytes;
   }
 }