Add input streams based on supplying data from lists

runtime/bin/stream_util.dart

 // 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 _BaseDataInputStream {
+  abstract int available();
+
+  List<int> read([int len]) {
+    if (_closeCallbackCalled) return null;
+    int bytesToRead = available();
+    if (bytesToRead == 0) {
+      _checkScheduleCallbacks();
+      return null;
+    }
+    if (len !== null) {
+      if (len <= 0) {
+        throw new StreamException("Illegal length $len");
+      } else if (bytesToRead > len) {
+        bytesToRead = len;
+      }
+    }
+    return _read(bytesToRead);
+  }
+
+  int readInto(List<int> buffer, [int offset = 0, int len]) {
+    if (_closeCallbackCalled) return null;
+    if (len === null) len = buffer.length;
+    if (offset < 0) throw new StreamException("Illegal offset $offset");
+    if (len < 0) throw new StreamException("Illegal length $len");
+    int bytesToRead = Math.min(len, available());
+    return _readInto(buffer, offset, bytesToRead);
+  }
+
+  bool get closed() => _closeCallbackCalled;
+
+  void set dataHandler(void callback()) {
+    _clientDataHandler = callback;
+    _checkScheduleCallbacks();
+  }
+
+  void set closeHandler(void callback()) {
+    _clientCloseHandler = callback;
+    _checkScheduleCallbacks();
+  }
+
+  void set errorHandler(void callback()) {
+    // No errors emitted by default.
+  }
+
+  abstract List<int> _read(int bytesToRead);
+
+  void _checkScheduleCallbacks() {
+    // TODO(sgjesse): Find a better way of scheduling callbacks from

[Mads Ager (google), 2011/12/19 15:04:18]

This comment used to be about the Timer with a 0ms argument. Maybe just remove
the TODO?

[Søren Gjesse, 2011/12/20 09:45:15]

Done.

+    // the event loop.
+    void issueDataCallback(Timer timer) {
+      _scheduledDataCallback = null;
+      if (_clientDataHandler !== null) {
+        _clientDataHandler();
+        _checkScheduleCallbacks();
+      }
+    }
+
+    void issueCloseCallback(Timer timer) {
+      _scheduledCloseCallback = null;
+      if (_clientCloseHandler !== null) _clientCloseHandler();
+    }
+
+    // Schedule data callback if there is more data to read. Schedule
+    // close callback once when all data has been read. Only schedule
+    // a new callback if the previous one has actually been called.
+    if (!_closeCallbackCalled) {
+      if (available() > 0) {
+        if (_scheduledDataCallback == null) {
+          _scheduledDataCallback = new Timer(issueDataCallback, 0);
+        }
+      } else if (_streamMarkedClosed && !_closeCallbackCalled) {
+        _scheduledCloseCallback = new Timer(issueCloseCallback, 0);
+        _closeCallbackCalled = true;
+      }
+    }
+  }
+
+  // Stream is marked closed available is now all remaining data.

[Mads Ager (google), 2011/12/19 15:04:18]

Comment not completely understood. I think you should have a period after closed
and then explain the point about 'available' with more words. :)

[Søren Gjesse, 2011/12/20 09:45:15]

Done.

+  bool _streamMarkedClosed = false;
+
+  // The close callback has now been called and stream is fully closed.

[Mads Ager (google), 2011/12/19 15:04:18]

and stream -> and the stream

[Søren Gjesse, 2011/12/20 09:45:15]

Done.

+  bool _closeCallbackCalled = false;
+
+  Timer _scheduledDataCallback;
+  Timer _scheduledCloseCallback;
+  var _clientDataHandler;
+  var _clientCloseHandler;
+}
+
+
 void _pipe(InputStream input, OutputStream output, [bool close]) {
   Function pipeDataHandler;
   Function pipeCloseHandler;
   Function pipeNoPendingWriteHandler;
 
   Function _inputCloseHandler;
 
   pipeDataHandler = () {
     List<int> data;
     while ((data = input.read()) !== null) {
@@ skipping 14 matching lines @@
     input.dataHandler = pipeDataHandler;
     output.noPendingWriteHandler = null;
   };
 
   _inputCloseHandler = input._clientCloseHandler;
   input.dataHandler = pipeDataHandler;
   input.closeHandler = pipeCloseHandler;
   output.noPendingWriteHandler = null;
 }