1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 from flumotion.common import log
20 from flumotion.component.misc.httpserver.httpcached import common
21 from flumotion.component.misc.httpserver.httpcached import http_utils
22 from flumotion.component.misc.httpserver.httpcached import server_selection
23
24
25 LOG_CATEGORY = "request-manager"
29
30 logCategory = LOG_CATEGORY
31
33 """
34 Selector: a ServerSelector
35 Client: HttpClient (StreamRequester)
36 """
37 self.selector = selector
38 self.client = client
39
40 - def retrieve(self, consumer, url,
41 ifModifiedSince=None, ifUnmodifiedSince=None,
42 start=None, size=None):
43 """
44 Consumer: a StreamConsumer
45 Url:
46 Start: Position from which to start the download
47 Size: Number of bytes to download
48 IfModifiedSince:
49 IfUnmodifiedSince:
50 """
51 servers = self.selector.getServers()
52 consumer_manager = ConsumerManager(consumer, url, start, size,
53 ifModifiedSince, ifUnmodifiedSince,
54 servers, self.client)
55 return consumer_manager.retrieve()
56
58 return self.selector.setup()
59
62
65
66 logCategory = LOG_CATEGORY
67
68 - def __init__(self, consumer, url, start, size, ifModifiedSince,
69 ifUnmodifiedSince, servers, client):
70 self.consumer = consumer
71 self.url = url
72 self.start = start
73 self.size = size
74 self.ifModifiedSince = ifModifiedSince
75 self.ifUnmodifiedSince = ifUnmodifiedSince
76 self.servers = servers
77 self.client = client
78 self.current_server = None
79 self.current_request = None
80 self.last_error = None
81 self.last_message = None
82
83 self.logName = common.log_id(self)
84
85 @property
87 if self.current_request:
88 return self.current_request.host
89 return None
90
91 @property
93 if self.current_request:
94 return self.current_request.port
95 return None
96
98 try:
99 s = self.servers.next()
100 self.current_server = s
101 if self.size is None or self.start is None:
102 self.debug("Retrieving %s from %s:%s", self.url,
103 self.current_server.ip, self.current_server.port)
104 else:
105 self.debug("Retrieving range %s-%s (%s B) of %s from %s:%s",
106 self.start, self.start + self.size, self.size,
107 self.url, self.current_server.ip,
108 self.current_server.port)
109 proxy_address = s.ip
110 proxy_port = s.port
111 self.current_request =\
112 self.client.retrieve(self, self.url,
113 proxyAddress=proxy_address,
114 proxyPort=proxy_port,
115 ifModifiedSince=self.ifModifiedSince,
116 ifUnmodifiedSince=self.ifUnmodifiedSince,
117 start=self.start, size=self.size)
118 self.log("Retrieving data using %s", self.current_request.logName)
119 return self
120 except StopIteration:
121 code = self.last_error or common.SERVER_UNAVAILABLE
122 message = self.last_message or ""
123 self.consumer.serverError(self, code, message)
124 return self
125
127 self.log("Pausing request %s", self.url)
128 self.current_request.pause()
129
131 self.log("Resuming request %s", self.url)
132 self.current_request.resume()
133
135 self.debug("Canceling request %s", self.url)
136 self.current_request.cancel()
137 self.current_request = None
138
152
159
165
166 - def onInfo(self, getter, info):
170
171 - def onData(self, getter, data):
175
180