Redo StreamTransformers so they work with Stack traces.

sdk/lib/io/websocket_impl.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;
 
 const String _webSocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
 
 class _WebSocketMessageType {
   static const int NONE = 0;
@@ skipping 23 matching lines @@
 
 /**
  * The web socket protocol transformer handles the protocol byte stream
  * which is supplied through the [:handleData:]. As the protocol is processed,
  * it'll output frame data as either a List<int> or String.
  *
  * Important infomation about usage: Be sure you use cancelOnError, so the
  * socket will be closed when the processer encounter an error. Not using it
  * will lead to undefined behaviour.
  */
-class _WebSocketProtocolTransformer extends StreamEventTransformer {
+class _WebSocketProtocolTransformer implements StreamTransformer, EventSink {
   static const int START = 0;
   static const int LEN_FIRST = 1;
   static const int LEN_REST = 2;
   static const int MASK = 3;
   static const int PAYLOAD = 4;
   static const int CLOSED = 5;
   static const int FAILURE = 6;
 
   bool _serverSide;
+  EventSink _eventSink;
 
   _WebSocketProtocolTransformer([bool this._serverSide = false]) {
     _prepareForNextFrame();
     _currentMessageType = _WebSocketMessageType.NONE;
   }
 
+  Stream bind(Stream stream) {
+    return new Stream.eventTransformed(
+        stream,
+        (EventSink eventSink) {
+          if (_eventSink != null) {
+            throw new StateError("WebSocket transformer already used.");
+          }
+          _eventSink = eventSink;
+          return this;
+        });
+  }
+
+  void addError(Object error, [StackTrace stackTrace]) {
+    _eventSink.addError(error, stackTrace);
+  }
+
+  void close() => _eventSink.close();
+
   /**
    * Process data received from the underlying communication channel.
    */
-  void handleData(Uint8List buffer, EventSink sink) {
+  void add(Uint8List buffer) {
     int count = buffer.length;
     int index = 0;
     int lastIndex = count;
     try {
       if (_state == CLOSED) {
         throw new WebSocketException("Data on closed connection");
       }
       if (_state == FAILURE) {
         throw new WebSocketException("Data on failed connection");
       }
@@ skipping 17 matching lines @@
             case _WebSocketOpcode.TEXT:
               if (_currentMessageType != _WebSocketMessageType.NONE) {
                 throw new WebSocketException("Protocol error");
               }
               _currentMessageType = _WebSocketMessageType.TEXT;
               _controller = new StreamController(sync: true);
               _controller.stream
                   .transform(UTF8.decoder)
                   .fold(new StringBuffer(), (buffer, str) => buffer..write(str))
                   .then((buffer) {
-                    sink.add(buffer.toString());
-                  }, onError: sink.addError);
+                    _eventSink.add(buffer.toString());
+                  }, onError: _eventSink.addError);
               break;
 
             case _WebSocketOpcode.BINARY:
               if (_currentMessageType != _WebSocketMessageType.NONE) {
                 throw new WebSocketException("Protocol error");
               }
               _currentMessageType = _WebSocketMessageType.BINARY;
               _controller = new StreamController(sync: true);
               _controller.stream
                   .fold(new BytesBuilder(), (buffer, data) => buffer..add(data))
                   .then((buffer) {
-                    sink.add(buffer.takeBytes());
-                  }, onError: sink.addError);
+                    _eventSink.add(buffer.takeBytes());
+                  }, onError: _eventSink.addError);
               break;
 
             case _WebSocketOpcode.CLOSE:
             case _WebSocketOpcode.PING:
             case _WebSocketOpcode.PONG:
               // Control frames cannot be fragmented.
               if (!_fin) throw new WebSocketException("Protocol error");
               break;
 
             default:
               throw new WebSocketException("Protocol error");
             }
             _state = LEN_FIRST;
             break;
 
           case LEN_FIRST:
             _masked = (byte & 0x80) != 0;
             _len = byte & 0x7F;
             if (_isControlFrame() && _len > 125) {
               throw new WebSocketException("Protocol error");
             }
             if (_len < 126) {
-              _lengthDone(sink);
+              _lengthDone();
             } else if (_len == 126) {
               _len = 0;
               _remainingLenBytes = 2;
               _state = LEN_REST;
             } else if (_len == 127) {
               _len = 0;
               _remainingLenBytes = 8;
               _state = LEN_REST;
             }
             break;
 
           case LEN_REST:
             _len = _len << 8 | byte;
             _remainingLenBytes--;
             if (_remainingLenBytes == 0) {
-              _lengthDone(sink);
+              _lengthDone();
             }
             break;
 
           case MASK:
             _maskingKey = _maskingKey << 8 | byte;
             _remainingMaskingKeyBytes--;
             if (_remainingMaskingKeyBytes == 0) {
-              _maskDone(sink);
+              _maskDone();
             }
             break;
 
           case PAYLOAD:
             // The payload is not handled one byte at a time but in blocks.
             int payload;
             if (lastIndex - index <= _remainingPayloadBytes) {
               payload = lastIndex - index;
             } else {
               payload = _remainingPayloadBytes;
@@ skipping 15 matching lines @@
                 // Allocate a buffer for collecting the control frame
                 // payload if any.
                 if (_controlPayload == null) {
                   _controlPayload = new List<int>();
                 }
                 _controlPayload.addAll(buffer.sublist(index, index + payload));
                 index += payload;
               }
 
               if (_remainingPayloadBytes == 0) {
-                _controlFrameEnd(sink);
+                _controlFrameEnd();
               }
             } else {
               if (_currentMessageType != _WebSocketMessageType.TEXT &&
                   _currentMessageType != _WebSocketMessageType.BINARY) {
                   throw new WebSocketException("Protocol error");
               }
               _controller.add(
                   new Uint8List.view(buffer.buffer, index, payload));
               index += payload;
               if (_remainingPayloadBytes == 0) {
-                _messageFrameEnd(sink);
+                _messageFrameEnd();
               }
             }
 
             // Hack - as we always do index++ below.
             index--;
             break;
         }
 
         // Move to the next byte.
         index++;
       }
-    } catch (e) {
+    } catch (e, stackTrace) {
       _state = FAILURE;
-      sink.addError(e);
+      _eventSink.addError(e, stackTrace);
     }
   }
 
-  void _lengthDone(EventSink sink) {
+  void _lengthDone() {
     if (_masked) {
       if (!_serverSide) {
         throw new WebSocketException("Received masked frame from server");
       }
       _state = MASK;
       _remainingMaskingKeyBytes = 4;
     } else {
       if (_serverSide) {
         throw new WebSocketException("Received unmasked frame from client");
       }
       _remainingPayloadBytes = _len;
-      _startPayload(sink);
+      _startPayload();
     }
   }
 
-  void _maskDone(EventSink sink) {
+  void _maskDone() {
     _remainingPayloadBytes = _len;
-    _startPayload(sink);
+    _startPayload();
   }
 
-  void _startPayload(EventSink sink) {
+  void _startPayload() {
     // If there is no actual payload perform perform callbacks without
     // going through the PAYLOAD state.
     if (_remainingPayloadBytes == 0) {
       if (_isControlFrame()) {
         switch (_opcode) {
           case _WebSocketOpcode.CLOSE:
             _state = CLOSED;
-            sink.close();
+            _eventSink.close();
             break;
           case _WebSocketOpcode.PING:
-            sink.add(new _WebSocketPing());
+            _eventSink.add(new _WebSocketPing());
             break;
           case _WebSocketOpcode.PONG:
-            sink.add(new _WebSocketPong());
+            _eventSink.add(new _WebSocketPong());
             break;
         }
         _prepareForNextFrame();
       } else {
-        _messageFrameEnd(sink);
+        _messageFrameEnd();
       }
     } else {
       _state = PAYLOAD;
     }
   }
 
-  void _messageFrameEnd(EventSink sink) {
+  void _messageFrameEnd() {
     if (_fin) {
       switch (_currentMessageType) {
         case _WebSocketMessageType.TEXT:
           _controller.close();
           break;
         case _WebSocketMessageType.BINARY:
           _controller.close();
           break;
       }
       _controller = null;
       _currentMessageType = _WebSocketMessageType.NONE;
     }
     _prepareForNextFrame();
   }
 
-  void _controlFrameEnd(EventSink sink) {
+  void _controlFrameEnd() {
     switch (_opcode) {
       case _WebSocketOpcode.CLOSE:
         closeCode = WebSocketStatus.NO_STATUS_RECEIVED;
         if (_controlPayload.length > 0) {
           if (_controlPayload.length == 1) {
             throw new WebSocketException("Protocol error");
           }
           closeCode = _controlPayload[0] << 8 | _controlPayload[1];
           if (closeCode == WebSocketStatus.NO_STATUS_RECEIVED) {
             throw new WebSocketException("Protocol error");
           }
           if (_controlPayload.length > 2) {
             closeReason = UTF8.decode(_controlPayload.sublist(2));
           }
         }
         _state = CLOSED;
-        sink.close();
+        _eventSink.close();
         break;
 
       case _WebSocketOpcode.PING:
-        sink.add(new _WebSocketPing(_controlPayload));
+        _eventSink.add(new _WebSocketPing(_controlPayload));
         break;
 
       case _WebSocketOpcode.PONG:
-        sink.add(new _WebSocketPong(_controlPayload));
+        _eventSink.add(new _WebSocketPong(_controlPayload));
         break;
     }
     _prepareForNextFrame();
   }
 
   bool _isControlFrame() {
     return _opcode == _WebSocketOpcode.CLOSE ||
            _opcode == _WebSocketOpcode.PING ||
            _opcode == _WebSocketOpcode.PONG;
   }
@@ skipping 107 matching lines @@
     }
     String key = request.headers.value("Sec-WebSocket-Key");
     if (key == null) {
       return false;
     }
     return true;
   }
 }
 
 
-class _WebSocketOutgoingTransformer extends StreamEventTransformer {
+class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink {
   final _WebSocketImpl webSocket;
+  EventSink _eventSink;
 
   _WebSocketOutgoingTransformer(_WebSocketImpl this.webSocket);
 
-  void handleData(message, EventSink<List<int>> sink) {
+  Stream bind(Stream stream) {
+    return new Stream.eventTransformed(
+        stream,
+        (EventSink eventSink) {
+          if (_eventSink != null) {
+            throw new StateError("WebSocket transformer already used");

[Lasse Reichstein Nielsen, 2013/10/07 11:47:00]

If it didn't try to put everything into one class, it would be reusable.

[floitsch, 2013/10/10 15:39:57]

Agreed.
Added a TODO. I just wanted to keep the changes small.

+          }
+          _eventSink = eventSink;
+          return this;
+        });
+  }
+
+  void add(message) {
     if (message is _WebSocketPong) {
-      addFrame(_WebSocketOpcode.PONG, message.payload, sink);
+      addFrame(_WebSocketOpcode.PONG, message.payload);
       return;
     }
     if (message is _WebSocketPing) {
-      addFrame(_WebSocketOpcode.PING, message.payload, sink);
+      addFrame(_WebSocketOpcode.PING, message.payload);
       return;
     }
     List<int> data;
     int opcode;
     if (message != null) {
       if (message is String) {
         opcode = _WebSocketOpcode.TEXT;
         data = UTF8.encode(message);
       } else {
         if (message is !List<int>) {
           throw new ArgumentError(message);
         }
         opcode = _WebSocketOpcode.BINARY;
         data = message;
       }
     } else {
       opcode = _WebSocketOpcode.TEXT;
     }
-    addFrame(opcode, data, sink);
+    addFrame(opcode, data);
   }
 
-  void handleDone(EventSink<List<int>> sink) {
+  void addError(Object error, [StackTrace stackTrace]) {
+    _eventSink.addError(error, stackTrace);
+  }
+
+  void close() {
     int code = webSocket._outCloseCode;
     String reason = webSocket._outCloseReason;
     List<int> data;
     if (code != null) {
       data = new List<int>();
       data.add((code >> 8) & 0xFF);
       data.add(code & 0xFF);
       if (reason != null) {
         data.addAll(UTF8.encode(reason));
       }
     }
-    addFrame(_WebSocketOpcode.CLOSE, data, sink);
-    sink.close();
+    addFrame(_WebSocketOpcode.CLOSE, data);
+    _eventSink.close();
   }
 
-  void addFrame(int opcode, List<int> data, EventSink<List<int>> sink) {
-    createFrame(opcode, data, webSocket._serverSide).forEach(sink.add);
+  void addFrame(int opcode, List<int> data) {
+    createFrame(opcode, data, webSocket._serverSide).forEach(_eventSink.add);
   }
 
   static Iterable createFrame(int opcode, List<int> data, bool serverSide) {
     bool mask = !serverSide;  // Masking not implemented for server.
     int dataLength = data == null ? 0 : data.length;
     // Determine the header size.
     int headerSize = (mask) ? 6 : 2;
     if (dataLength > 65535) {
       headerSize += 8;
     } else if (dataLength > 125) {
@@ skipping 386 matching lines @@
            (code < WebSocketStatus.NORMAL_CLOSURE ||
             code == WebSocketStatus.RESERVED_1004 ||
             code == WebSocketStatus.NO_STATUS_RECEIVED ||
             code == WebSocketStatus.ABNORMAL_CLOSURE ||
             (code > WebSocketStatus.INTERNAL_SERVER_ERROR &&
              code < WebSocketStatus.RESERVED_1015) ||
             (code >= WebSocketStatus.RESERVED_1015 &&
              code < 3000));
   }
 }