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