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

Side by Side Diff: sdk/lib/io/http_impl.dart

Issue 12317147: Implement addStream for HttpClientRequest/HttpResponse and propegate all write-errors from the sock… (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 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
OLDNEW
1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2013, 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 part of dart.io; 5 part of dart.io;
6 6
7 class _HttpIncoming 7 class _HttpIncoming
8 extends Stream<List<int>> implements StreamSink<List<int>> { 8 extends Stream<List<int>> implements StreamSink<List<int>> {
9 final int _transferLength; 9 final int _transferLength;
10 final Completer _dataCompleter = new Completer(); 10 final Completer _dataCompleter = new Completer();
(...skipping 336 matching lines...) Expand 10 before | Expand all | Expand 10 after
347 void add(List<int> data) { 347 void add(List<int> data) {
348 _writeHeaders(); 348 _writeHeaders();
349 if (_ignoreBody) return; 349 if (_ignoreBody) return;
350 if (_chunked) { 350 if (_chunked) {
351 _ChunkedTransformer._addChunk(data, super.add); 351 _ChunkedTransformer._addChunk(data, super.add);
352 } else { 352 } else {
353 super.add(data); 353 super.add(data);
354 } 354 }
355 } 355 }
356 356
357 Future addStream(Stream<List<int>> stream) {
358 _writeHeaders();
359 if (_ignoreBody) return new Future.immediate(this);
360 if (_chunked) {
361 // Transform when chunked.
362 stream = stream.transform(new _ChunkedTransformer(writeEnd: false));
363 }
364 return super.addStream(stream).then((_) => this);
365 }
366
357 void close() { 367 void close() {
358 if (!_headersWritten && !_ignoreBody && headers.chunkedTransferEncoding) { 368 if (!_headersWritten && !_ignoreBody && headers.chunkedTransferEncoding) {
359 // If no body was written, _ignoreBody is false (it's not a HEAD 369 // If no body was written, _ignoreBody is false (it's not a HEAD
360 // request) and the content-length is unspecified, set contentLength to 0. 370 // request) and the content-length is unspecified, set contentLength to 0.
361 headers.contentLength = 0; 371 headers.contentLength = 0;
362 } 372 }
363 _writeHeaders(); 373 _writeHeaders();
364 if (!_ignoreBody) { 374 if (!_ignoreBody) {
365 if (_chunked) { 375 if (_chunked) {
366 _ChunkedTransformer._addChunk([], super.add); 376 _ChunkedTransformer._addChunk([], super.add);
(...skipping 296 matching lines...) Expand 10 before | Expand all | Expand 10 after
663 headers._finalize(); 673 headers._finalize();
664 674
665 // Write headers. 675 // Write headers.
666 headers._write(this); 676 headers._write(this);
667 writeCRLF(); 677 writeCRLF();
668 } 678 }
669 } 679 }
670 680
671 681
672 // Transformer that transforms data to HTTP Chunked Encoding. 682 // Transformer that transforms data to HTTP Chunked Encoding.
673 class _ChunkedTransformer implements StreamTransformer<List<int>, List<int>> { 683 class _ChunkedTransformer extends StreamEventTransformer<List<int>, List<int>> {
674 final StreamController<List<int>> _controller 684 final bool writeEnd;
675 = new StreamController<List<int>>(); 685 _ChunkedTransformer({this.writeEnd: true});
676 686
677 Stream<List<int>> bind(Stream<List<int>> stream) { 687 void handleData(List<int> data, StreamSink<List<int>> sink) {
678 var subscription = stream.listen( 688 _addChunk(data, sink.add);
679 (data) { 689 }
680 if (data.length == 0) return; // Avoid close on 0-bytes payload. 690
681 _addChunk(data, _controller.add); 691 void handleDone(StreamSink<List<int>> sink) {
682 }, 692 if (writeEnd) {
683 onDone: () { 693 _addChunk([], sink.add);
684 _addChunk([], _controller.add); 694 }
685 _controller.close(); 695 sink.close();
686 });
687 return _controller.stream;
688 } 696 }
689 697
690 static void _addChunk(List<int> data, void add(List<int> data)) { 698 static void _addChunk(List<int> data, void add(List<int> data)) {
691 add(_chunkHeader(data.length)); 699 add(_chunkHeader(data.length));
692 if (data.length > 0) add(data); 700 if (data.length > 0) add(data);
693 add(_chunkFooter); 701 add(_chunkFooter);
694 } 702 }
695 703
696 static List<int> _chunkHeader(int length) { 704 static List<int> _chunkHeader(int length) {
697 const hexDigits = const [0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 705 const hexDigits = const [0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37,
(...skipping 203 matching lines...) Expand 10 before | Expand all | Expand 10 after
901 }, test: (error) => error is StateError) 909 }, test: (error) => error is StateError)
902 .catchError((error) { 910 .catchError((error) {
903 // We are done with the socket. 911 // We are done with the socket.
904 destroy(); 912 destroy();
905 request._onError(error); 913 request._onError(error);
906 }); 914 });
907 915
908 return _socket.addStream(stream) 916 return _socket.addStream(stream)
909 .catchError((e) { 917 .catchError((e) {
910 destroy(); 918 destroy();
911 if (e.error is HttpException) throw e; 919 throw e;
912 // TODO(ajohnsen): Where to send Socket errors?
913 }); 920 });
914 }); 921 });
915 return request; 922 return request;
916 } 923 }
917 924
918 Future<Socket> detachSocket() { 925 Future<Socket> detachSocket() {
919 return _streamFuture 926 return _streamFuture
920 .then((_) => new _DetachedSocket(_socket, _httpParser.detachIncoming()), 927 .then((_) => new _DetachedSocket(_socket, _httpParser.detachIncoming()),
921 onError: (_) {}); 928 onError: (_) {});
922 } 929 }
(...skipping 284 matching lines...) Expand 10 before | Expand all | Expand 10 after
1207 // request is now processed. 1214 // request is now processed.
1208 _subscription.resume(); 1215 _subscription.resume();
1209 } else { 1216 } else {
1210 // Close socket, keep-alive not used or body sent before 1217 // Close socket, keep-alive not used or body sent before
1211 // received data was handled. 1218 // received data was handled.
1212 destroy(); 1219 destroy();
1213 } 1220 }
1214 }) 1221 })
1215 .catchError((e) { 1222 .catchError((e) {
1216 destroy(); 1223 destroy();
1217 if (e.error is HttpException) throw e; 1224 throw e;
1218 // TODO(ajohnsen): Where to send Socket errors?
1219 }); 1225 });
1220 }); 1226 });
1221 response._ignoreBody = request.method == "HEAD"; 1227 response._ignoreBody = request.method == "HEAD";
1222 response._httpRequest = request; 1228 response._httpRequest = request;
1223 _httpServer._handleRequest(request); 1229 _httpServer._handleRequest(request);
1224 }, 1230 },
1225 onDone: () { 1231 onDone: () {
1226 destroy(); 1232 destroy();
1227 }, 1233 },
1228 onError: (error) { 1234 onError: (error) {
(...skipping 372 matching lines...) Expand 10 before | Expand all | Expand 10 after
1601 1607
1602 1608
1603 class _RedirectInfo implements RedirectInfo { 1609 class _RedirectInfo implements RedirectInfo {
1604 const _RedirectInfo(int this.statusCode, 1610 const _RedirectInfo(int this.statusCode,
1605 String this.method, 1611 String this.method,
1606 Uri this.location); 1612 Uri this.location);
1607 final int statusCode; 1613 final int statusCode;
1608 final String method; 1614 final String method;
1609 final Uri location; 1615 final Uri location;
1610 } 1616 }
OLDNEW
« no previous file with comments | « no previous file | sdk/lib/io/io_stream_consumer.dart » ('j') | tests/standalone/io/http_server_response_test.dart » ('J')

Powered by Google App Engine
This is Rietveld 408576698