Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(258)

Side by Side Diff: runtime/bin/file_impl.dart

Issue 9474004: Make FileInputStream and FileOutputStream actually asynchronous. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Address comments. Created 8 years, 9 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch | Annotate | Revision Log
« no previous file with comments | « runtime/bin/file.dart ('k') | runtime/bin/stream_util.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a 2 // for details. All rights reserved. Use of this source code is governed by a
3 // BSD-style license that can be found in the LICENSE file. 3 // BSD-style license that can be found in the LICENSE file.
4 4
5 class _FileInputStream extends _BaseDataInputStream implements InputStream { 5 class _FileInputStream extends _BaseDataInputStream implements InputStream {
6 _FileInputStream(RandomAccessFile this._file, int this._length) { 6 _FileInputStream(String name) {
7 _streamMarkedClosed = true; 7 _file = new File(name);
8 _checkScheduleCallbacks(); 8 _data = [];
9 _position = 0;
10 _file.errorHandler = (String s) {
11 if (_clientErrorHandler != null) {
12 _clientErrorHandler();
13 }
14 };
15 _file.open();
16 _file.openHandler = (openedFile) {
17 _readDataFromFile(openedFile);
18 };
9 } 19 }
10 20
11 _FileInputStream.fromStdio(int fd) { 21 _FileInputStream.fromStdio(int fd) {
12 assert(fd == 0); 22 assert(fd == 0);
13 _file = _File._openStdioSync(fd); 23 _file = _File._openStdioSync(fd);
14 _length = _file.lengthSync(); 24 _data = [];
15 _streamMarkedClosed = true; 25 _position = 0;
16 _checkScheduleCallbacks(); 26 _readDataFromFile(_file);
27 }
28
29 void _readDataFromFile(RandomAccessFile openedFile) {
30 openedFile.errorHandler = (String s) {
31 if (_clientErrorHandler != null) {
32 _clientErrorHandler();
33 }
34 };
35 openedFile.length();
36 openedFile.lengthHandler = (length) {
37 var contents = new ByteArray(length);
38 if (length != 0) {
39 openedFile.readList(contents, 0, length);
40 openedFile.readListHandler = (read) {
41 if (read != length) {
42 if (_clientErrorHandler != null) {
43 _clientErrorHandler();
44 }
45 _streamMarkedClosed = true;
46 _checkScheduleCallbacks();
47 } else {
48 _data = contents;
49 _streamMarkedClosed = true;
50 _checkScheduleCallbacks();
51 }
52 openedFile.close();
53 };
54 } else {
55 _streamMarkedClosed = true;
56 _checkScheduleCallbacks();
57 openedFile.close();
58 }
59 };
17 } 60 }
18 61
19 int available() { 62 int available() {
20 return _closed ? 0 : _length - _file.positionSync(); 63 return _closed ? 0 : _data.length - _position;
21 } 64 }
22 65
23 void pipe(OutputStream output, [bool close = true]) { 66 void pipe(OutputStream output, [bool close = true]) {
24 _pipe(this, output, close: close); 67 _pipe(this, output, close: close);
25 } 68 }
26 69
27 List<int> _read(int bytesToRead) { 70 List<int> _read(int bytesToRead) {
28 ByteArray result = new ByteArray(bytesToRead); 71 ByteArray result = new ByteArray(bytesToRead);
29 int bytesRead = _file.readListSync(result, 0, bytesToRead); 72 result.setRange(0, bytesToRead, _data, _position);
30 if (bytesRead < bytesToRead) { 73 _position += bytesToRead;
31 ByteArray buffer = new ByteArray(bytesRead);
32 buffer.setRange(0, bytesRead, result);
33 result = buffer;
34 }
35 _checkScheduleCallbacks(); 74 _checkScheduleCallbacks();
36 return result; 75 return result;
37 } 76 }
38 77
39 int _readInto(List<int> buffer, int offset, int len) { 78 int _readInto(List<int> buffer, int offset, int len) {
40 int result = _file.readListSync(buffer, offset, len); 79 buffer.setRange(offset, len, _data, _position);
80 _position += len;
41 _checkScheduleCallbacks(); 81 _checkScheduleCallbacks();
42 return result; 82 return len;
43 } 83 }
44 84
45 void _close() { 85 void _close() {
46 if (_closed) return; 86 if (_closed) return;
47 _file.closeSync();
48 _closed = true; 87 _closed = true;
49 } 88 }
50 89
51 RandomAccessFile _file; 90 File _file;
52 int _length; 91 List<int> _data;
92 int _position;
53 bool _closed = false; 93 bool _closed = false;
54 } 94 }
55 95
56 96
57 class _FileOutputStream implements OutputStream { 97 class _FileOutputStream implements OutputStream {
58 _FileOutputStream(this._file); 98 _FileOutputStream(String name, int mode) {
99 _pendingOperations = new List<List<int>>();
100 var f = new File(name);
101 f.open(mode);
102 f.openHandler = (openedFile) {
103 _file = openedFile;
104 _setupFileHandlers();
105 _processPendingOperations();
106 };
107 f.errorHandler = (e) {
108 if (_errorHandler != null) _errorHandler();
109 };
110 }
59 111
60 _FileOutputStream.fromStdio(int fd) { 112 _FileOutputStream.fromStdio(int fd) {
61 assert(1 <= fd && fd <= 2); 113 assert(1 <= fd && fd <= 2);
62 _file = _File._openStdioSync(fd); 114 _file = _File._openStdioSync(fd);
115 _setupFileHandlers();
116 }
117
118
119 void _setupFileHandlers() {
120 _file.errorHandler = (e) {
121 if (_errorHandler != null) _errorHandler();
122 };
123 _file.noPendingWriteHandler = () {
124 if (!_streamMarkedClosed && _noPendingWriteHandler != null) {
125 _noPendingWriteHandler();
126 }
127 };
128 _file.closeHandler = () {
129 if (_closeHandler != null) _closeHandler();
130 };
63 } 131 }
64 132
65 bool write(List<int> buffer, [bool copyBuffer = false]) { 133 bool write(List<int> buffer, [bool copyBuffer = false]) {
66 bool result = _write(buffer, 0, buffer.length); 134 var data = buffer;
67 if (result) { 135 if (copyBuffer) {
68 _checkScheduleCallbacks(); 136 var length = buffer.length;
137 data = new ByteArray(length);
138 data.setRange(0, length, buffer, 0);
69 } 139 }
70 return result; 140 if (_file == null) {
141 _pendingOperations.add(data);
142 } else {
143 _write(data, 0, data.length);
144 }
145 return false;
71 } 146 }
72 147
73 bool writeFrom(List<int> buffer, [int offset = 0, int len]) { 148 bool writeFrom(List<int> buffer, [int offset = 0, int len]) {
74 bool result = _write( 149 // A copy is required by the interface.
75 buffer, offset, (len == null) ? buffer.length - offset : len); 150 var length = buffer.length - offset;
76 if (result) { 151 if (len != null) {
77 _checkScheduleCallbacks(); 152 if (len > length) throw new IndexOutOfRangeException(len);
153 length = len;
78 } 154 }
79 return result; 155 var copy = new ByteArray(length);
156 copy.setRange(0, length, buffer, offset);
157 return write(copy);
80 } 158 }
81 159
82 void close() { 160 void close() {
83 if (_scheduledNoPendingWriteCallback != null) { 161 if (_file == null) {
84 _scheduledNoPendingWriteCallback.cancel(); 162 _pendingOperations.add(null);
85 } 163 } else if (!_streamMarkedClosed) {
86 if (!_streamMarkedClosed) { 164 _file.close();
87 _file.closeSync();
88 _streamMarkedClosed = true; 165 _streamMarkedClosed = true;
89 _checkScheduleCallbacks();
90 } 166 }
91 } 167 }
92 168
93 void set noPendingWriteHandler(void callback()) { 169 void set noPendingWriteHandler(void callback()) {
94 _noPendingWriteHandler = callback; 170 _noPendingWriteHandler = callback;
95 _checkScheduleCallbacks();
96 } 171 }
97 172
98 void set closeHandler(void callback()) { 173 void set closeHandler(void callback()) {
99 _closeHandler = callback; 174 _closeHandler = callback;
100 _checkScheduleCallbacks();
101 } 175 }
102 176
103 void set errorHandler(void callback()) { 177 void set errorHandler(void callback()) {
104 // TODO(sgjesse): How to handle this? 178 _errorHandler = callback;
105 } 179 }
106 180
107 bool _write(List<int> buffer, int offset, int len) { 181 void _processPendingOperations() {
108 int bytesWritten = _file.writeListSync(buffer, offset, len); 182 _pendingOperations.forEach((buffer) {
109 if (bytesWritten == len) { 183 (buffer != null) ? write(buffer) : close();
110 return true; 184 });
111 } else { 185 _pendingOperations = null;
112 throw "FileOutputStream: write error";
113 }
114 } 186 }
115 187
116 void _checkScheduleCallbacks() { 188 void _write(List<int> buffer, int offset, int len) {
117 void issueNoPendingWriteCallback(Timer timer) { 189 _file.writeList(buffer, offset, len);
118 _scheduledNoPendingWriteCallback = null;
119 if (_noPendingWriteHandler !== null) {
120 _noPendingWriteHandler();
121 _checkScheduleCallbacks();
122 }
123 }
124
125 void issueCloseCallback(Timer timer) {
126 if (_closeHandler !== null) _closeHandler();
127 }
128
129 // Schedule no pending write callbacks if the stream is not yet
130 // closed and close callback if it is closing.
131 if (!_closeCallbackCalled) {
132 if (_scheduledNoPendingWriteCallback == null) {
133 _scheduledNoPendingWriteCallback =
134 new Timer(issueNoPendingWriteCallback, 0);
135 }
136 if (_streamMarkedClosed && _scheduledCloseCallback == null) {
137 _scheduledCloseCallback = new Timer(issueCloseCallback, 0);
138 }
139 }
140 } 190 }
141 191
142 RandomAccessFile _file; 192 RandomAccessFile _file;
143 193
144 // When this is set to true the stream is marked closed. When a 194 // When this is set to true the stream is marked closed. When a
145 // stream is marked closed no more data can be written. 195 // stream is marked closed no more data can be written.
146 bool _streamMarkedClosed = false; 196 bool _streamMarkedClosed = false;
147 197
148 // When this is set to true the close callback has been called and 198 // When this is set to true the close callback has been called and
149 // the stream is fully closed. 199 // the stream is fully closed.
150 bool _closeCallbackCalled = false; 200 bool _closeCallbackCalled = false;
151 201
152 Timer _scheduledNoPendingWriteCallback; 202 // List of pending writes that were issued before the underlying
153 Timer _scheduledCloseCallback; 203 // file was successfully opened.
204 List<List<int>> _pendingOperations;
205
154 Function _noPendingWriteHandler; 206 Function _noPendingWriteHandler;
155 Function _closeHandler; 207 Function _closeHandler;
208 Function _errorHandler;
156 } 209 }
157 210
158 211
159 // Helper class containing static file helper methods. 212 // Helper class containing static file helper methods.
160 class _FileUtils { 213 class _FileUtils {
161 static final kExistsRequest = 0; 214 static final kExistsRequest = 0;
162 static final kCreateRequest = 1; 215 static final kCreateRequest = 1;
163 static final kDeleteRequest = 2; 216 static final kDeleteRequest = 2;
164 static final kOpenRequest = 3; 217 static final kOpenRequest = 3;
165 static final kFullPathRequest = 4; 218 static final kFullPathRequest = 4;
(...skipping 294 matching lines...) Expand 10 before | Expand all | Expand 10 after
460 } 513 }
461 return result; 514 return result;
462 } 515 }
463 516
464 void openInputStream() { 517 void openInputStream() {
465 _asyncUsed = true; 518 _asyncUsed = true;
466 // Create a new file object to handle the opening of the file for 519 // Create a new file object to handle the opening of the file for
467 // creating an input stream. Currently the file input stream uses 520 // creating an input stream. Currently the file input stream uses
468 // synchronous calls on the opened file so we need to open it 521 // synchronous calls on the opened file so we need to open it
469 // synchronously. 522 // synchronously.
470 File file = new File(this._name); 523 new Timer((t) {
471 file.errorHandler = (String error) { 524 if (_inputStreamHandler != null) {
472 if (_errorHandler != null) _errorHandler(error); 525 _inputStreamHandler(new _FileInputStream(_name));
473 }; 526 }
474 RandomAccessFile openedFile = file.openSync(); 527 }, 0);
475 InputStream stream =
476 new _FileInputStream(openedFile, openedFile.lengthSync());
477 new Timer(
478 (Timer ignore) {
479 if (_inputStreamHandler != null) _inputStreamHandler(stream);
480 }, 0);
481 } 528 }
482 529
483 InputStream openInputStreamSync() { 530 InputStream openInputStreamSync() {
484 if (_asyncUsed) { 531 return new _FileInputStream(_name);
485 throw new FileIOException(
486 "Mixed use of synchronous and asynchronous API");
487 }
488 RandomAccessFile openedFile = openSync();
489 return new _FileInputStream(openedFile, openedFile.lengthSync());
490 } 532 }
491 533
492 void openOutputStream([FileMode mode = FileMode.WRITE]) { 534 void openOutputStream([FileMode mode = FileMode.WRITE]) {
493 _asyncUsed = true; 535 _asyncUsed = true;
494 if (mode != FileMode.WRITE && 536 if (mode != FileMode.WRITE &&
495 mode != FileMode.APPEND) { 537 mode != FileMode.APPEND) {
496 throw new FileIOException( 538 throw new FileIOException(
497 "Wrong FileMode. Use FileMode.WRITE or FileMode.APPEND"); 539 "Wrong FileMode. Use FileMode.WRITE or FileMode.APPEND");
498 } 540 }
499 // Create a new file object to handle the opening of the file for 541 OutputStream stream = new _FileOutputStream(_name, mode);
500 // creating an input stream. Currently the file input stream uses
501 // synchronous calls on the opened file so we need to open it
502 // synchronously.
503 File file = new File(this._name);
504 file.errorHandler = (String error) {
505 if (_errorHandler != null) _errorHandler(error);
506 };
507 RandomAccessFile openedFile = file.openSync(mode);
508 OutputStream stream = new _FileOutputStream(openedFile);
509 new Timer( 542 new Timer(
510 (Timer ignore) { 543 (Timer ignore) {
511 if (_outputStreamHandler != null) _outputStreamHandler(stream); 544 if (_outputStreamHandler != null) _outputStreamHandler(stream);
512 }, 0); 545 }, 0);
513 } 546 }
514 547
515 OutputStream openOutputStreamSync([FileMode mode = FileMode.WRITE]) { 548 OutputStream openOutputStreamSync([FileMode mode = FileMode.WRITE]) {
516 if (_asyncUsed) { 549 if (_asyncUsed) {
517 throw new FileIOException( 550 throw new FileIOException(
518 "Mixed use of synchronous and asynchronous API"); 551 "Mixed use of synchronous and asynchronous API");
519 } 552 }
520 if (mode != FileMode.WRITE && 553 if (mode != FileMode.WRITE &&
521 mode != FileMode.APPEND) { 554 mode != FileMode.APPEND) {
522 throw new FileIOException( 555 throw new FileIOException(
523 "Wrong FileMode. Use FileMode.WRITE or FileMode.APPEND"); 556 "Wrong FileMode. Use FileMode.WRITE or FileMode.APPEND");
524 } 557 }
525 RandomAccessFile openedFile = openSync(mode); 558 return new _FileOutputStream(_name, mode);
526 return new _FileOutputStream(openedFile);
527 } 559 }
528 560
529 void readAsBytes() { 561 void readAsBytes() {
530 _asyncUsed = true; 562 _asyncUsed = true;
531 var chunks = new _BufferList(); 563 var chunks = new _BufferList();
532 openInputStream(); 564 var stream = openInputStreamSync();
533 inputStreamHandler = (inputStream) { 565 stream.closeHandler = () {
534 inputStream.closeHandler = () { 566 if (_readAsBytesHandler != null) {
535 if (_readAsBytesHandler != null) { 567 _readAsBytesHandler(chunks.readBytes(chunks.length));
536 _readAsBytesHandler(chunks.readBytes(chunks.length)); 568 }
537 } 569 };
538 }; 570 stream.dataHandler = () {
539 inputStream.dataHandler = () { 571 var chunk = stream.read();
540 var chunk = inputStream.read(); 572 chunks.add(chunk);
541 chunks.add(chunk); 573 };
542 }; 574 stream.errorHandler = () {
543 inputStream.errorHandler = () { 575 if (_errorHandler != null) {
544 if (_errorHandler != null) { 576 _errorHandler("Failed to read file as bytes: $_name");
545 _errorHandler("Failed to read file as bytes: $_name"); 577 }
546 }
547 };
548 }; 578 };
549 } 579 }
550 580
551 List<int> readAsBytesSync() { 581 List<int> readAsBytesSync() {
552 if (_asyncUsed) { 582 if (_asyncUsed) {
553 throw new FileIOException( 583 throw new FileIOException(
554 "Mixed use of synchronous and asynchronous API"); 584 "Mixed use of synchronous and asynchronous API");
555 } 585 }
556 var opened = openSync(); 586 var opened = openSync();
557 var length = opened.lengthSync(); 587 var length = opened.lengthSync();
558 var result = new ByteArray(length); 588 var result = new ByteArray(length);
559 var read = opened.readListSync(result, 0, length); 589 var read = opened.readListSync(result, 0, length);
560 if (read != length) { 590 if (read != length) {
561 throw new FileIOException("Failed reading file as bytes: $_name"); 591 throw new FileIOException("Failed reading file as bytes: $_name");
562 } 592 }
593 opened.close();
563 return result; 594 return result;
564 } 595 }
565 596
566 _StringDecoder _getDecoder(encoding) { 597 _StringDecoder _getDecoder(encoding) {
567 if (encoding == "UTF-8") { 598 if (encoding == "UTF-8") {
568 return new _UTF8Decoder(); 599 return new _UTF8Decoder();
569 } else if (encoding == "ISO-8859-1") { 600 } else if (encoding == "ISO-8859-1") {
570 return new _Latin1Decoder(); 601 return new _Latin1Decoder();
571 } else if (encoding == "ASCII") { 602 } else if (encoding == "ASCII") {
572 return new _AsciiDecoder(); 603 return new _AsciiDecoder();
(...skipping 526 matching lines...) Expand 10 before | Expand all | Expand 10 after
1099 void set readByteHandler(void handler(int byte)) { 1130 void set readByteHandler(void handler(int byte)) {
1100 _readByteHandler = handler; 1131 _readByteHandler = handler;
1101 } 1132 }
1102 1133
1103 void set readListHandler(void handler(int read)) { 1134 void set readListHandler(void handler(int read)) {
1104 _readListHandler = handler; 1135 _readListHandler = handler;
1105 } 1136 }
1106 1137
1107 void set noPendingWriteHandler(void handler()) { 1138 void set noPendingWriteHandler(void handler()) {
1108 _noPendingWriteHandler = handler; 1139 _noPendingWriteHandler = handler;
1140 if (_pendingWrites == 0) {
1141 _noPendingWriteTimer = new Timer((t) {
1142 if (_noPendingWriteHandler != null) _noPendingWriteHandler();
1143 }, 0);
1144 }
1109 } 1145 }
1110 1146
1111 void set positionHandler(void handler(int pos)) { 1147 void set positionHandler(void handler(int pos)) {
1112 _positionHandler = handler; 1148 _positionHandler = handler;
1113 } 1149 }
1114 1150
1115 void set setPositionHandler(void handler()) { 1151 void set setPositionHandler(void handler()) {
1116 _setPositionHandler = handler; 1152 _setPositionHandler = handler;
1117 } 1153 }
1118 1154
1119 void set truncateHandler(void handler()) { 1155 void set truncateHandler(void handler()) {
1120 _truncateHandler = handler; 1156 _truncateHandler = handler;
1121 } 1157 }
1122 1158
1123 void set lengthHandler(void handler(int length)) { 1159 void set lengthHandler(void handler(int length)) {
1124 _lengthHandler = handler; 1160 _lengthHandler = handler;
1125 } 1161 }
1126 1162
1127 void set flushHandler(void handler()) { 1163 void set flushHandler(void handler()) {
1128 _flushHandler = handler; 1164 _flushHandler = handler;
1129 } 1165 }
1130 1166
1131 void _ensureFileService() { 1167 void _ensureFileService() {
1132 if (_fileService == null) { 1168 if (_fileService == null) {
1133 _fileService = _FileUtils.newServicePort(); 1169 _fileService = _FileUtils.newServicePort();
1134 } 1170 }
1135 } 1171 }
1136 1172
1137 void _writeEnqueued() => _pendingWrites++; 1173 void _writeEnqueued() {
1174 _pendingWrites++;
1175 if (_noPendingWriteTimer != null) {
1176 _noPendingWriteTimer.cancel();
1177 _noPendingWriteTimer = null;
1178 }
1179 }
1138 1180
1139 void _writeCompleted() { 1181 void _writeCompleted() {
1140 _pendingWrites--; 1182 _pendingWrites--;
1141 if (_pendingWrites == 0 && _noPendingWriteHandler != null) { 1183 if (_pendingWrites == 0 && _noPendingWriteHandler != null) {
1142 _noPendingWriteHandler(); 1184 _noPendingWriteHandler();
1143 } 1185 }
1144 } 1186 }
1145 1187
1146 1188
1147 String _name; 1189 String _name;
1148 int _id; 1190 int _id;
1149 bool _asyncUsed; 1191 bool _asyncUsed;
1150 int _pendingWrites = 0; 1192 int _pendingWrites = 0;
1151 1193
1152 SendPort _fileService; 1194 SendPort _fileService;
1153 1195
1196 Timer _noPendingWriteTimer;
1197
1154 Function _closeHandler; 1198 Function _closeHandler;
1155 Function _readByteHandler; 1199 Function _readByteHandler;
1156 Function _readListHandler; 1200 Function _readListHandler;
1157 Function _noPendingWriteHandler; 1201 Function _noPendingWriteHandler;
1158 Function _positionHandler; 1202 Function _positionHandler;
1159 Function _setPositionHandler; 1203 Function _setPositionHandler;
1160 Function _truncateHandler; 1204 Function _truncateHandler;
1161 Function _lengthHandler; 1205 Function _lengthHandler;
1162 Function _flushHandler; 1206 Function _flushHandler;
1163 Function _errorHandler; 1207 Function _errorHandler;
1164 } 1208 }
OLDNEW
« no previous file with comments | « runtime/bin/file.dart ('k') | runtime/bin/stream_util.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698