Implement list based output stream and add pipe to list based input streams

runtime/bin/list_stream.dart

Index: runtime/bin/list_stream.dart
diff --git a/runtime/bin/list_stream.dart b/runtime/bin/list_stream.dart
new file mode 100644
index 0000000000000000000000000000000000000000..a00ba65231c12e8ac87028b45a26f34a478a3c95
--- /dev/null
+++ b/runtime/bin/list_stream.dart
@@ -0,0 +1,147 @@
+// Copyright (c) 2011, 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.
+
+class ListInputStream extends _BaseDataInputStream implements InputStream {

[Søren Gjesse, 2011/12/20 14:52:46]

Line 1 - 61 is unchanged (rename list_input_stream.dart -> list_stream.dart).
The rest is new.

+  ListInputStream(List<int> this._buffer) {
+    _streamMarkedClosed = true;
+  }
+
+  int available() => _buffer.length - _offset;
+
+  List<int> _read(int bytesToRead) {
+    if (_offset == 0 && bytesToRead == _buffer.length) {
+      _offset = _buffer.length;
+      return _buffer;
+    } else {
+      List<int> result = _buffer.getRange(_offset, bytesToRead);
+      _offset += bytesToRead;
+      return result;
+    }
+  }
+
+  int _readInto(List<int> buffer, int offset, int bytesToRead) {
+    buffer.setRange(offset, bytesToRead, _buffer, _offset);
+    _offset += bytesToRead;
+    return bytesToRead;
+  }
+
+  List<int> _buffer;
+  int _offset = 0;
+}
+
+
+class DynamicListInputStream
+    extends _BaseDataInputStream implements InputStream {
+  DynamicListInputStream() : _bufferList = new _BufferList();
+
+  int available() => _bufferList.length;
+
+  void write(List<int> data) {
+    _bufferList.add(data);
+    _checkScheduleCallbacks();
+  }
+
+  void markEndOfStream() {
+    _streamMarkedClosed = true;
+    _checkScheduleCallbacks();
+  }
+
+  List<int> _read(int bytesToRead) {
+    return _bufferList.readBytes(bytesToRead);
+  }
+
+  int _readInto(List<int> buffer, int offset, int bytesToRead) {
+    List<int> tmp = _bufferList.readBytes(byteToRead);
+    buffer.setRange(offset, bytesToRead, tmp, _offset);
+    return bytesToRead;
+  }
+
+  _BufferList _bufferList;
+}
+
+
+class ListOutputStream implements OutputStream {
+  ListOutputStream() : _bufferList = new _BufferList();
+
+  bool write(List<int> buffer, [bool copyBuffer = false]) {
+    if (_streamMarkedClosed) throw new StreamException.streamClosed();
+    if (copyBuffer) {
+      _bufferList.add(buffer.getRange(0, buffer.length));
+    } else {
+      _bufferList.add(buffer);
+    }
+  }
+
+  bool writeFrom(List<int> buffer, [int offset = 0, int len]) {
+    if (_streamMarkedClosed) throw new StreamException.streamClosed();
+    _bufferList.add(
+        buffer.getRange(offset, (len == null) ? buffer.length - offset : len));
+  }
+
+  void close() {
+    if (_streamMarkedClosed) throw new StreamException.streamClosed();
+    _streamMarkedClosed = true;
+  }
+
+  void destroy() {
+    close();
+  }
+
+  void set noPendingWriteHandler(void callback()) {
+    _clientNoPendingWriteHandler = callback;
+    _checkScheduleCallbacks();
+  }
+
+  void set closeHandler(void callback()) {
+    _clientCloseHandler = callback;
+  }
+
+  void set errorHandler(void callback()) {
+    // No errors emitted.
+  }
+
+  List<int> content() => _bufferList.readBytes(_bufferList.length);

[Mads Ager (google), 2011/12/20 15:52:46]

contents?

[Søren Gjesse, 2011/12/20 16:02:31]

Done.

+
+  void _checkScheduleCallbacks() {
+    void issueNoPendingWriteCallback(Timer timer) {
+      _scheduledNoPendingWriteCallback = null;
+      if (_clientNoPendingWriteHandler !== null) {
+        _clientNoPendingWriteHandler();
+        _checkScheduleCallbacks();
+      }
+    }
+
+    void issueCloseCallback(Timer timer) {
+      _scheduledCloseCallback = null;
+      if (_clientCloseHandler !== null) _clientCloseHandler();
+    }
+
+    // Schedule no pending callback if there is a callback set as this
+    // output stream does not wait for any tramsmission. Schedule

[Mads Ager (google), 2011/12/20 15:52:46]

transmission

[Søren Gjesse, 2011/12/20 16:02:31]

Done.

+    // close callback once when the stream is closed. Only schedule a
+    // new callback if the previous one has actually been called.
+    if (!_closeCallbackCalled) {
+      if (!_streamMarkedClosed) {
+        if (_clientNoPendingWriteHandler != null &&
+            _scheduledNoPendingWriteCallback == null) {
+          _scheduledNoPendingWriteCallback =
+              new Timer(issueNoPendingWriteCallback, 0);
+        }
+      } else if (_clientCloseHandler != null &&
+                 _streamMarkedClosed &&
+                 !_closeCallbackCalled) {
+        _scheduledCloseCallback = new Timer(issueCloseCallback, 0);
+        _closeCallbackCalled = true;
+      }
+    }
+  }
+
+  _BufferList _bufferList;
+  bool _streamMarkedClosed = false;
+  bool _closeCallbackCalled = false;
+  Timer _scheduledNoPendingWriteCallback;
+  Timer _scheduledCloseCallback;
+  Function _clientNoPendingWriteHandler;
+  Function _clientCloseHandler;
+}