| OLD | NEW |
| 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 extends Stream<List<int>> { | 7 class _HttpIncoming extends Stream<List<int>> { |
| 8 final int _transferLength; | 8 final int _transferLength; |
| 9 final Completer _dataCompleter = new Completer(); | 9 final Completer _dataCompleter = new Completer(); |
| 10 Stream<List<int>> _stream; | 10 Stream<List<int>> _stream; |
| (...skipping 524 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 535 void _writeHeader(); // TODO(ajohnsen): Better name. | 535 void _writeHeader(); // TODO(ajohnsen): Better name. |
| 536 } | 536 } |
| 537 | 537 |
| 538 | 538 |
| 539 class _HttpOutboundConsumer implements StreamConsumer { | 539 class _HttpOutboundConsumer implements StreamConsumer { |
| 540 final _HttpOutboundMessage _outbound; | 540 final _HttpOutboundMessage _outbound; |
| 541 StreamController _controller; | 541 StreamController _controller; |
| 542 StreamSubscription _subscription; | 542 StreamSubscription _subscription; |
| 543 Completer _closeCompleter = new Completer(); | 543 Completer _closeCompleter = new Completer(); |
| 544 Completer _completer; | 544 Completer _completer; |
| 545 bool _socketError = false; |
| 545 | 546 |
| 546 _HttpOutboundConsumer(_HttpOutboundMessage this._outbound); | 547 _HttpOutboundConsumer(_HttpOutboundMessage this._outbound); |
| 547 | 548 |
| 548 void _onPause() { | 549 void _cancel() { |
| 549 if (_controller.isPaused) { | 550 if (_subscription != null) { |
| 550 _subscription.pause(); | |
| 551 } else { | |
| 552 _subscription.resume(); | |
| 553 } | |
| 554 } | |
| 555 | |
| 556 void _onListen() { | |
| 557 if (!_controller.hasListener && _subscription != null) { | |
| 558 _subscription.cancel(); | 551 _subscription.cancel(); |
| 559 } | 552 } |
| 560 } | 553 } |
| 561 | 554 |
| 562 _ensureController() { | 555 _ensureController() { |
| 563 if (_controller != null) return; | 556 if (_controller != null) return; |
| 564 _controller = new StreamController(onPause: _onPause, | 557 _controller = new StreamController(onPause: () => _subscription.pause(), |
| 565 onResume: _onPause, | 558 onResume: () => _subscription.resume(), |
| 566 onListen: _onListen, | 559 onCancel: _cancel); |
| 567 onCancel: _onListen); | |
| 568 _outbound._addStream(_controller.stream) | 560 _outbound._addStream(_controller.stream) |
| 569 .then((_) { | 561 .then((_) { |
| 570 _onListen(); // Make sure we unsubscribe. | 562 _cancel(); |
| 571 _done(); | 563 _done(); |
| 572 _closeCompleter.complete(_outbound); | 564 _closeCompleter.complete(_outbound); |
| 573 }, | 565 }, |
| 574 onError: (error) { | 566 onError: (error) { |
| 575 if (!_done(error)) { | 567 _socketError = true; |
| 576 _closeCompleter.completeError(error); | 568 if (error is SocketIOException && |
| 569 _outbound is HttpResponse) { |
| 570 _cancel(); |
| 571 _done(); |
| 572 _closeCompleter.complete(_outbound); |
| 573 } else { |
| 574 if (!_done(error)) { |
| 575 _closeCompleter.completeError(error); |
| 576 } |
| 577 } | 577 } |
| 578 }); | 578 }); |
| 579 } | 579 } |
| 580 | 580 |
| 581 bool _done([error]) { | 581 bool _done([error]) { |
| 582 if (_completer == null) return false; | 582 if (_completer == null) return false; |
| 583 if (error != null) { | 583 if (error != null) { |
| 584 _completer.completeError(error); | 584 _completer.completeError(error); |
| 585 } else { | 585 } else { |
| 586 _completer.complete(_outbound); | 586 _completer.complete(_outbound); |
| 587 } | 587 } |
| 588 _completer = null; | 588 _completer = null; |
| 589 return true; | 589 return true; |
| 590 } | 590 } |
| 591 | 591 |
| 592 Future addStream(var stream) { | 592 Future addStream(var stream) { |
| 593 _ensureController(); | 593 // If we saw a socket error subscribe and then cancel, to ignore any data |
| 594 // on the stream. |
| 595 if (_socketError) { |
| 596 stream.listen(null).cancel(); |
| 597 return new Future.value(_outbound); |
| 598 } |
| 594 _completer = new Completer(); | 599 _completer = new Completer(); |
| 595 _subscription = stream.listen( | 600 _subscription = stream.listen( |
| 596 (data) { | 601 (data) { |
| 597 _controller.add(data); | 602 _controller.add(data); |
| 598 }, | 603 }, |
| 599 onDone: () { | 604 onDone: () { |
| 600 _done(); | 605 _done(); |
| 601 }, | 606 }, |
| 602 onError: (error) { | 607 onError: (error) { |
| 603 _done(error); | 608 _done(error); |
| 604 }, | 609 }, |
| 605 cancelOnError: true); | 610 cancelOnError: true); |
| 611 _ensureController(); |
| 606 return _completer.future; | 612 return _completer.future; |
| 607 } | 613 } |
| 608 | 614 |
| 609 Future close() { | 615 Future close() { |
| 610 Future closeOutbound() { | 616 Future closeOutbound() { |
| 617 if (_socketError) return new Future.value(_outbound); |
| 611 return _outbound._close().then((_) => _outbound); | 618 return _outbound._close().then((_) => _outbound); |
| 612 } | 619 } |
| 613 if (_controller == null) return closeOutbound(); | 620 if (_controller == null) return closeOutbound(); |
| 614 _controller.close(); | 621 _controller.close(); |
| 615 return _closeCompleter.future.then((_) => closeOutbound()); | 622 return _closeCompleter.future.then((_) => closeOutbound()); |
| 616 } | 623 } |
| 617 } | 624 } |
| 618 | 625 |
| 619 | 626 |
| 620 class _BufferTransformer extends StreamEventTransformer<List<int>, List<int>> { | 627 class _BufferTransformer extends StreamEventTransformer<List<int>, List<int>> { |
| (...skipping 1131 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 1752 destroy(); | 1759 destroy(); |
| 1753 }); | 1760 }); |
| 1754 response._ignoreBody = request.method == "HEAD"; | 1761 response._ignoreBody = request.method == "HEAD"; |
| 1755 response._httpRequest = request; | 1762 response._httpRequest = request; |
| 1756 _httpServer._handleRequest(request); | 1763 _httpServer._handleRequest(request); |
| 1757 }, | 1764 }, |
| 1758 onDone: () { | 1765 onDone: () { |
| 1759 destroy(); | 1766 destroy(); |
| 1760 }, | 1767 }, |
| 1761 onError: (error) { | 1768 onError: (error) { |
| 1762 _httpServer._handleError(error); | 1769 // Ignore failed requests that was closed before headers was received. |
| 1763 destroy(); | 1770 destroy(); |
| 1764 }); | 1771 }); |
| 1765 } | 1772 } |
| 1766 | 1773 |
| 1767 void destroy() { | 1774 void destroy() { |
| 1768 if (_state == _CLOSING || _state == _DETACHED) return; | 1775 if (_state == _CLOSING || _state == _DETACHED) return; |
| 1769 _state = _CLOSING; | 1776 _state = _CLOSING; |
| 1770 _socket.destroy(); | 1777 _socket.destroy(); |
| 1771 _httpServer._connectionClosed(this); | 1778 _httpServer._connectionClosed(this); |
| 1772 } | 1779 } |
| (...skipping 539 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 2312 | 2319 |
| 2313 | 2320 |
| 2314 class _RedirectInfo implements RedirectInfo { | 2321 class _RedirectInfo implements RedirectInfo { |
| 2315 const _RedirectInfo(int this.statusCode, | 2322 const _RedirectInfo(int this.statusCode, |
| 2316 String this.method, | 2323 String this.method, |
| 2317 Uri this.location); | 2324 Uri this.location); |
| 2318 final int statusCode; | 2325 final int statusCode; |
| 2319 final String method; | 2326 final String method; |
| 2320 final Uri location; | 2327 final Uri location; |
| 2321 } | 2328 } |
| OLD | NEW |