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

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

Issue 8413034: New OutputStream interface (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 9 years, 1 month 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/socket_impl.dart ('k') | tests/standalone/src/EchoServerStreamTest.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) 2011, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2011, 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 SocketInputStream implements InputStream { 5 class SocketInputStream implements InputStream {
6 SocketInputStream(Socket socket) { 6 SocketInputStream(Socket socket) {
7 _socket = socket; 7 _socket = socket;
8 } 8 }
9 9
10 List<int> read([int len]) { 10 List<int> read([int len]) {
(...skipping 38 matching lines...) Expand 10 before | Expand all | Expand 10 after
49 } 49 }
50 50
51 void set errorHandler(void callback()) { 51 void set errorHandler(void callback()) {
52 _socket.setErrorHandler(callback); 52 _socket.setErrorHandler(callback);
53 } 53 }
54 54
55 Socket _socket; 55 Socket _socket;
56 } 56 }
57 57
58 58
59 class _BufferList2 {
Søren Gjesse 2011/10/28 12:49:25 This is quite similar to the BufferList class used
60 _BufferList2() : _index = 0, _length = 0, _buffers = new Queue();
61
62 // Adds a new buffer to the list possibly with an offset of the
63 // first byte of interest. The offset can only be specified if the
64 // buffer list is empty.
65 void add(List<int> buffer, [int offset = 0]) {
66 assert(offset == 0 || _buffers.isEmpty());
67 _buffers.addLast(buffer);
68 _length += buffer.length;
69 if (offset != 0) _index = offset;
70 }
71
72 List<int> get first() => _buffers.first();
73 int get index() => _index;
74
75 void removeBytes(int count) {
76 int firstRemaining = first.length - _index;
77 assert(count <= firstRemaining);
78 if (count == firstRemaining) {
79 _buffers.removeFirst();
80 _index = 0;
81 } else {
82 _index += count;
83 }
84 _length -= count;
85 }
86
87 int get length() => _length;
88
89 bool get empty() => _buffers.isEmpty();
90
91 int _length; // Total length of pending data.
92 Queue<List<int>> _buffers;
93 int _index; // Offset into the first buffer of next write position.
94 }
95
96
59 class SocketOutputStream implements OutputStream { 97 class SocketOutputStream implements OutputStream {
60 SocketOutputStream(Socket socket) : _socket = socket; 98 SocketOutputStream(Socket socket)
99 : _socket = socket, _pendingWrites = new _BufferList2() {
100 _socket.setWriteHandler(_writeHandler);
101 _socket.setErrorHandler(_errorHandler);
102 }
61 103
62 bool write(List<int> buffer, int offset, int len, void callback()) { 104 bool write(List<int> buffer) {
63 int bytesWritten = _socket.writeList(buffer, offset, len); 105 return _write(buffer, 0, buffer.length, false);
106 }
64 107
65 void finishWrite() { 108 bool writeFrom(List<int> buffer, [int offset = 0, int len]) {
66 bytesWritten += _socket.writeList( 109 return _write(buffer, offset, (len == null) ? buffer.length : len, true);
67 buffer, offset + bytesWritten, len - bytesWritten); 110 }
68 if (bytesWritten < len) { 111
69 _socket.setWriteHandler(finishWrite); 112 void end() {
70 } else { 113 if (_ending || _ended) throw new StreamException("Stream ended");
71 assert(bytesWritten == len); 114 _ending = true;
72 if (callback !== null) { 115 if (_pendingWrites.empty) {
73 callback(); 116 close();
74 } 117 }
75 } 118 }
119
120 void close() {
121 _socket.setWriteHandler(null);
122 _pendingWrites.clear();
123 _socket.close();
124 _ended = true;
125 }
126
127 void set noPendingWriteHandler(void callback()) {
128 _noPendingWriteHandler = callback;
129 _socket.setWriteHandler(_writeHandler);
130 }
131
132 void set closeHandler(void callback()) {
133 _socket.setCloseHandler(callback());
134 }
135
136 void set errorHandler(void callback()) {
137 _streamErrorHandler = callback();
138 }
139
140 bool _write(List<int> buffer, int offset, int len, bool copyBuffer) {
141 if (_ending || _ended) throw new StreamException("Stream ended");
142 if (len == null) len = buffer.length;
143 int bytesWritten = 0;
144 if (_pendingWrites.empty) {
145 // If nothing is buffered write as much as possible and buffer
146 // the rest.
147 bytesWritten = _socket.writeList(buffer, offset, len);
148 if (bytesWritten == len) return true;
76 } 149 }
77 150
78 if (bytesWritten == len) { 151 // Place remaining data on the pending writes queue.
79 return true; 152 if (copyBuffer) {
153 List<int> newBuffer =
154 new List.fromList(buffer, offset + bytesWritten, buffer.length);
155 _pendingWrites.add(newBuffer);
156 } else {
157 _pendingWrites.add(buffer, bytesWritten);
80 } 158 }
81 _socket.setWriteHandler(finishWrite); 159 }
82 return false; 160
161 void _writeHandler() {
162 _socket.setWriteHandler(_writeHandler);
163 // Write as much buffered data to the socket as possible.
164 while (!_pendingWrites.empty) {
165 List<int> buffer = _pendingWrites.first;
166 int offset = _pendingWrites.index;
167 int bytesToWrite = buffer.length - offset;
168 int bytesWritten = _socket.writeList(buffer, offset, bytesToWrite);
169 _pendingWrites.removeBytes(bytesWritten);
170 if (bytesWritten < bytesToWrite) return;
171 }
172
173 // All buffered data was written.
174 if (_ending) {
175 _socket.close();
176 _ended = true;
177 } else {
178 if (_noPendingWriteHandler != null) _noPendingWriteHandler();
179 }
180 }
181
182 void _errorHandler() {
183 close();
184 if (_streamErrorHandler != null) _streamErrorHandler();
83 } 185 }
84 186
85 Socket _socket; 187 Socket _socket;
188 _BufferList2 _pendingWrites;
189 bool _ending = false;
190 bool _ended = false;
191 var _noPendingWriteHandler;
192 var _streamErrorHandler;
86 } 193 }
OLDNEW
« no previous file with comments | « runtime/bin/socket_impl.dart ('k') | tests/standalone/src/EchoServerStreamTest.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698