Index: twisted/web/test/test_newclient.py =================================================================== --- twisted/web/test/test_newclient.py (revision 32500) +++ twisted/web/test/test_newclient.py (working copy) @@ -16,6 +16,7 @@ from twisted.internet.error import ConnectionDone from twisted.internet.defer import Deferred, succeed, fail from twisted.internet.protocol import Protocol +from twisted.internet import reactor from twisted.trial.unittest import TestCase from twisted.test.proto_helpers import StringTransport, AccumulatingProtocol from twisted.web._newclient import UNKNOWN_LENGTH, STATUS, HEADER, BODY, DONE @@ -27,6 +28,7 @@ from twisted.web._newclient import ConnectionAborted from twisted.web._newclient import BadHeaders, ResponseDone, PotentialDataLoss, ExcessWrite from twisted.web._newclient import TransportProxyProducer, LengthEnforcingConsumer, makeStatefulDispatcher +from twisted.web._newclient import TIMEOUT_100_CONTINUE from twisted.web.http_headers import Headers from twisted.web.http import _DataLoss from twisted.web.iweb import IBodyProducer, IResponse @@ -793,6 +795,7 @@ """ method = 'GET' stopped = False + headers = Headers() def writeTo(self, transport): self.finished = Deferred() @@ -811,12 +814,32 @@ returns a succeeded L{Deferred}. This vaguely emulates the behavior of a L{Request} with no body producer. """ + headers = Headers() def writeTo(self, transport): transport.write('SOME BYTES') return succeed(None) +class BufferProducer(object): + """ + A body producer that outputs a string with a known length. + """ + implements(IBodyProducer) + def __init__(self, data): + self.data = data + self.length = len(data) + + def startProducing(self, consumer): + consumer.write(self.data) + return succeed(None) + + def pauseProducing(self): + pass + + def stopProducing(self): + pass + class HTTP11ClientProtocolTests(TestCase): """ Tests for the HTTP 1.1 client protocol implementation, @@ -879,6 +902,7 @@ L{RequestGenerationFailed} wrapping the underlying failure. """ class BrokenRequest: + headers = Headers() def writeTo(self, transport): return fail(ArbitraryException()) @@ -901,6 +925,7 @@ a L{Failure} of L{RequestGenerationFailed} wrapping that exception. """ class BrokenRequest: + headers = Headers() def writeTo(self, transport): raise ArbitraryException() @@ -1312,8 +1337,119 @@ [ConnectionAborted, _DataLoss]) return deferred.addCallback(checkError) + def test_expect100ContinueGot100Continue(self): + """ + When we expect 100-Continue and we receive an 100-Continue L{Response} + the user writes the L{Request} body and receives back the second(final) + L{Response}. + """ + producer = BufferProducer('X' * 10) + headers = Headers({'host': ['example.com'], + 'expect': ['100-Continue']}) + d = self.protocol.request(Request('POST', '/foo', headers, producer)) + + self.transport.clear() + + def cb100Continue(response): + self.assertEqual(response.code, 200) + + d.addCallback(cb100Continue) + + self.protocol.dataReceived( + "HTTP/1.1 100 Continue\r\n" + "Content-Length: 3\r\n" + "\r\n" + '123' + ) + + self.protocol.dataReceived( + "HTTP/1.1 200 OK\r\n" + "Content-Length: 0\r\n" + "\r\n" + ) + + self.assertEqual(self.transport.value(), 'X' * 10) + + return d + + + def test_expect100ContinueGotFinalStatus(self): + """ + When we expect 100-Continue and we receive a final status L{Response} + we don't write the L{Request} body anymore and the user receives the + first L{Response}. + """ + producer = BufferProducer('X' * 10) + + headers = Headers({'host': ['example.com'], 'expect': ['100-Continue']}) + + d = self.protocol.request(Request('POST', '/foo', headers, producer)) + + self.transport.clear() + + def cb100Continue(response): + self.assertEqual(response.code, 200) + + d.addCallback(cb100Continue) + + + self.protocol.dataReceived( + "HTTP/1.1 200 OK\r\n" + "Content-Length: 0\r\n" + "\r\n" + ) + + self.assertEqual(self.transport.value(), '') + + return d + + def test_expect100ContinueServerNotSupporting(self): + """ + When we expect 100-continue and the server doesn't respond with a + response for upto TIMEOUT_100_CONTINUE seconds (maybe because it does + not correctly handle this expectation) we continue with sending the + request body. + """ + producer = BufferProducer('X' * 10) + + headers = Headers({'host': ['example.com'], 'expect': ['100-Continue']}) + + d = self.protocol.request(Request('POST', '/foo', headers, producer)) + + self.transport.clear() + + def cbResponse(response): + self.assertEqual(response.code, 200) + + d.addCallback(cbResponse) + + d2 = Deferred() + + d.chainDeferred(d2) + + def laterCall(): + try: + self.assertEqual(self.transport.value(), 'X'*10) + except Exception, e: + d2.errback(e) + return + + self.protocol.dataReceived( + "HTTP/1.1 200 OK\r\n" + "Content-Length: 0\r\n" + "\r\n" + ) + + reactor.callLater(TIMEOUT_100_CONTINUE + 1.0, laterCall) + + self.assertEqual(self.transport.value(), '') + + return d2 + + + class StringProducer: """ L{StringProducer} is a dummy body producer. Index: twisted/web/_newclient.py =================================================================== --- twisted/web/_newclient.py (revision 32500) +++ twisted/web/_newclient.py (working copy) @@ -37,8 +37,10 @@ from twisted.internet.interfaces import IConsumer, IPushProducer from twisted.internet.error import ConnectionDone from twisted.internet.defer import Deferred, succeed, fail, maybeDeferred +from twisted.internet import reactor from twisted.internet.protocol import Protocol from twisted.protocols.basic import LineReceiver +from twisted.protocols.wire import Discard from twisted.web.iweb import UNKNOWN_LENGTH, IResponse from twisted.web.http_headers import Headers from twisted.web.http import NO_CONTENT, NOT_MODIFIED @@ -51,7 +53,10 @@ BODY = 'BODY' DONE = 'DONE' +# Interval to wait for a Response to a Request with 'Expect: 100-Continue' +TIMEOUT_100_CONTINUE = 1.0 + class BadHeaders(Exception): """ Headers passed to L{Request} were in some way invalid. @@ -516,8 +521,11 @@ self._responseDeferred.errback(Failure(ResponseFailed([reason]))) del self._responseDeferred + def _hasResponse(self): + return hasattr(self, 'response') + class Request: """ A L{Request} instance describes an HTTP request to be sent to an HTTP @@ -566,12 +574,11 @@ transport.writeSequence(requestLines) - def _writeToChunked(self, transport): + def _writeBodyToChunked(self, transport): """ - Write this request to the given transport using chunked + Write this request's body to the given transport using chunked transfer-encoding to frame the body. """ - self._writeHeaders(transport, 'Transfer-Encoding: chunked\r\n') encoder = ChunkedEncoder(transport) encoder.registerProducer(self.bodyProducer, True) d = self.bodyProducer.startProducing(encoder) @@ -590,14 +597,11 @@ return d - def _writeToContentLength(self, transport): + def _writeBodyToContentLength(self, transport): """ - Write this request to the given transport using content-length to frame - the body. + Write this request's body to the given transport using content-length + to frame the body. """ - self._writeHeaders( - transport, - 'Content-Length: %d\r\n' % (self.bodyProducer.length,)) # This Deferred is used to signal an error in the data written to the # encoder below. It can only errback and it will only do so before too @@ -709,16 +713,43 @@ been completely written to the transport or with a L{Failure} if there is any problem generating the request bytes. """ + + self._writeHeadersTo(transport) + return self._writeBodyTo(transport) + + def _writeHeadersTo(self, transport): + """ + Format this L{Request}'s headers as HTTP/1.1 and write them + synchronously to the given transport + """ + TEorCL = None if self.bodyProducer is not None: if self.bodyProducer.length is UNKNOWN_LENGTH: - return self._writeToChunked(transport) + TEorCL = "Transfer-Encoding: chunked\r\n" else: - return self._writeToContentLength(transport) + TEorCL = 'Content-Length: %d\r\n' % (self.bodyProducer.length,) + + self._writeHeaders(transport, TEorCL) + + def _writeBodyTo(self, transport): + """ + Write this L{Request}'s body to the given transport, framing it + according to the given headers('Transport-Encoding' or + 'Content-Length'). + + @return: A L{Deferred} which fires with C{None} when the request has + been completely written the transport or with a L{Failure} if there + is any problem generating the request body bytes. If bodyProducer + is None the returned L{Deferred} is already fired. + """ + if self.bodyProducer is not None: + if self.bodyProducer.length is UNKNOWN_LENGTH: + return self._writeBodyToChunked(transport) + else: + return self._writeBodyToContentLength(transport) else: - self._writeHeaders(transport, None) return succeed(None) - def stopWriting(self): """ Stop writing this request to the transport. This can only be called @@ -1158,7 +1189,29 @@ self._producer.pauseProducing() +class DiscardWithDeferred(Protocol): + """ + A L{Protocol} that discards all received data and that fires a L{Deferred} + when all data has been received. + @ivar finishedDeferred: L{Deferred} which fires with C{None} when all data + has been received and with L{Failure} on error. + + """ + + def __init__(self): + self.finishedDeferred = Deferred() + + def dataReceived(self, data): + pass + + def connectionLost(self, reason): + if reason.type == ResponseDone: + self.finishedDeferred.callback(None) + else: + self.finishedDeferred.errback(reason) + + class HTTP11ClientProtocol(Protocol): """ L{HTTP11ClientProtocol} is an implementation of the HTTP 1.1 client @@ -1243,18 +1296,103 @@ return fail(RequestNotSent()) self._state = 'TRANSMITTING' - _requestDeferred = maybeDeferred(request.writeTo, self.transport) - self._finishedRequest = Deferred() # Keep track of the Request object in case we need to call stopWriting # on it. self._currentRequest = request - self._transportProxy = TransportProxyProducer(self.transport) - self._parser = HTTPClientParser(request, self._finishResponse) - self._parser.makeConnection(self._transportProxy) - self._responseDeferred = self._parser._responseDeferred + self._finishedRequest = Deferred() + if request.headers.hasHeader('expect'): + expectations = request.headers.getRawHeaders('expect') + _expects100Continue = '100-continue' in [x.lower() for x in + expectations] + else: + _expects100Continue = False + + if _expects100Continue: + # This is synchronous, so jump right into WAITING + request._writeHeadersTo(self.transport) + + self._state = 'WAITING' + + self._setupParser(request) + + _100ContinueResponseDeferred = self._parser._responseDeferred + + self._responseDeferred = Deferred() + + _requestDeferred = Deferred() + + def cb100ContinueBody(ignored): + # Start sending the request body + + self._state = 'TRANSMITTING' + + self._disconnectParser(None) + + self._setupParser(request) + + _requestBodyDeferred = maybeDeferred( + request._writeBodyTo, self.transport) + + _requestBodyDeferred.chainDeferred(_requestDeferred) + + self._parser._responseDeferred.chainDeferred( + self._responseDeferred) + + def brokenServerTimer(): + # In case the parser hasn't received any data yet, send + # ourselves a fake 100-Continue response. + + if not self._parser._hasResponse(): + self.dataReceived( + "HTTP/1.1 100 Continue\r\n" + "Content-length: 0\r\n" + "\r\n") + + _brokenServerDelayed = reactor.callLater(TIMEOUT_100_CONTINUE, + brokenServerTimer) + + def ebFirstResponse(err): + if _brokenServerDelayed.active(): + _brokenServerDelayed.cancel() + + # Tell the producer we don't need its body and forward the + # error + request.stopWriting() + self._finishedRequest.errback(err) + + def cbFirstResponse(response): + if _brokenServerDelayed.active(): + _brokenServerDelayed.cancel() + + if response.code == 100: + # Read and discard the response body + + discarder = DiscardWithDeferred() + + response.deliverBody(discarder) + + discarder.finishedDeferred.addCallbacks(cb100ContinueBody, + ebFirstResponse) + + else: + # Tell the producer we don't need its body and forward the + # response + request.stopWriting() + self._finishedRequest.callback(response) + + _100ContinueResponseDeferred.addCallbacks( + cbFirstResponse, ebFirstResponse) + + else: + _requestDeferred = maybeDeferred(request.writeTo, self.transport) + + self._setupParser(request) + + self._responseDeferred = self._parser._responseDeferred + def cbRequestWrotten(ignored): if self._state == 'TRANSMITTING': self._state = 'WAITING' @@ -1279,6 +1417,19 @@ return self._finishedRequest + def _setupParser(self, request): + """ + Setup a L{HTTPClientParser} for a L{Response} to a L{Request}. If this + is not the first parser associated with this protocol, call + L{HTTP11ClientProtocol._disconnectParser} first. + + @param request: L{Request} waiting for a L{Response} + """ + self._transportProxy = TransportProxyProducer(self.transport) + self._parser = HTTPClientParser(request, self._finishResponse) + self._parser.makeConnection(self._transportProxy) + + def _finishResponse(self, rest): """ Called by an L{HTTPClientParser} to indicate that it has parsed a