=== modified file 'twisted/web2/channel/http.py'
--- twisted/web2/channel/http.py	2007-01-01 23:41:33 +0000
+++ twisted/web2/channel/http.py	2008-03-19 16:36:16 +0000
@@ -10,7 +10,7 @@
 from twisted.web2 import http_headers
 from twisted.web2 import http
 
-PERSIST_NO_PIPELINE, PERSIST_PIPELINE = (1,2)
+PERSIST_NO_PIPELINE, PERSIST_PIPELINE = (False, True)
 
 _cachedHostNames = {}
 def _cachedGetHostByAddr(hostaddr):

=== modified file 'twisted/web2/client/http.py'
--- twisted/web2/client/http.py	2007-10-19 08:16:52 +0000
+++ twisted/web2/client/http.py	2008-03-19 16:36:16 +0000
@@ -9,6 +9,7 @@
 from zope.interface import implements
 
 from twisted.internet.defer import Deferred
+from twisted.internet import error
 from twisted.protocols.basic import LineReceiver
 from twisted.protocols.policies import TimeoutMixin
 
@@ -152,8 +153,8 @@
         the C{responseDefer} if no response has been sent yet, or close the
         stream.
         """
-        self.abortParse()
         if hasattr(self, 'stream') and self.stream is not None:
+            self.abortParse()
             self.stream.finish(err)
         else:
             self.responseDefer.errback(err)
@@ -314,7 +315,7 @@
 
         self.manager.clientBusy(self)
         if closeAfter:
-            self.readPersistent = False
+            self.readPersistent = PERSIST_NO_PIPELINE
 
         self.outRequest = chanRequest = HTTPClientChannelRequest(self,
                                             request, closeAfter)
@@ -329,7 +330,7 @@
         self.outRequest = None
         # Tell the manager if more requests can be submitted.
         self.setTimeout(self.inputTimeOut)
-        if self.readPersistent is PERSIST_PIPELINE:
+        if self.readPersistent == PERSIST_PIPELINE:
             self.manager.clientPipelining(self)
 
     def requestReadFinished(self, request):
@@ -339,7 +340,7 @@
         self.firstLine = True
 
         if not self.inRequests:
-            if self.readPersistent:
+            if self.readPersistent == PERSIST_PIPELINE:
                 self.setTimeout(None)
                 self.manager.clientIdle(self)
             else:
@@ -347,14 +348,14 @@
 
     def setReadPersistent(self, persist):
         self.readPersistent = persist
-        if not persist:
+        if persist == PERSIST_NO_PIPELINE:
             # Tell all requests but first to abort.
             for request in self.inRequests[1:]:
-                request.connectionLost(None)
+                request.connectionLost(error.ConnectionDone())
             del self.inRequests[1:]
 
     def connectionLost(self, reason):
-        self.readPersistent = False
+        self.readPersistent = PERSIST_NO_PIPELINE
         self.setTimeout(None)
         self.manager.clientGone(self)
         # Tell all requests to abort.

=== modified file 'twisted/web2/test/test_client.py'
--- twisted/web2/test/test_client.py	2007-10-19 08:16:52 +0000
+++ twisted/web2/test/test_client.py	2008-03-19 16:29:47 +0000
@@ -5,7 +5,7 @@
 Tests for HTTP client.
 """
 
-from twisted.internet import protocol, defer
+from twisted.internet import protocol, defer, error
 
 from twisted.web2.client import http
 from twisted.web2 import http_headers
@@ -13,7 +13,6 @@
 from twisted.web2.test.test_http import LoopbackRelay, HTTPTests, TestConnection
 
 
-
 class TestServer(protocol.Protocol):
     data = ""
     done = False
@@ -61,8 +60,10 @@
     def assertReceived(self, cxn, expectedStatus, expectedHeaders,
                        expectedContent=None):
         self.iterate(cxn)
-
-        headers, content = cxn.server.data.split('\r\n\r\n', 1)
+    
+        data = cxn.server.data
+        cxn.server.data = ''
+        headers, content = data.split('\r\n\r\n', 1)
         status, headers = headers.split('\r\n', 1)
         headers = headers.split('\r\n')
 
@@ -74,6 +75,7 @@
         for x in headers:
             self.assertIn(x, expectedHeaders)
 
+
         if not expectedContent:
             expectedContent = ''
 
@@ -152,7 +154,6 @@
 
         d = cxn.client.submitRequest(req).addCallback(gotResp)
 
-
         self.assertReceived(cxn, 'GET / HTTP/1.1',
                             ['Connection: close'])
 
@@ -188,6 +189,60 @@
 
         return d
 
+    def test_abortPipelining(self):
+        """
+        Ensure that pending requests are aborted if the server closes the
+        connection.
+        """
+        cxn = self.connect(inputTimeOut=None)
+        deferreds = []
+        
+        req1 = http.ClientRequest('GET', '/', None, None)
+        d1 = cxn.client.submitRequest(req1, closeAfter=False).addCallback(
+            self.checkResponse, 200, [], 10, None)
+        deferreds.append(d1)
+
+        self.assertReceived(cxn, 'GET / HTTP/1.1',
+                            ['Connection: Keep-Alive'])
+
+        self.writeLines(cxn, ('HTTP/1.1 200 OK',
+                              'Content-Length: 10',
+                              'Connection: Keep-Alive',
+                              '',
+                              '1234567890'))
+
+        # pipeline two requests, after the first one the server closes the
+        # connection
+        req2 = http.ClientRequest('GET', '/', None, None)
+        d2 = cxn.client.submitRequest(req2, closeAfter=False).addCallback(
+            self.checkResponse, 200, [], 0, None)
+        deferreds.append(d2)
+        
+        self.assertReceived(cxn, 'GET / HTTP/1.1',
+                            ['Connection: Keep-Alive'])
+
+        def requestEb(failure):
+            failure.trap(error.ConnectionDone)
+
+        def requestCb(result):
+            self.fail('this should not be reached')
+
+        req3 = http.ClientRequest('GET', '/', None, None)
+        d3 = cxn.client.submitRequest(req3, closeAfter=False)
+        d3.addCallbacks(requestCb, requestEb)
+        deferreds.append(d3)
+
+        self.assertReceived(cxn, 'GET / HTTP/1.1',
+                            ['Connection: Keep-Alive'])
+       
+        # close the connection, the client should cancel req3
+        self.writeLines(cxn, ('HTTP/1.1 200 OK',
+                              'Content-Length: 0',
+                              'Connection: close',
+                              '\r\n'))
+
+        return defer.DeferredList(deferreds)
+
     def test_userHeaders(self):
         """
         Make sure that headers get through in both directions.