1 # -*- test-case-name: twisted.web.test.test_distrib -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Distributed web servers.
8 This is going to have to be refactored so that argument parsing is done
9 by each subprocess and not by the main web server (i.e. GET, POST etc.).
13 import types, os, copy, cStringIO
19 from xml.dom.minidom import Element, Text
22 from twisted.spread import pb
23 from twisted.spread.banana import SIZE_LIMIT
24 from twisted.web import http, resource, server, html, static
25 from twisted.web.http_headers import Headers
26 from twisted.python import log
27 from twisted.persisted import styles
28 from twisted.internet import address, reactor
31 class _ReferenceableProducerWrapper(pb.Referenceable):
32 def __init__(self, producer):
33 self.producer = producer
35 def remote_resumeProducing(self):
36 self.producer.resumeProducing()
38 def remote_pauseProducing(self):
39 self.producer.pauseProducing()
41 def remote_stopProducing(self):
42 self.producer.stopProducing()
45 class Request(pb.RemoteCopy, server.Request):
47 A request which was received by a L{ResourceSubscription} and sent via
48 PB to a distributed node.
50 def setCopyableState(self, state):
52 Initialize this L{twisted.web.distrib.Request} based on the copied
53 state so that it closely resembles a L{twisted.web.server.Request}.
55 for k in 'host', 'client':
57 addrdesc = {'INET': 'TCP', 'UNIX': 'UNIX'}[tup[0]]
58 addr = {'TCP': lambda: address.IPv4Address(addrdesc,
60 'UNIX': lambda: address.UNIXAddress(tup[1])}[addrdesc]()
62 state['requestHeaders'] = Headers(dict(state['requestHeaders']))
63 pb.RemoteCopy.setCopyableState(self, state)
64 # Emulate the local request interface --
65 self.content = cStringIO.StringIO(self.content_data)
66 self.finish = self.remote.remoteMethod('finish')
67 self.setHeader = self.remote.remoteMethod('setHeader')
68 self.addCookie = self.remote.remoteMethod('addCookie')
69 self.setETag = self.remote.remoteMethod('setETag')
70 self.setResponseCode = self.remote.remoteMethod('setResponseCode')
71 self.setLastModified = self.remote.remoteMethod('setLastModified')
73 # To avoid failing if a resource tries to write a very long string
74 # all at once, this one will be handled slightly differently.
75 self._write = self.remote.remoteMethod('write')
78 def write(self, bytes):
80 Write the given bytes to the response body.
82 @param bytes: The bytes to write. If this is longer than 640k, it
83 will be split up into smaller pieces.
88 self._write(bytes[start:end])
91 if start >= len(bytes):
95 def registerProducer(self, producer, streaming):
96 self.remote.callRemote("registerProducer",
97 _ReferenceableProducerWrapper(producer),
98 streaming).addErrback(self.fail)
100 def unregisterProducer(self):
101 self.remote.callRemote("unregisterProducer").addErrback(self.fail)
103 def fail(self, failure):
107 pb.setUnjellyableForClass(server.Request, Request)
110 def __init__(self, request):
111 self.request = request
113 def finished(self, result):
114 if result != server.NOT_DONE_YET:
115 assert isinstance(result, types.StringType),\
116 "return value not a string"
117 self.request.write(result)
118 self.request.finish()
120 def failed(self, failure):
122 failure = str(failure)
124 resource.ErrorPage(http.INTERNAL_SERVER_ERROR,
125 "Server Connection Lost",
126 "Connection to distributed server lost:" +
128 render(self.request))
129 self.request.finish()
133 class ResourceSubscription(resource.Resource):
136 def __init__(self, host, port):
137 resource.Resource.__init__(self)
141 self.publisher = None
143 def __getstate__(self):
144 """Get persistent state for this ResourceSubscription.
146 # When I unserialize,
147 state = copy.copy(self.__dict__)
148 # Publisher won't be connected...
149 state['publisher'] = None
150 # I won't be making a connection
152 # There will be no pending requests.
153 state['pending'] = []
156 def connected(self, publisher):
157 """I've connected to a publisher; I'll now send all my requests.
159 log.msg('connected to publisher')
160 publisher.broker.notifyOnDisconnect(self.booted)
161 self.publisher = publisher
163 for request in self.pending:
167 def notConnected(self, msg):
168 """I can't connect to a publisher; I'll now reply to all pending
171 log.msg("could not connect to distributed web service: %s" % msg)
173 self.publisher = None
174 for request in self.pending:
175 request.write("Unable to connect to distributed server.")
180 self.notConnected("connection dropped")
182 def render(self, request):
183 """Render this request, from my server.
185 This will always be asynchronous, and therefore return NOT_DONE_YET.
186 It spins off a request to the pb client, and either adds it to the list
187 of pending issues or requests it immediately, depending on if the
188 client is already connected.
190 if not self.publisher:
191 self.pending.append(request)
194 bf = pb.PBClientFactory()
196 if self.host == "unix":
197 reactor.connectUNIX(self.port, bf, timeout)
199 reactor.connectTCP(self.host, self.port, bf, timeout)
200 d = bf.getRootObject()
201 d.addCallbacks(self.connected, self.notConnected)
205 self.publisher.callRemote('request', request).addCallbacks(i.finished, i.failed)
206 return server.NOT_DONE_YET
210 class ResourcePublisher(pb.Root, styles.Versioned):
212 L{ResourcePublisher} exposes a remote API which can be used to respond
215 @ivar site: The site which will be used for resource lookup.
216 @type site: L{twisted.web.server.Site}
218 def __init__(self, site):
221 persistenceVersion = 2
223 def upgradeToVersion2(self):
224 self.application.authorizer.removeIdentity("web")
225 del self.application.services[self.serviceName]
228 del self.perspectiveName
230 def getPerspectiveNamed(self, name):
234 def remote_request(self, request):
236 Look up the resource for the given request and render it.
238 res = self.site.getResourceFor(request)
240 result = res.render(request)
241 if result is not server.NOT_DONE_YET:
242 request.write(result)
244 return server.NOT_DONE_YET
248 class UserDirectory(resource.Resource):
250 A resource which lists available user resources and serves them as
253 @ivar _pwd: An object like L{pwd} which is used to enumerate users and
254 their home directories.
257 userDirName = 'public_html'
258 userSocketName = '.twistd-web-pb'
263 <title>twisted.web.distrib.UserDirectory</title>
268 font-family: Lucida, Verdana, Helvetica, Arial, sans-serif;
270 text-decoration: none;
275 font-family: Lucida, Verdana, Helvetica, Arial, sans-serif;
277 text-decoration: none;
283 font-family: "Courier New", Courier, monospace;
286 p, body, td, ol, ul, menu, blockquote, div
288 font-family: Lucida, Verdana, Helvetica, Arial, sans-serif;
295 <h1>twisted.web.distrib.UserDirectory</h1>
302 def __init__(self, userDatabase=None):
303 resource.Resource.__init__(self)
304 if userDatabase is None:
306 self._pwd = userDatabase
311 Return a list of two-tuples giving links to user resources and text to
312 associate with those links.
315 for user in self._pwd.getpwall():
316 name, passwd, uid, gid, gecos, dir, shell = user
317 realname = gecos.split(',')[0]
320 if os.path.exists(os.path.join(dir, self.userDirName)):
321 users.append((name, realname + ' (file)'))
322 twistdsock = os.path.join(dir, self.userSocketName)
323 if os.path.exists(twistdsock):
324 linkName = name + '.twistd'
325 users.append((linkName, realname + ' (twistd)'))
329 def render_GET(self, request):
331 Render as HTML a listing of all known users with links to their
334 listing = Element('ul')
335 for link, text in self._users():
336 linkElement = Element('a')
337 linkElement.setAttribute('href', link + '/')
340 linkElement.appendChild(textNode)
342 item.appendChild(linkElement)
343 listing.appendChild(item)
344 return self.template % {'users': listing.toxml()}
347 def getChild(self, name, request):
353 if name[-len(td):] == td:
354 username = name[:-len(td)]
360 pw_name, pw_passwd, pw_uid, pw_gid, pw_gecos, pw_dir, pw_shell \
361 = self._pwd.getpwnam(username)
363 return resource.NoResource()
365 twistdsock = os.path.join(pw_dir, self.userSocketName)
366 rs = ResourceSubscription('unix',twistdsock)
367 self.putChild(name, rs)
370 path = os.path.join(pw_dir, self.userDirName)
371 if not os.path.exists(path):
372 return resource.NoResource()
373 return static.File(path)