rendered paste body=== 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)