Fix subscription handling in stream decoder

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) {

[floitsch, 2013/02/26 14:47:26]

make it private.

[Søren Gjesse, 2013/02/26 15:23:33]

Done.

+    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();
-    }
+  void handleData(String data, StreamSink<String> sink) {
+    handle(data, sink, false);
+  }
+
+  void handleDone(StreamSink<String> sink) {
+    handle("", sink, true);

[floitsch, 2013/02/26 14:47:26]

sink.close()

[Søren Gjesse, 2013/02/26 15:23:33]

Thanks.

   }
 }
 
 
-class _SingleByteDecoder implements StreamTransformer<List<int>, String> {
-  StreamSubscription<List<int>> _subscription;
-  StreamController<String> _controller;
+class _SingleByteDecoder extends StreamEventTransformer<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;
+  void handleData(List<int> data, StreamSink<String> sink) {
+    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;
+    }
+    sink.add(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;
-
-  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;
+class _SingleByteEncoder extends StreamEventTransformer<String, List<int>> {
+  void handleData(String data, StreamSink<List<int>> sink) {
+    var bytes = _encode(data);
+    if (bytes == null) {
+      throw new FormatException("Invalid character for encoding");
+      sink.close();
+    } else {
+      sink.add(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 20 matching lines @@
 // single byte encoder, to avoid copying boilerplate.
 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;
+class _WindowsCodePageDecoder extends StreamEventTransformer<List<int>, String> {
+  void handleData(List<int> data, StreamSink<String> sink) {
+    sink.add(_decodeBytes(data));
   }
 
   external static String _decodeBytes(List<int> bytes);
-
-  void _pauseChanged() {
-    if (_controller.isPaused) {
-      _subscription.pause();
-    } else {
-      _subscription.resume();
-    }
-  }
-
-  void _subscriptionChanged() {
-    if (!_controller.hasSubscribers) {
-      _subscription.cancel();
-    }
-  }
 }