=== added directory 'elisa-plugins/elisa/plugins/http_plugin'
=== added file 'elisa-plugins/elisa/plugins/http_plugin/__init__.py'
=== added file 'elisa-plugins/elisa/plugins/http_plugin/http_media.py'
--- elisa-plugins/elisa/plugins/http_plugin/http_media.py	1970-01-01 00:00:00 +0000
+++ elisa-plugins/elisa/plugins/http_plugin/http_media.py	2008-03-19 10:44:57 +0000
@@ -0,0 +1,476 @@
+# -*- coding: utf-8 -*-
+# Elisa - Home multimedia server
+# Copyright (C) 2007 Fluendo Embedded S.L. (www.fluendo.com).
+# All rights reserved.
+#
+# This file is available under one of two license agreements.
+#
+# This file is licensed under the GPL version 2.
+# See "LICENSE.GPL" in the root of this distribution including a special
+# exception to use Elisa with Fluendo's plugins.
+#
+# The GPL part of Elisa is also available under a commercial licensing
+# agreement from Fluendo.
+# See "LICENSE.Elisa" in the root directory of this distribution package
+# for details on that license.
+
+__maintainer__ = 'Alessandro Decina <alessandro@fluendo.com>'
+
+from threading import Lock
+from twisted.internet import reactor, defer, error
+from twisted.internet.protocol import ClientFactory, ReconnectingClientFactory
+from twisted.web2.client.http import HTTPClientProtocol, ClientRequest
+from twisted.web2.channel.http import PERSIST_PIPELINE, PERSIST_NO_PIPELINE
+from twisted.web2.http import responsecode
+from twisted.web2.stream import BufferedStream
+from elisa.base_components.media_provider import MediaProvider
+from elisa.core.log import Loggable
+from elisa.core.media_uri import MediaUri
+from elisa.core.media_file import MediaFile
+from threading import Lock
+
+class ElisaHttpClientProtocol(HTTPClientProtocol):
+    def __init__(self, client):
+        HTTPClientProtocol.__init__(self)
+        self.client = client
+
+    def connectionMade(self):
+        self.client.connectionMade(self)
+
+    def requestWriteFinished(self, request):
+        HTTPClientProtocol.requestWriteFinished(self, request)
+        self.client.requestWriteFinished(request)
+
+    # we don't use requestReadFinished here because it's called even when a read
+    # is still going on and the connection is lost
+
+class ElisaHttpClientFactory(ReconnectingClientFactory):
+    # after 3 retries we give up
+    maxRetries = 3
+
+    def __init__(self, client):
+        self.client = client
+
+        # base has no init
+
+    def buildProtocol(self, addr):
+#        self.resetDelay()
+        protocol = ElisaHttpClientProtocol(self.client)
+        
+        # FIXME: check the spec, probably we shouldn't call this and wait for a
+        # Connection: Keep-Alive header from the server
+        persist = None
+        if self.client.pipeline:
+            persist = PERSIST_PIPELINE
+        else:
+            persist = PERSIST_NO_PIPELINE
+        protocol.setReadPersistent(persist)
+
+        return protocol
+
+    def startedConnecting(self, connector):
+        # log me
+        pass
+
+    def clientConnectionFailed(self, connector, reason):
+        self.client.connectionFailed(connector, reason)
+        # we don't want to reconnect in this case
+
+    def clientConnectionLost(self, connector, reason):
+        # let the client requeue its requests
+        if self.retries == self.maxRetries:
+            self.client.connectionLostForever(connector, reason)
+        else:
+            if not self.client.connectionLost(connector, reason):
+                self.stopTrying()
+        if self.retries == 0:
+            # hack: most of the time the connection here is lost because we are
+            # not using keep-alive, so we want to reconnect immediately but
+            # there is no way of doing it with
+            # ReconnectingClientFactory.clientConnectionLost
+            self.noisy = False
+
+            if not self.continueTrying:
+                if self.noisy:
+                    log.msg("Abandoning %s on explicit request" % (connector,))
+                return
+
+            self.connector = connector
+            self.retries += 1
+            def reconnector():
+                self._callID = None
+                connector.connect()
+            self._callID = reactor.callLater(0, reconnector)
+        else:
+            ReconnectingClientFactory.clientConnectionLost(self, connector, reason)
+
+class ElisaHttpClientNotOpenedError(Exception):
+    pass
+
+class ElisaHttpClientNotPipeliningError(Exception):
+    pass
+
+class ElisaHttpClient(Loggable):
+    """twisted.web2 based HTTP client."""
+
+    def __init__(self, pipeline=True):
+        super(ElisaHttpClient, self).__init__()
+
+        self._host = None
+        self._port = None
+        self.pipeline = pipeline
+        self._queued_requests = []
+        self._written_requests = []
+        self._closed = True
+        self._reset()
+
+    def _reset(self):
+        self._connector = None
+        self._protocol = None
+        self._open_defer = defer.Deferred()
+        self._close_defer = defer.Deferred()
+
+    def open(self, host, port=80):
+        """Open an HTTP connection.
+
+        @param host:        hostname or ip address of the server
+        @type host:         str
+        @param port:        tcp port
+        @type port:         int
+
+        """
+        self._host = host
+        self._port = port
+        self._closed = False
+        self._connector = reactor.connectTCP(host, port,
+                ElisaHttpClientFactory(self))
+
+        return self._open_defer
+
+    def request(self, uri, method='GET'):
+        return self.request_full(ClientRequest(method, uri, {}, None))
+
+    def request_full(self, request):
+        """Send an HTTP request.
+
+        @param request:     request to submit
+        @type request:      L{twisted.web2.client.http.ClientRequest}
+        @rtype:             L{twisted.internet.defer.Deferred}
+        """
+
+        if self._closed:
+            return defer.fail(ElisaHttpClientNotOpenedError())
+
+        # set Host: if it isn't already set
+        if not request.headers.getHeader('Host'):
+            request.headers.setHeader('Host', self._host)
+
+        return self._queue_request(request)
+
+    def close(self):
+        """Close an HTTP connection"""
+        if self._closed or not self._connector:
+            # come on you haven't even called open
+            return defer.fail(ElisaHttpClientNotOpenedError())
+
+        self._closed = True
+
+        if self._connector.state != 'disconnected':
+            self._connector.disconnect()
+
+        return self._close_defer
+
+    def _queue_request(self, request):
+        res = None
+        busy = False
+        if self.pipeline:
+            busy = bool(self._queued_requests)
+            request.defer = defer.Deferred()
+            self._queued_requests.append(request)
+        else:
+            busy = bool(self._queued_requests or self._written_requests)
+            if busy:
+                res = defer.fail(ElisaHttpClientNotPipeliningError())
+            else:
+                request.defer = defer.Deferred()
+                self._queued_requests.append(request)
+
+        if res:
+            # client busy and not pipelining
+            return res
+
+        if not busy:
+            self._submit_next_request()
+
+        return request.defer
+
+    def _submit_next_request(self):
+        try:
+            request = self._queued_requests[0]
+        except IndexError:
+            # no more requests
+            return
+
+        def request_failure(failure, request):
+            # when pipelining the server might decide to close the connection,
+            # we want to retry in that case
+            failure.trap(error.ConnectionDone)
+            self.debug('request %s failed: %s. (going to be resubmitted)',
+                request, failure)
+
+        request_dfr = self._protocol.submitRequest(request,
+                closeAfter=not self.pipeline)
+        request_dfr.addCallback(self._request_done, request)
+        # FIXME: we now chain the request defer here, so client code is called
+        # when this request is still being processed, which make some sense.  On
+        # the other hand, if a client wants to submit another request in this
+        # callback and it isn't pipelining, it must do it in
+        # reactor.callLater(0, do_next_request) not to get a NotPipeliningError
+        request_dfr.addCallback(request.defer.callback)
+        request_dfr.addErrback(request_failure, request)
+
+    ### callbacks called by our ElisaHttpClientFactory
+    def connectionFailed(self, connector, reason):
+        open_defer, self._open_defer = self._open_defer, defer.Deferred()
+        open_defer.errback(reason)
+
+    def connectionLost(self, connector, reason):
+        if self._closed:
+            # FIXME: this part is hackish but I need a way to let the caller
+            # re-enter and finish its work otherwise after a .close() on the
+            # client the reactor is still in a dirty state and trial complains
+            # about that
+            reactor.callLater(0, self._close_defer.callback, None)
+            return False
+
+        # the factory is going to retry a connection, requeue unfinished
+        # requests
+        self._protocol = None
+        requests = []
+        for request in self._written_requests + self._queued_requests:
+            newreq = ClientRequest(request.method, request.uri,
+                    request.headers, request.stream)
+            newreq.defer = request.defer
+            requests.append(newreq)
+        self._written_requests = []
+        self._queued_requests = requests
+
+        return not self._closed
+
+    def connectionLostForever(self, connector, reason):
+        self._protocol = None
+        written_requests, self._written_requests = \
+                self._written_requests, []
+        queued_requests, self._queued_requests = \
+                self._queued_requests, []
+    
+        # errback unlocked
+        for request in written_requests + queued_requests:
+            request.defer.errback(reason)
+        reactor.callLater(0, self._close_defer.callback, None)
+
+    ### callbacks called by our ElisaHttpClientProtocol
+    def connectionMade(self, protocol):
+        assert self._protocol is None
+        self._protocol = protocol
+
+        open_defer, self._open_defer = self._open_defer, defer.Deferred()
+        reactor.callLater(0, open_defer.callback, protocol)
+
+        self._submit_next_request()
+
+    def requestWriteFinished(self, req):
+        pipeline = False
+        request = self._queued_requests.pop(0)
+        assert request == req.request
+        self._written_requests.append(request)
+
+        if self.pipeline and self._queued_requests:
+            pipeline = True
+
+        if pipeline:
+            self._submit_next_request()
+
+    def _request_done(self, response, request):
+        assert request == self._written_requests.pop(0)
+
+        self._submit_next_request()
+
+        return response
+
+
+class HttpMediaFileDescriptor(object):
+    def __init__(self, uri):
+        self.uri = uri
+        self.stream = None
+        self.stream_dfr = None
+
+class HttpMediaClient(ElisaHttpClient):
+    """An ElisaHttpClient specialization used by the http media provider.
+    
+    open() and close() can be called multiple times on this object to simplify
+    connection sharing in the media provider.
+    """
+
+    def __init__(self):
+        super(HttpMediaClient, self).__init__()
+
+        self.use_count = 0
+        self.connected_dfr = None
+
+    def open(self, host, port=80):
+        self.use_count += 1
+        if self.use_count == 1:
+            self.connected_dfr = ElisaHttpClient.open(self, host, port)
+
+        return self.connected_dfr
+
+    def close(self):
+        self.use_count -= 1
+        if self.use_count <= 0:
+            def close_done(result):
+                # when the connection is closed return True in the deferred
+                return True
+
+            self.use_count = 0
+            dfr = ElisaHttpClient.close(self)
+            dfr.addCallback(close_done)
+            return dfr
+        else:
+            # return False in the deferred to mean that the connection has not
+            # been closed
+            return defer.succeed(False)
+
+class HttpMedia(MediaProvider):
+    def __init__(self):
+        self._clients = {}
+
+    def scannable_uri_schemes__get(self):
+        return []
+
+    def supported_uri_schemes__get(self):
+        # FIXME: set a good rank
+        return {'http': 1000}
+
+    def get_real_uri(self, uri):
+        return uri
+
+    def get_media_type(self, uri):
+        return defer.fail(NotImplementedError())
+
+    def is_directory(self, uri):
+        return False
+
+    # FIXME: open/close need testing
+    def _open_client(self, uri):
+        try:
+            client = self._clients[(uri.host, uri.port)]
+        except KeyError:
+            client = HttpMediaClient()
+            self._clients[(uri.host, uri.port)] = client
+
+        dfr = client.open(uri.host, uri.port or 80)
+        dfr.addCallback(lambda result: client)
+
+        return dfr
+
+    def _get_client(self, uri):
+        return self._clients[(uri.host, uri.port)]
+
+    def _close_client(self, uri):
+        client = self._get_client(uri)
+        dfr = client.close()
+
+        def close_done(last):
+            if last:
+                del self._clients[(uri.host, uri.port)]
+            
+            return last
+
+        dfr.addCallback(close_done)
+
+        return dfr
+
+    def _submit_next_request(self, method, uri):
+        def request_done(response):
+            if 300 <= response.code < 400:
+                # redirect, hmpf
+                # FIXME: handle redirect loops
+                location = response.headers.getHeader('Location')
+                new_uri = MediaUri(uri)
+                redir_uri = MediaUri(location)
+                new_uri.host = redir_uri.host
+                new_uri.path = redir_uri.path
+                self.debug('%d redirect from %s to %s',
+                        response.code, uri, new_uri)
+                return self._submit_next_request(method, new_uri)
+
+            return response
+
+        client = self._get_client(uri)
+        request_dfr = client.request(uri.path.encode('utf-8'), method)
+        request_dfr.addCallback(request_done)
+
+        return request_dfr
+
+    def open(self, uri, mode=None):
+        def open_client_done(client):
+            # do an HEAD request to see if the resource is there
+            dfr = self._submit_next_request('HEAD', uri)
+            dfr.addCallback(head_response)
+            # FIXME: the API doesn't say what I should return in case of failure
+
+            return dfr
+
+        def head_response(response):
+            descr = HttpMediaFileDescriptor(uri)
+            media_file = MediaFile(self, descr)
+
+            return media_file
+
+        dfr = self._open_client(uri)
+        dfr.addCallback(open_client_done)
+
+        return dfr
+
+    def close(self, media_file):
+        return self._close_client(media_file.descriptor.uri)
+
+    def seek(self, media_file, offset, whence=0):
+        return defer.fail(NotImplementedError())
+
+    def read(self, media_file, size=None):
+        read_dfr = defer.Deferred()
+
+        if size == -1:
+            size = None
+
+        if media_file.descriptor.stream_dfr is None:
+            def request_done(response):
+                stream = BufferedStream(response.stream)
+                media_file.descriptor.stream = stream
+
+                return stream
+
+            media_file.descriptor.stream_dfr = \
+                    self._submit_next_request('GET', media_file.descriptor.uri)
+            media_file.descriptor.stream_dfr.addCallback(request_done)
+
+        def stream_done(stream):
+            stream.readExactly(size).chainDeferred(read_dfr)
+
+            return stream
+
+        def stream_failure(failure):
+            read_dfr.errback(failure)
+
+        media_file.descriptor.stream_dfr.addCallbacks(stream_done,
+                stream_failure)
+    
+        return read_dfr
+
+    def uri_is_monitorable(self, uri):
+        return False
+
+    def uri_is_monitored(self, uri):
+        return False

=== added directory 'elisa-plugins/elisa/plugins/http_plugin/tests'
=== added file 'elisa-plugins/elisa/plugins/http_plugin/tests/__init__.py'
=== added file 'elisa-plugins/elisa/plugins/http_plugin/tests/test_http_media.py'
--- elisa-plugins/elisa/plugins/http_plugin/tests/test_http_media.py	1970-01-01 00:00:00 +0000
+++ elisa-plugins/elisa/plugins/http_plugin/tests/test_http_media.py	2008-03-19 10:49:22 +0000
@@ -0,0 +1,525 @@
+# -*- coding: utf-8 -*-
+# Elisa - Home multimedia server
+# Copyright (C) 2007 Fluendo Embedded S.L. (www.fluendo.com).
+# All rights reserved.
+#
+# This file is available under one of two license agreements.
+#
+# This file is licensed under the GPL version 2.
+# See "LICENSE.GPL" in the root of this distribution including a special
+# exception to use Elisa with Fluendo's plugins.
+#
+# The GPL part of Elisa is also available under a commercial licensing
+# agreement from Fluendo.
+# See "LICENSE.Elisa" in the root directory of this distribution package
+# for details on that license.
+
+from twisted.internet import reactor, defer, error
+from twisted.web2.http import StatusResponse
+from twisted.web2.channel.http import HTTPFactory
+from twisted.web2 import iweb, server
+from zope.interface import implements
+
+from elisa.core.tests.elisa_test_case import ElisaTestCase
+from elisa.core.tests.component_test_case import ComponentTestCase
+from elisa.core.tests.test_media_provider import TestMediaProvider
+from elisa.plugins.http_plugin.http_media import ElisaHttpClient, \
+        HttpMedia, HttpMediaClient, ElisaHttpClientNotOpenedError
+from elisa.core.media_uri import MediaUri
+from elisa.core.media_file import MediaFile
+
+import twisted.trial.util
+
+__maintainer__ = 'Alessandro Decina <alessandro@fluendo.com>'
+
+SERVER_PORT = 4242
+SERVER_ADDR = 'http://localhost:%d' % SERVER_PORT
+
+class TestHttpServerCloseConnectionResource(object):
+    implements(iweb.IResource)
+
+    def locateChild(self, request, segments):
+        if segments[0] == '':
+            return (self, server.StopTraversal)
+
+        return (None, segments)
+
+    def renderHTTP(self, request):
+        # yahoo.com sends Connection: close with a 301 response
+        response = StatusResponse(301, 'ktnxbye',
+                'at yahoo, closin ur connescionz')
+        response.headers.setHeader('Connection', ['close'])
+        return response
+
+class TestHttpServerRootResource(object):
+    implements(iweb.IResource)
+
+    def __init__(self):
+        self.my_children = {'': self,
+            'redirect_and_close': TestHttpServerCloseConnectionResource()}
+
+    def locateChild(self, request, segments):
+        res = self.my_children.get(segments[0])
+        if res == self:
+            return (self, server.StopTraversal)
+        elif res == None:
+            return (None, segments)
+        else:
+            return (res, segments[1:])
+
+    def renderHTTP(self, request):
+        response = StatusResponse(200, 'Facebook here I come',
+                'Elisa test HTTP Server BETA')
+        return response
+
+class TestHttpServerRequest(server.Request):
+    site = server.Site(TestHttpServerRootResource())
+
+class TestHttpServerFactory(HTTPFactory):
+    crashed = False
+
+    def buildProtocol(self, addr):
+        if self.crashed:
+            # just refuse the connection
+            return None
+
+        return HTTPFactory.buildProtocol(self, addr)
+
+class SetupServerMixin(object):
+    def setUpClass(self):
+        self.http_factory = TestHttpServerFactory(TestHttpServerRequest)
+        self._port = reactor.listenTCP(SERVER_PORT, self.http_factory)
+
+    def tearDownClass(self):
+        return defer.maybeDeferred(self._port.stopListening)
+
+class TestElisaHttpClientNoPipeline(ElisaTestCase, SetupServerMixin):
+    def setUp(self):
+        self.client = ElisaHttpClient(pipeline=False)
+        setup_dfr = self.client.open('localhost', SERVER_PORT)
+
+        return setup_dfr
+
+    def tearDown(self):
+        return self.client.close()
+    
+    def test_connection_error(self):
+        client = ElisaHttpClient()
+        # connecting to port 0 seems to work on linux, not sure about other oses
+        open_dfr = client.open('localhost', 0)
+
+        def open_done(result):
+            raise Exception('this should not be reached')
+
+        def open_failure(failure):
+            # we expect a connection refused
+            failure.trap(error.ConnectionRefusedError)
+
+        open_dfr.addCallbacks(open_done, open_failure)
+
+        return open_dfr
+
+    def test_close_not_opened(self):
+        client = ElisaHttpClient()
+
+        def close_done(result):
+            raise Exception('this should not be reached')
+
+        def close_failure(failure):
+            failure.trap(ElisaHttpClientNotOpenedError)
+
+        close_dfr = client.close()
+        close_dfr.addCallbacks(close_done, close_failure)
+
+        return close_dfr
+    
+    def test_request_not_opened(self):
+        client = ElisaHttpClient()
+
+        def request_done(result):
+            raise Exception('this should not be reached')
+
+        def request_failure(failure):
+            failure.trap(ElisaHttpClientNotOpenedError)
+
+        request_dfr = client.request('/')
+        request_dfr.addCallbacks(request_done, request_failure)
+
+        return request_dfr
+
+    def test_one(self):
+        """Test a single request and validate the response. Basic test, if this
+        fails something is really wrong.
+        """
+        res_dfr = defer.Deferred()
+
+        def request_done(response):
+            self.assertEqual(response.code, 200)
+
+            return response
+
+        request_dfr = self.client.request('/')
+        request_dfr.addCallback(request_done)
+        request_dfr.chainDeferred(res_dfr)
+
+        return res_dfr
+
+    def test_consecutive(self):
+        """Test two consecutive requests on the same URI. This tests that our
+        client reconnects correctly.
+        """
+        res_dfr = defer.Deferred()
+
+        def first_request_done(response):
+            self.assertEqual(response.code, 200)
+
+            def do_second_request():
+                second_request_dfr = self.client.request('/')
+                second_request_dfr.addCallback(second_request_done)
+                second_request_dfr.chainDeferred(res_dfr)
+           
+#            reactor.callLater(0, do_second_request)
+            do_second_request()
+
+            return response
+
+        def second_request_done(response):
+            self.assertEqual(response.code, 200)
+            
+            return response
+    
+        import pdb
+        pdb.set_trace()
+
+        first_request_dfr = self.client.request('/')
+        first_request_dfr.addCallback(first_request_done)
+        first_request_dfr.addErrback(res_dfr.errback)
+
+        return res_dfr
+
+class TestElisaHttpClientPipeline(ElisaTestCase, SetupServerMixin):
+    def setUp(self):
+        self.client = ElisaHttpClient(pipeline=True)
+        setup_dfr = self.client.open('localhost', SERVER_PORT)
+
+        return setup_dfr
+
+    def tearDown(self):
+        self.http_factory.crashed = False
+        return self.client.close()
+
+    def test_pipeline(self):
+        """Pipeline two requests, expect no errors"""
+
+        res_dfr = defer.Deferred()
+
+        got_first_response = False
+
+        def first_request_done(response):
+            global got_first_response
+            got_first_response = True
+            self.assertEqual(response.code, 200)
+
+            return response
+
+        def second_request_done(response):
+            global got_first_response
+            self.failUnless(got_first_response)
+            
+            self.assertEqual(response.code, 200)
+
+            return response
+
+        first_request_dfr = self.client.request('/')
+        second_request_dfr = self.client.request('/')
+
+        first_request_dfr.addCallback(first_request_done)
+        first_request_dfr.addErrback(res_dfr.errback)
+        second_request_dfr.addCallback(second_request_done)
+        second_request_dfr.chainDeferred(res_dfr)
+
+        return res_dfr
+
+    def test_pipeline_with_close_at_start(self):
+        """Pipeline two requests, the first response is a redirect that closes
+        the current connection.
+
+        This checks that our client correctly reconnects if for some reason the
+        server closes the connection in the first request of a pipeline
+        """
+        res_dfr = defer.Deferred()
+
+        got_first_response = False
+
+        def first_request_done(response):
+            self.assertEqual(response.code, 301)
+            global got_first_response
+            got_first_response = True
+
+            return response
+
+        def second_request_done(response):
+            global got_first_response
+            self.failUnless(got_first_response)
+            self.assertEqual(response.code, 200)
+            
+            return response
+
+        first_request_dfr = self.client.request('/redirect_and_close')
+        second_request_dfr = self.client.request('/')
+
+        first_request_dfr.addCallback(first_request_done)
+        first_request_dfr.addErrback(res_dfr.errback)
+        second_request_dfr.addCallback(second_request_done)
+        second_request_dfr.chainDeferred(res_dfr)
+
+        return res_dfr
+        
+    def test_pipeline_with_close(self):
+        """Pipeline three requests, the first returns OK, the second OBJECT
+        MOVED and closes the connection, the third again OK.
+
+        This checks that our client correctly reconnects if for some reason the
+        server closes the connection in the middle of a pipeline
+        """
+        res_dfr = defer.Deferred()
+
+        got_first_response = False
+        got_second_response = False
+
+        def first_request_done(response):
+            self.assertEqual(response.code, 200)
+            global got_first_response
+            got_first_response = True
+
+            return response
+
+        def second_request_done(response):
+            global got_second_response
+            got_second_response = True
+            global got_first_response
+            self.failUnless(got_first_response)
+            self.assertEqual(response.code, 301)
+            
+            return response
+        
+        def third_request_done(response):
+            global got_second_response
+            self.failUnless(got_second_response)
+            self.assertEqual(response.code, 200)
+             
+            return response
+
+        first_request_dfr = self.client.request('/')
+        second_request_dfr = self.client.request('/redirect_and_close')
+        third_request_dfr = self.client.request('/')
+
+        first_request_dfr.addCallback(first_request_done)
+        first_request_dfr.addErrback(res_dfr.errback)
+        second_request_dfr.addCallback(second_request_done)
+        second_request_dfr.addErrback(res_dfr.errback)
+        third_request_dfr.addCallback(third_request_done)
+        third_request_dfr.chainDeferred(res_dfr)
+
+        return res_dfr
+    
+    def test_pipeline_with_errors(self):
+        """Pipeline three requests, for the first we get a response, for the
+        last two the server closes the connection without writing a response.
+
+        This checks that the clients doesn't get stuck retrying requests if the
+        server stops working.
+        """
+        res_dfr = defer.Deferred()
+
+        got_first_response = False
+        got_second_failure = False
+
+        def first_request_done(response):
+            self.assertEqual(response.code, 301)
+            global got_first_response
+            got_first_response = True
+
+            # do not process successive requests
+            self.http_factory.crashed = True
+
+            return response
+
+        def second_request_done(response):
+            raise Exception('this should not be reached')
+
+        def second_request_failure(failure):
+            global got_second_failure
+            got_second_failure = True
+            global got_first_response
+            self.failUnless(got_first_response)
+            failure.trap(error.ConnectionDone)
+       
+        def third_request_done(response):
+            raise Exception('this should not be reached')
+
+        def third_request_failure(failure):
+            global got_second_response
+            self.failUnless(got_second_failure)
+            failure.trap(error.ConnectionDone)
+
+        first_request_dfr = self.client.request('/redirect_and_close')
+        second_request_dfr = self.client.request('/')
+        third_request_dfr = self.client.request('/')
+
+        first_request_dfr.addCallback(first_request_done)
+        first_request_dfr.addErrback(res_dfr.errback)
+        second_request_dfr.addCallbacks(second_request_done,
+                second_request_failure)
+        third_request_dfr.addCallbacks(third_request_done,
+                third_request_failure)
+        third_request_dfr.chainDeferred(res_dfr)
+
+        return res_dfr
+
+class TestHttpMediaClient(ElisaTestCase, SetupServerMixin):
+    def test_open_and_close(self):
+        client = HttpMediaClient()
+        def open_done(result):
+            dfr = client.close()
+            dfr.addCallback(close_done)
+            
+            return dfr
+
+        def close_done(result):
+            # now the client is closed, a request will return in an error
+            dfr = client.request('/')
+            dfr.addCallbacks(request_done, request_failure)
+
+            return dfr
+
+        def request_done(result):
+            raise Exception('this should not be reached')
+
+        def request_failure(failure):
+            # ok this is what we wanted
+            return None
+
+        dfr = client.open('localhost', SERVER_PORT)
+        dfr.addCallback(open_done)
+
+        return dfr
+
+    def test_two_open_and_close(self):
+        client = HttpMediaClient()
+        res_dfr = defer.Deferred()
+
+        def open_done(result):
+            # use callLater to avoid running the callbacks of the open()
+            # deferred recursively
+
+            def call_open_later():
+                dfr = client.open('localhost', SERVER_PORT)
+                dfr.addCallback(open2_done)
+                dfr.chainDeferred(res_dfr)
+
+            reactor.callLater(0, call_open_later)
+
+            return result
+        
+        def open2_done(result):
+            dfr = client.close()
+            dfr.addCallback(close_done)
+
+            return dfr
+
+        def close_done(result):
+            # here a request should succeed, because two clients called open and
+            # only one called close, so the client is still connected
+            dfr = client.request('/')
+            dfr.addCallback(request_done)
+
+            return dfr
+
+        def request_done(result):
+            dfr = client.close()
+            return dfr
+
+        dfr = client.open('localhost', SERVER_PORT)
+        dfr.addCallback(open_done)
+
+        return res_dfr
+
+    def test_open_same_time(self):
+        client = HttpMediaClient()
+
+        open_sequence = []
+        def open_done(result, open_num):
+            open_sequence.append(open_num)
+            self.assertEqual(len(open_sequence), open_num)
+            return client.close()
+
+        open1_dfr = client.open('localhost', SERVER_PORT)
+        open1_dfr.addCallback(open_done, 1)
+        open2_dfr = client.open('localhost', SERVER_PORT)
+        open2_dfr.addCallback(open_done, 2)
+        open3_dfr = client.open('localhost', SERVER_PORT)
+        open3_dfr.addCallback(open_done, 3)
+        
+        return defer.DeferredList([open1_dfr, open2_dfr, open3_dfr])
+
+#FIXME: i need more tests
+class TestElisaHttpMedia(ComponentTestCase, SetupServerMixin):
+    root_uri = MediaUri(SERVER_ADDR + '/')
+    component_class = HttpMedia
+
+    def test_open_and_close(self):
+        component = self.component
+
+        def open_done(media_file):
+            return component.close(media_file)
+
+        dfr = component.open(self.root_uri)
+        dfr.addCallback(open_done)
+
+        return dfr
+
+    def test_read_all(self):
+        component = self.component
+
+        def open_done(media_file):
+            dfr = media_file.read()
+            dfr.addCallback(read_done, media_file)
+
+            return dfr
+
+        def read_done(data, media_file):
+            # FIXME: don't hardcode 144
+            self.assertEqual(len(data), 144)
+
+            return component.close(media_file)
+
+        dfr = component.open(self.root_uri)
+        dfr.addCallback(open_done)
+
+        return dfr
+
+    def test_read_chunks(self):
+        component = self.component
+
+        def open_done(media_file):
+            dfr = media_file.read(72)
+            dfr.addCallback(read1_done, media_file)
+
+            return dfr
+
+        def read1_done(data, media_file):
+            self.assertEqual(len(data), 72)
+            
+            dfr = media_file.read(72)
+            dfr.addCallback(read2_done, media_file)
+
+            return dfr
+
+        def read2_done(data, media_file):
+            self.assertEqual(len(data), 72)
+
+            return component.close(media_file)
+
+        dfr = component.open(self.root_uri)
+        dfr.addCallback(open_done)