mirror of
https://github.com/xen-project/xen.git
synced 2026-10-28 18:21:39 +01:00
bitkeeper revision 1.1159.1.165 (4151594cmY68ImBqRlb00RkEm61RcQ)
Split the xend client code into separate synch and asynch client protocols, and load the asynch one on demand. This speeds up xm as it doesn't load the asynch code.
This commit is contained in:
1 parent
a23e4c0121
commit
c2c9f7989d
6 files changed
+277
-243
No files matched your search
@@ -1,4 +1,5 @@
|
||||
from xen.xend.XendClient import aserver as server
|
||||
from xen.xend.XendClient import getAsynchServer
|
||||
server = getAsynchServer()
|
||||
from xen.xend import PrettyPrint
|
||||
|
||||
from xen.sv.HTMLBase import HTMLBase
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
# Copyright (C) 2004 Mike Wray <mike.wray@hp.com>
|
||||
|
||||
from twisted.protocols import http
|
||||
from twisted.internet.protocol import ClientCreator
|
||||
from twisted.internet.defer import Deferred
|
||||
from twisted.internet import reactor
|
||||
|
||||
from XendProtocol import XendClientProtocol, XendRequest
|
||||
|
||||
class AsynchXendClient(http.HTTPClient):
|
||||
"""A subclass of twisted's HTTPClient to deal with a connection to xend.
|
||||
Makes the request when connected, and delegates handling responses etc.
|
||||
to its protocol (usually an AsynchXendClientProtocol instance).
|
||||
"""
|
||||
def __init__(self, protocol, request):
|
||||
self.protocol = protocol
|
||||
self.request = request
|
||||
|
||||
def connectionMade(self):
|
||||
request = self.request
|
||||
url = self.request.url
|
||||
self.sendCommand(request.method, url.fullpath())
|
||||
self.sendHeader('Host', url.location())
|
||||
for (k, v) in request.headers.items():
|
||||
self.sendHeader(k, v)
|
||||
if request.data:
|
||||
self.sendHeader('Content-Length', len(request.data))
|
||||
self.endHeaders()
|
||||
if request.data:
|
||||
self.transport.write(request.data)
|
||||
|
||||
def handleStatus(self, version, status, message):
|
||||
return self.protocol.handleStatus(version, status, message)
|
||||
|
||||
def handleHeader(self, key, val):
|
||||
return self.protocol.handleHeader(key, val)
|
||||
|
||||
def handleResponse(self, data):
|
||||
return self.protocol.handleResponse(data)
|
||||
|
||||
class AsynchXendClientProtocol(XendClientProtocol):
|
||||
"""An asynchronous xend client. Uses twisted to connect to xend
|
||||
and make the request. It does not block waiting for the result,
|
||||
but sets up a deferred that is called when the result becomes available.
|
||||
|
||||
Uses AsynchXendClient to manage the connection.
|
||||
"""
|
||||
def __init__(self):
|
||||
self.err = None
|
||||
self.headers = {}
|
||||
|
||||
def xendRequest(self, url, method, args=None):
|
||||
"""Make a request to xend. The returned deferred is called when
|
||||
the result is available.
|
||||
|
||||
@param url: xend request url
|
||||
@param method: http method: POST or GET
|
||||
@param args: request arguments (dict)
|
||||
@return: deferred
|
||||
"""
|
||||
request = XendRequest(url, method, args)
|
||||
self.deferred = Deferred()
|
||||
clientCreator = ClientCreator(reactor, AsynchXendClient, self, request)
|
||||
clientCreator.connectTCP(url.host, url.port)
|
||||
return self.deferred
|
||||
|
||||
def callErrback(self, err):
|
||||
if not self.deferred.called:
|
||||
self.err = err
|
||||
self.deferred.errback(err)
|
||||
return err
|
||||
|
||||
def callCallback(self, val):
|
||||
if not self.deferred.called:
|
||||
self.deferred.callback(val)
|
||||
return val
|
||||
|
||||
def handleException(self, err):
|
||||
return self.callErrback(err)
|
||||
|
||||
def handleHeader(self, key, val):
|
||||
self.headers[key.lower()] = val
|
||||
|
||||
def getHeader(self, key):
|
||||
return self.headers.get(key.lower())
|
||||
|
||||
def handleResponse(self, data):
|
||||
if self.err: return self.err
|
||||
val = XendClientProtocol.handleResponse(self, data)
|
||||
if isinstance(val, Exception):
|
||||
self.callErrback(val)
|
||||
else:
|
||||
self.callCallback(val)
|
||||
return val
|
||||
@@ -11,27 +11,14 @@ is accessible via this API.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import httplib
|
||||
import types
|
||||
from StringIO import StringIO
|
||||
|
||||
|
||||
from twisted.protocols import http
|
||||
from twisted.internet.protocol import ClientCreator
|
||||
from twisted.internet.defer import Deferred
|
||||
from twisted.internet import reactor
|
||||
|
||||
from encode import *
|
||||
import sxp
|
||||
import PrettyPrint
|
||||
from XendProtocol import XendClientProtocol, SynchXendClientProtocol, XendError
|
||||
|
||||
DEBUG = 0
|
||||
|
||||
class XendError(RuntimeError):
|
||||
"""Error class for 'expected errors' when talking to xend.
|
||||
"""
|
||||
pass
|
||||
|
||||
def fileof(val):
|
||||
"""Converter for passing configs or other 'large' data.
|
||||
Handles lists, files directly.
|
||||
@@ -102,229 +89,6 @@ class URL:
|
||||
query=query,
|
||||
frag=frag)
|
||||
|
||||
class XendRequest:
|
||||
"""A request to xend.
|
||||
"""
|
||||
|
||||
def __init__(self, url, method, args):
|
||||
"""Create a request. Sets up the headers, argument data, and the
|
||||
url.
|
||||
|
||||
@param url: the url to request
|
||||
@param method: request method, GET or POST
|
||||
@param args: dict containing request args, if any
|
||||
"""
|
||||
if url.proto != 'http':
|
||||
raise ValueError('Invalid protocol: ' + url.proto)
|
||||
(hdr, data) = encode_data(args)
|
||||
if args and method == 'GET':
|
||||
url.query = data
|
||||
data = None
|
||||
if method == "POST" and url.path.endswith('/'):
|
||||
url.path = url.path[:-1]
|
||||
|
||||
self.headers = hdr
|
||||
self.data = data
|
||||
self.url = url
|
||||
self.method = method
|
||||
|
||||
class XendClientProtocol:
|
||||
"""Abstract class for xend clients.
|
||||
"""
|
||||
def xendRequest(self, url, method, args=None):
|
||||
"""Make a request to xend.
|
||||
Implement in a subclass.
|
||||
|
||||
@param url: xend request url
|
||||
@param method: http method: POST or GET
|
||||
@param args: request arguments (dict)
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def xendGet(self, url, args=None):
|
||||
"""Make a xend request using HTTP GET.
|
||||
Requests using GET are usually 'safe' and may be repeated without
|
||||
nasty side-effects.
|
||||
|
||||
@param url: xend request url
|
||||
@param data: request arguments (dict)
|
||||
"""
|
||||
return self.xendRequest(url, "GET", args)
|
||||
|
||||
def xendPost(self, url, args):
|
||||
"""Make a xend request using HTTP POST.
|
||||
Requests using POST potentially cause side-effects, and should
|
||||
not be repeated unless you really want to repeat the side
|
||||
effect.
|
||||
|
||||
@param url: xend request url
|
||||
@param args: request arguments (dict)
|
||||
"""
|
||||
return self.xendRequest(url, "POST", args)
|
||||
|
||||
def handleStatus(self, version, status, message):
|
||||
"""Handle the status returned from the request.
|
||||
"""
|
||||
status = int(status)
|
||||
if status in [ http.NO_CONTENT ]:
|
||||
return None
|
||||
if status not in [ http.OK, http.CREATED, http.ACCEPTED ]:
|
||||
return self.handleException(XendError(message))
|
||||
return 'ok'
|
||||
|
||||
def handleResponse(self, data):
|
||||
"""Handle the data returned in response to the request.
|
||||
"""
|
||||
if data is None: return None
|
||||
type = self.getHeader('Content-Type')
|
||||
if type != sxp.mime_type:
|
||||
return data
|
||||
try:
|
||||
pin = sxp.Parser()
|
||||
pin.input(data);
|
||||
pin.input_eof()
|
||||
val = pin.get_val()
|
||||
except sxp.ParseError, err:
|
||||
return self.handleException(err)
|
||||
if isinstance(val, types.ListType) and sxp.name(val) == 'xend.err':
|
||||
err = XendError(val[1])
|
||||
return self.handleException(err)
|
||||
return val
|
||||
|
||||
def handleException(self, err):
|
||||
"""Handle an exception during the request.
|
||||
May be overridden in a subclass.
|
||||
"""
|
||||
raise err
|
||||
|
||||
def getHeader(self, key):
|
||||
"""Get a header from the response.
|
||||
Case is ignored in the key.
|
||||
|
||||
@param key: header key
|
||||
@return: header
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
class SynchXendClientProtocol(XendClientProtocol):
|
||||
"""A synchronous xend client. This will make a request, wait for
|
||||
the reply and return the result.
|
||||
"""
|
||||
|
||||
resp = None
|
||||
|
||||
def xendRequest(self, url, method, args=None):
|
||||
"""Make a request to xend.
|
||||
|
||||
@param url: xend request url
|
||||
@param method: http method: POST or GET
|
||||
@param args: request arguments (dict)
|
||||
"""
|
||||
self.request = XendRequest(url, method, args)
|
||||
conn = httplib.HTTPConnection(url.location())
|
||||
if DEBUG: conn.set_debuglevel(1)
|
||||
conn.request(method, url.fullpath(), self.request.data, self.request.headers)
|
||||
resp = conn.getresponse()
|
||||
self.resp = resp
|
||||
val = self.handleStatus(resp.version, resp.status, resp.reason)
|
||||
if val is None:
|
||||
data = None
|
||||
else:
|
||||
data = resp.read()
|
||||
conn.close()
|
||||
val = self.handleResponse(data)
|
||||
return val
|
||||
|
||||
def getHeader(self, key):
|
||||
return self.resp.getheader(key)
|
||||
|
||||
|
||||
class AsynchXendClient(http.HTTPClient):
|
||||
"""A subclass of twisted's HTTPClient to deal with a connection to xend.
|
||||
Makes the request when connected, and delegates handling responses etc.
|
||||
to its protocol (usually an AsynchXendClientProtocol instance).
|
||||
"""
|
||||
def __init__(self, protocol, request):
|
||||
self.protocol = protocol
|
||||
self.request = request
|
||||
|
||||
def connectionMade(self):
|
||||
request = self.request
|
||||
url = self.request.url
|
||||
self.sendCommand(request.method, url.fullpath())
|
||||
self.sendHeader('Host', url.location())
|
||||
for (k, v) in request.headers.items():
|
||||
self.sendHeader(k, v)
|
||||
if request.data:
|
||||
self.sendHeader('Content-Length', len(request.data))
|
||||
self.endHeaders()
|
||||
if request.data:
|
||||
self.transport.write(request.data)
|
||||
|
||||
def handleStatus(self, version, status, message):
|
||||
return self.protocol.handleStatus(version, status, message)
|
||||
|
||||
def handleHeader(self, key, val):
|
||||
return self.protocol.handleHeader(key, val)
|
||||
|
||||
def handleResponse(self, data):
|
||||
return self.protocol.handleResponse(data)
|
||||
|
||||
class AsynchXendClientProtocol(XendClientProtocol):
|
||||
"""An asynchronous xend client. Uses twisted to connect to xend
|
||||
and make the request. It does not block waiting for the result,
|
||||
but sets up a deferred that is called when the result becomes available.
|
||||
|
||||
Uses AsynchXendClient to manage the connection.
|
||||
"""
|
||||
def __init__(self):
|
||||
self.err = None
|
||||
self.headers = {}
|
||||
|
||||
def xendRequest(self, url, method, args=None):
|
||||
"""Make a request to xend. The returned deferred is called when
|
||||
the result is available.
|
||||
|
||||
@param url: xend request url
|
||||
@param method: http method: POST or GET
|
||||
@param args: request arguments (dict)
|
||||
@return: deferred
|
||||
"""
|
||||
request = XendRequest(url, method, args)
|
||||
self.deferred = Deferred()
|
||||
clientCreator = ClientCreator(reactor, AsynchXendClient, self, request)
|
||||
clientCreator.connectTCP(url.host, url.port)
|
||||
return self.deferred
|
||||
|
||||
def callErrback(self, err):
|
||||
if not self.deferred.called:
|
||||
self.err = err
|
||||
self.deferred.errback(err)
|
||||
return err
|
||||
|
||||
def callCallback(self, val):
|
||||
if not self.deferred.called:
|
||||
self.deferred.callback(val)
|
||||
return val
|
||||
|
||||
def handleException(self, err):
|
||||
return self.callErrback(err)
|
||||
|
||||
def handleHeader(self, key, val):
|
||||
self.headers[key.lower()] = val
|
||||
|
||||
def getHeader(self, key):
|
||||
return self.headers.get(key.lower())
|
||||
|
||||
def handleResponse(self, data):
|
||||
if self.err: return self.err
|
||||
val = XendClientProtocol.handleResponse(self, data)
|
||||
if isinstance(val, Exception):
|
||||
self.callErrback(val)
|
||||
else:
|
||||
self.callCallback(val)
|
||||
return val
|
||||
|
||||
class Xend:
|
||||
"""Client interface to Xend.
|
||||
"""
|
||||
@@ -586,9 +350,28 @@ class Xend:
|
||||
{'op' : 'inject',
|
||||
'event' : fileof(sxpr) })
|
||||
|
||||
def getAsynchXendClientProtocol():
|
||||
"""Load AsynchXendClientProtocol on demand to avoid the cost.
|
||||
"""
|
||||
global AsynchXendClientProtocol
|
||||
try:
|
||||
AsynchXendClientProtocol
|
||||
except:
|
||||
from XendAsynchProtocol import AsynchXendClientProtocol
|
||||
return AsynchXendClientProtocol
|
||||
|
||||
def getAsynchServer():
|
||||
"""Load AsynchXendClientProtocol and create an asynch xend client.
|
||||
|
||||
@return asynch Xend
|
||||
"""
|
||||
getAsynchXendClientProtocol()
|
||||
return Xend(AsynchXendClientProtocol())
|
||||
|
||||
def xendmain(srv, asynch, fn, args):
|
||||
if asynch:
|
||||
client = AsynchXendClientProtocol()
|
||||
getAsynchXendClientProtocol()
|
||||
client = AsynchXendClientProtocol()
|
||||
else:
|
||||
client = None
|
||||
xend = Xend(srv=srv, client=client)
|
||||
@@ -655,4 +438,3 @@ if __name__ == "__main__":
|
||||
main(sys.argv)
|
||||
else:
|
||||
server = Xend()
|
||||
aserver = Xend( AsynchXendClientProtocol() )
|
||||
@@ -0,0 +1,156 @@
|
||||
# Copyright (C) 2004 Mike Wray <mike.wray@hp.com>
|
||||
|
||||
import httplib
|
||||
import types
|
||||
|
||||
from encode import *
|
||||
import sxp
|
||||
|
||||
DEBUG = 0
|
||||
|
||||
HTTP_OK = 200
|
||||
HTTP_CREATED = 201
|
||||
HTTP_ACCEPTED = 202
|
||||
HTTP_NO_CONTENT = 204
|
||||
|
||||
class XendError(RuntimeError):
|
||||
"""Error class for 'expected errors' when talking to xend.
|
||||
"""
|
||||
pass
|
||||
|
||||
class XendRequest:
|
||||
"""A request to xend.
|
||||
"""
|
||||
|
||||
def __init__(self, url, method, args):
|
||||
"""Create a request. Sets up the headers, argument data, and the
|
||||
url.
|
||||
|
||||
@param url: the url to request
|
||||
@param method: request method, GET or POST
|
||||
@param args: dict containing request args, if any
|
||||
"""
|
||||
if url.proto != 'http':
|
||||
raise ValueError('Invalid protocol: ' + url.proto)
|
||||
(hdr, data) = encode_data(args)
|
||||
if args and method == 'GET':
|
||||
url.query = data
|
||||
data = None
|
||||
if method == "POST" and url.path.endswith('/'):
|
||||
url.path = url.path[:-1]
|
||||
|
||||
self.headers = hdr
|
||||
self.data = data
|
||||
self.url = url
|
||||
self.method = method
|
||||
|
||||
class XendClientProtocol:
|
||||
"""Abstract class for xend clients.
|
||||
"""
|
||||
def xendRequest(self, url, method, args=None):
|
||||
"""Make a request to xend.
|
||||
Implement in a subclass.
|
||||
|
||||
@param url: xend request url
|
||||
@param method: http method: POST or GET
|
||||
@param args: request arguments (dict)
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def xendGet(self, url, args=None):
|
||||
"""Make a xend request using HTTP GET.
|
||||
Requests using GET are usually 'safe' and may be repeated without
|
||||
nasty side-effects.
|
||||
|
||||
@param url: xend request url
|
||||
@param data: request arguments (dict)
|
||||
"""
|
||||
return self.xendRequest(url, "GET", args)
|
||||
|
||||
def xendPost(self, url, args):
|
||||
"""Make a xend request using HTTP POST.
|
||||
Requests using POST potentially cause side-effects, and should
|
||||
not be repeated unless you really want to repeat the side
|
||||
effect.
|
||||
|
||||
@param url: xend request url
|
||||
@param args: request arguments (dict)
|
||||
"""
|
||||
return self.xendRequest(url, "POST", args)
|
||||
|
||||
def handleStatus(self, version, status, message):
|
||||
"""Handle the status returned from the request.
|
||||
"""
|
||||
status = int(status)
|
||||
if status in [ HTTP_NO_CONTENT ]:
|
||||
return None
|
||||
if status not in [ HTTP_OK, HTTP_CREATED, HTTP_ACCEPTED ]:
|
||||
return self.handleException(XendError(message))
|
||||
return 'ok'
|
||||
|
||||
def handleResponse(self, data):
|
||||
"""Handle the data returned in response to the request.
|
||||
"""
|
||||
if data is None: return None
|
||||
type = self.getHeader('Content-Type')
|
||||
if type != sxp.mime_type:
|
||||
return data
|
||||
try:
|
||||
pin = sxp.Parser()
|
||||
pin.input(data);
|
||||
pin.input_eof()
|
||||
val = pin.get_val()
|
||||
except sxp.ParseError, err:
|
||||
return self.handleException(err)
|
||||
if isinstance(val, types.ListType) and sxp.name(val) == 'xend.err':
|
||||
err = XendError(val[1])
|
||||
return self.handleException(err)
|
||||
return val
|
||||
|
||||
def handleException(self, err):
|
||||
"""Handle an exception during the request.
|
||||
May be overridden in a subclass.
|
||||
"""
|
||||
raise err
|
||||
|
||||
def getHeader(self, key):
|
||||
"""Get a header from the response.
|
||||
Case is ignored in the key.
|
||||
|
||||
@param key: header key
|
||||
@return: header
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
class SynchXendClientProtocol(XendClientProtocol):
|
||||
"""A synchronous xend client. This will make a request, wait for
|
||||
the reply and return the result.
|
||||
"""
|
||||
|
||||
resp = None
|
||||
|
||||
def xendRequest(self, url, method, args=None):
|
||||
"""Make a request to xend.
|
||||
|
||||
@param url: xend request url
|
||||
@param method: http method: POST or GET
|
||||
@param args: request arguments (dict)
|
||||
"""
|
||||
self.request = XendRequest(url, method, args)
|
||||
conn = httplib.HTTPConnection(url.location())
|
||||
if DEBUG: conn.set_debuglevel(1)
|
||||
conn.request(method, url.fullpath(), self.request.data, self.request.headers)
|
||||
resp = conn.getresponse()
|
||||
self.resp = resp
|
||||
val = self.handleStatus(resp.version, resp.status, resp.reason)
|
||||
if val is None:
|
||||
data = None
|
||||
else:
|
||||
data = resp.read()
|
||||
conn.close()
|
||||
val = self.handleResponse(data)
|
||||
return val
|
||||
|
||||
def getHeader(self, key):
|
||||
return self.resp.getheader(key)
|
||||
|
||||