TripeServiceJob
OptParse
Queue
+ SelIOWatcher
TripeCommand
TripeSynchronousCommand
TripeAsynchronousCommand
TripeCommandIterator
TripeConnection
TripeCommandDispatcher
- SelCommandDispatcher
- TripeServiceManager
+ TripeServiceManager
TripeService
TripeServiceCommand
"""
def switch(me, *args, **kw):
assert _Coroutine.getcurrent() is rootcr
+ if _debug: print '* %s' % me
_Coroutine.switch(me, *args, **kw)
+ if _debug: print '* %s' % rootcr
###--------------------------------------------------------------------------
### Default places for things.
"""
Nonblocking read from SOCK.
- Try to return LEN bytes. If couldn't read anything, return None. EOF is
+ Try to return LEN bytes. If couldn't read anything, return `None'. EOF is
returned as an empty string.
"""
try:
me.socket = socket
me.sock = None
me.lbuf = None
+ me.iowatch = SelIOWatcher(me)
def connect(me):
"""
Disconnect the physical connection.
Invoke the `disconnected' method, giving the provided REASON, which
- should be either None or an exception.
+ should be either `None' or an exception.
"""
if not me.sock: return
me.disconnected(reason)
Call `line' on each complete line, and `eof' if the connection closed.
Subclasses which attach this class to an I/O-event system should call
- this method when the socket (CONN.sock) is ready for reading.
+ this method when the socket (the `sock' attribute) is ready for reading.
"""
while me.sock is not None:
try:
To be overridden by subclasses to react to a connection being
established.
"""
- pass
+ me.iowatch.connected(me.sock)
def disconnected(me, reason):
"""
To be overridden by subclasses to react to a connection being severed.
"""
- pass
+ me.iowatch.disconnected()
def eof(me):
"""To be overridden by subclasses to handle end-of-file."""
"""To be overridden by subclasses to handle incoming lines."""
pass
+###--------------------------------------------------------------------------
+### I/O loop integration.
+
+class SelIOWatcher (object):
+ """
+ Integration with mLib's I/O event system.
+
+ You can replace this object with a different one for integration with,
+ e.g., glib's main loop, by setting `CONN.iowatcher' to a different object
+ while the CONN is disconnected.
+ """
+
+ def __init__(me, conn):
+ me._conn = conn
+ me._selfile = None
+
+ def connected(me, sock):
+ """
+ Called when a connection is made.
+
+ SOCK is the socket. The watcher must arrange to call `CONN.receive' when
+ data is available.
+ """
+ me._selfile = M.SelFile(sock.fileno(), M.SEL_READ, me._conn.receive)
+ me._selfile.enable()
+
+ def disconnected(me):
+ """
+ Called when the connection is lost.
+ """
+ me._selfile = None
+
+ def iterate(me):
+ """
+ Wait for something interesting to happen, and issue events.
+
+ That is, basically, do one iteration of a main select loop, processing
+ all of the events, and then return. This isn't needed for
+ `TripeCommandDispatcher', but `runservices' wants it.
+ """
+ M.select()
+
+###--------------------------------------------------------------------------
+### Inter-coroutine communication.
+
+class Queue (object):
+ """
+ A queue of things arriving asynchronously.
+
+ This is a very simple single-reader multiple-writer queue. It's useful for
+ more complex coroutines which need to cope with a variety of possible
+ incoming events.
+ """
+
+ def __init__(me):
+ """Create a new empty queue."""
+ me.contents = M.Array()
+ me.waiter = None
+
+ def _wait(me):
+ """
+ Internal: wait for an item to arrive in the queue.
+
+ Complain if someone is already waiting, because this is just a
+ single-reader queue.
+ """
+ if me.waiter:
+ raise ValueError('queue already being waited on')
+ try:
+ me.waiter = Coroutine.getcurrent()
+ while not me.contents:
+ me.waiter.parent.switch()
+ finally:
+ me.waiter = None
+
+ def get(me):
+ """
+ Remove and return the item at the head of the queue.
+
+ If the queue is empty, wait until an item arrives.
+ """
+ me._wait()
+ return me.contents.shift()
+
+ def peek(me):
+ """
+ Return the item at the head of the queue without removing it.
+
+ If the queue is empty, wait until an item arrives.
+ """
+ me._wait()
+ return me.contents[0]
+
+ def put(me, thing):
+ """
+ Write THING to the queue.
+
+ If someone is waiting on the queue, wake him up immediately; otherwise
+ just leave the item there for later.
+ """
+ me.contents.push(thing)
+ if me.waiter:
+ me.waiter.switch()
+
###--------------------------------------------------------------------------
### Dispatching coroutine.
Subclasses must implement a method to handle server responses:
- * response(CODE, *ARGS): CODE is one of the strings 'OK', 'INFO' or
- 'FAIL'; ARGS are the remaining tokens from the server's response.
+ * response(CODE, *ARGS): CODE is one of the strings `OK', `INFO' or
+ `FAIL'; ARGS are the remaining tokens from the server's response.
"""
def __init__(me, words):
terminating response (`OK' or `FAIL') is received or become very
confused.
- Mostly it's better to use the TripeCommandIterator to do this
+ Mostly it's better to use the `TripeCommandIterator' to do this
automatically.
"""
class TripeError (StandardError):
"""
- A tripe command failed with an error (a FAIL code). The args attribute
+ A tripe command failed with an error (a `FAIL' code). The args attribute
contains a list of the server's message tokens.
"""
pass
Iterator interface to a tripe command.
The values returned by the iterator are lists of tokens from the server's
- INFO lines, as processed by the given filter function, if any. The
- iterator completes normally (by raising StopIteration) if the server
- reported OK, and raises an exception if the command failed for some reason.
+ `INFO' lines, as processed by the given filter function, if any. The
+ iterator completes normally (by raising `StopIteration') if the server
+ reported `OK', and raises an exception if the command failed for some reason.
- A TripeError is raised if the server issues a FAIL code. If the connection
- failed, some other exception is raised.
+ A `TripeError' is raised if the server issues a `FAIL' code. If the
+ connection failed, some other exception is raised.
"""
def __init__(me, dispatcher, words, bg = False, filter = None):
"""
Iterator protocol: return the next piece of information from the server.
- INFO responses are filtered and returned as the values of the iteration.
- FAIL and CONNERR responses are turned into exceptions and raised.
- Finally, OK is turned into StopIteration, which should cause a normal end
- to the iteration process.
+ `INFO' responses are filtered and returned as the values of the
+ iteration. `FAIL' and `CONNERR' responses are turned into exceptions and
+ raised. Finally, `OK' is turned into `StopIteration', which should cause
+ a normal end to the iteration process.
"""
thing = me.dcr.switch()
code, rest = thing
return ' '.join(words)
def _keyvals(iter):
- """Return a dictionary formed from the KEY=VALUE pairs returned by the
+ """Return a dictionary formed from the `KEY=VALUE' pairs returned by the
iterator ITER."""
kv = {}
for ww in iter:
def _kwopts(kw, allowed):
"""Parse keyword arguments into options. ALLOWED is a list of allowable
- keywords; raise errors if other keywords are present. KEY = VALUE becomes
- an option pair -KEY VALUE if VALUE is a string, just the option -KEY if
- VALUE is a true non-string, or nothing if VALUE is false.. Insert a `--'
- at the end to stop the parser getting confused."""
+ keywords; raise errors if other keywords are present. `KEY = VALUE'
+ becomes an option pair `-KEY VALUE' if VALUE is a string, just the option
+ `-KEY' if VALUE is a true non-string, or nothing if VALUE is false. Insert
+ a `--' at the end to stop the parser getting confused."""
opts = []
amap = {}
for a in allowed: amap[a] = True
opts.append('--')
return opts
+## Deferral.
+_deferq = []
+def defer(func, *args, **kw):
+ """Call FUNC(*ARGS, **KW) later, in the root coroutine."""
+ _deferq.append((func, args, kw))
+
+def funargstr(func, args, kw):
+ items = [repr(a) for a in args]
+ for k, v in kw.iteritems():
+ items.append('%s = %r' % (k, v))
+ return '%s(%s)' % (func.__name__, ', '.join(items))
+
+def spawn(func, *args, **kw):
+ """Call FUNC, passing ARGS and KW, in a fresh coroutine."""
+ defer(lambda: (Coroutine(func, name = funargstr(func, args, kw))
+ .switch(*args, **kw)))
+
+## Asides.
+_asideq = Queue()
+def _runasides():
+ """
+ Read (FUNC, ARGS, KW) triples from queue and invoke FUNC(*ARGS, **KW).
+ """
+ while True:
+ func, args, kw = _asideq.get()
+ try:
+ func(*args, **kw)
+ except:
+ SYS.excepthook(*SYS.exc_info())
+
+def aside(func, *args, **kw):
+ """Call FUNC(*ARGS, **KW) later, in a non-root coroutine."""
+ defer(_asideq.put, (func, args, kw))
+
class TripeCommandDispatcher (TripeConnection):
"""
Command dispatcher.
This is probably the most important class in this module to understand.
Lines from the server are parsed into tokens. The first token is a code
- (OK or NOTE or something) explaining what kind of line this is. The
+ (`OK' or `NOTE' or something) explaining what kind of line this is. The
`handler' attribute is a dictionary mapping server line codes to handler
functions, which are applied to the words of the line as individual
- arguments. *Exception*: the content of TRACE lines is not tokenized.
+ arguments. *Exception*: the content of `TRACE' lines is not tokenized.
There are default handlers for server codes which respond to commands.
- Commands arrive as TripeCommand instances through the `rawcommand'
+ Commands arrive as `TripeCommand' instances through the `rawcommand'
interface. The dispatcher keeps track of which command objects represent
which jobs, and sends responses on to the appropriate command objects by
- invoking their `response' methods. Command objects don't see the
- BG... codes, because the dispatcher has already transformed them into
- regular codes when it was looking up job code.
+ invoking their `response' methods. Command objects don't see the `BG...'
+ codes, because the dispatcher has already transformed them into regular
+ codes when it was looking up the job tag.
- The dispatcher also has a special response code of its own: CONNERR
+ The dispatcher also has a special response code of its own: `CONNERR'
indicates that the connection failed and the command has therefore been
- lost; the
+ lost. This is sent to all outstanding commands when a connection error is
+ encountered: rather than a token list, it is accompanied by an exception
+ object which is the cause of the disconnection, which may be `None' if the
+ disconnection is expected (e.g., the direct result of a user request).
"""
## --- Infrastructure ---
for i in 'OK', 'INFO', 'FAIL':
me.handler[i] = me._fgresponse
+ def quitp(me):
+ """Should we quit the main loop? Subclasses should override."""
+ return False
+
+ def mainloop(me, quitp = None):
+ """
+ Iterate the I/O watcher until QUITP returns true.
+
+ Arranges for asides and deferred calls to be made at the right times.
+ """
+
+ global _deferq
+ assert _Coroutine.getcurrent() is rootcr
+ Coroutine(_runasides, name = '_runasides').switch()
+ if quitp is None:
+ quitp = me.quitp
+ while not quitp():
+ while _deferq:
+ q = _deferq
+ _deferq = []
+ for func, args, kw in q:
+ func(*args, **kw)
+ me.iowatch.iterate()
+
def connected(me):
"""
Connection hook.
"""
me.queue = M.Array()
me.cmd = {}
+ TripeConnection.connected(me)
def disconnected(me, reason):
"""
Disconnection hook.
If a subclass hooks overrides this method, it must call us; sends a
- special CONNERR code to all incomplete commands.
+ special `CONNERR' code to all incomplete commands.
"""
+ TripeConnection.disconnected(me, reason)
for cmd in me.cmd.itervalues():
cmd.response('CONNERR', reason)
for cmd in me.queue:
def _detach(me, _, tag):
"""
- Respond to a BGDETACH TAG message.
+ Respond to a `BGDETACH' TAG message.
Move the current foreground command to the background.
"""
def _response(me, code, tag, *w):
"""
- Respond to an OK, INFO or FAIL message.
+ Respond to an `OK', `INFO' or `FAIL' message.
If this is a message for a background job, find the tag; then dispatch
- the result to the command object.
+ the result to the command object. This is also called by `_fgresponse'
+ (wth TAG set to `None') to handle responses for foreground commands, and
+ is therefore a useful method to extend or override in subclasses.
"""
if code.startswith('BG'):
code = code[2:]
def rawcommand(me, cmd):
"""
- Submit the TripeCommand CMD to the server, and look after it until it
+ Submit the `TripeCommand' CMD to the server, and look after it until it
completes.
"""
if not me.connectedp():
def add(me, peer, *addr, **kw):
return _simple(me.command(bg = True,
*['ADD'] +
- _kwopts(kw, ['tunnel', 'keepalive', 'cork']) +
+ _kwopts(kw, ['tunnel', 'keepalive',
+ 'key', 'priv', 'cork',
+ 'mobile']) +
[peer] +
list(addr)))
def addr(me, peer):
return _oneline(me.command('ADDR', peer))
- def algs(me):
- return _keyvals(me.command('ALGS'))
+ def algs(me, peer = None):
+ return _keyvals(me.command('ALGS',
+ *((peer is not None and [peer]) or [])))
def checkchal(me, chal):
return _simple(me.command('CHECKCHAL', chal))
def daemon(me):
###--------------------------------------------------------------------------
### Asynchronous commands.
-class Queue (object):
- """
- A queue of things arriving asynchronously.
-
- This is a very simple single-reader multiple-writer queue. It's useful for
- more complex coroutines which need to cope with a variety of possible
- incoming events.
- """
-
- def __init__(me):
- """Create a new empty queue."""
- me.contents = M.Array()
- me.waiter = None
-
- def _wait(me):
- """
- Internal: wait for an item to arrive in the queue.
-
- Complain if someone is already waiting, because this is just a
- single-reader queue.
- """
- if me.waiter:
- raise ValueError('queue already being waited on')
- try:
- me.waiter = Coroutine.getcurrent()
- while not me.contents:
- me.waiter.parent.switch()
- finally:
- me.waiter = None
-
- def get(me):
- """
- Remove and return the item at the head of the queue.
-
- If the queue is empty, wait until an item arrives.
- """
- me._wait()
- return me.contents.shift()
-
- def peek(me):
- """
- Return the item at the head of the queue without removing it.
-
- If the queue is empty, wait until an item arrives.
- """
- me._wait()
- return me.contents[0]
-
- def put(me, thing):
- """
- Write THING to the queue.
-
- If someone is waiting on the queue, wake him up immediately; otherwise
- just leave the item there for later.
- """
- me.contents.push(thing)
- if me.waiter:
- me.waiter.switch()
-
class TripeAsynchronousCommand (TripeCommand):
"""
Asynchronous commands.
and associate the command with the queue. Responses arriving for the
command will be put on the queue as an triple of the form (TAG, CODE, REST)
-- where TAG is an object of your choice, not interpreted by this class,
- CODE is the server's response code (OK, INFO, FAIL), and REST is the list
- of the rest of the server's tokens.
+ CODE is the server's response code (`OK', `INFO', `FAIL', or `CONNERR'),
+ and REST is the list of the rest of the server's tokens.
Using this, you can write coroutines which process many commands (and
possibly other events) simultaneously.
"""Handle a server response by writing it to the caller's queue."""
me.queue.put((me.tag, code, list(stuff)))
-###--------------------------------------------------------------------------
-### Selecting command dispatcher.
-
-class SelCommandDispatcher (TripeCommandDispatcher):
- """
- A command dispatcher which integrates with mLib's I/O-event system.
-
- To use, simply create an instance and run mLib.select in a loop in your
- main coroutine.
- """
-
- def __init__(me, socket):
- """
- Create an instance; SOCKET is the admin socket to connect to.
-
- Note that no connection is made initially.
- """
- TripeCommandDispatcher.__init__(me, socket)
- me.selfile = None
-
- def connected(me):
- """Connection hook: wires itself into the mLib select machinery."""
- TripeCommandDispatcher.connected(me)
- me.selfile = M.SelFile(me.sock.fileno(), M.SEL_READ, me.receive)
- me.selfile.enable()
-
- def disconnected(me, reason):
- """Disconnection hook: removes itself from the mLib select machinery."""
- TripeCommandDispatcher.disconnected(me, reason)
- me.selfile = None
-
###--------------------------------------------------------------------------
### Services.
"""
pass
-class TripeServiceManager (SelCommandDispatcher):
+class TripeServiceManager (TripeCommandDispatcher):
"""
A command dispatcher with added handling for incoming service requests.
There is usually only one instance of this class, called svcmgr. Some of
the support functions in this module assume that this is the case.
- To use, run mLib.select in a loop until the quitp method returns true;
+ To use, run `mLib.select' in a loop until the quitp method returns true;
then, in a non-root coroutine, register your services by calling `add', and
then call `running' when you've finished setting up.
- The instance handles server service messages SVCJOB, SVCCANCEL and
- SVCCLAIM. It maintains a table of running services. Incoming jobs cause
- the service's `job' method to be invoked; SVCCANCEL sends a
- TripeJobCancelled exception to the handler coroutine, and SVCCLAIM causes
- the relevant service to be deregistered.
+ The instance handles server service messages `SVCJOB', `SVCCANCEL' and
+ `SVCCLAIM'. It maintains a table of running services. Incoming jobs cause
+ the service's `job' method to be invoked; `SVCCANCEL' sends a
+ `TripeJobCancelled' exception to the handler coroutine, and `SVCCLAIM'
+ causes the relevant service to be deregistered.
There is no base class for jobs, but a job must implement two methods:
SOCKET is the administration socket to connect to.
"""
- SelCommandDispatcher.__init__(me, socket)
+ TripeCommandDispatcher.__init__(me, socket)
me.svc = {}
me.job = {}
me.runningp = False
me._quitp = 0
def addsvc(me, svc):
- """Register a new service; SVC is a TripeService instance."""
+ """Register a new service; SVC is a `TripeService' instance."""
assert svc.name not in me.svc
me.svcclaim(svc.name, svc.version)
me.svc[svc.name] = svc
process can quit without anyone caring).
"""
return me._quitp or (me.runningp and ((not me.svc and not me.job) or
- not me.selfile))
+ not me.sock))
def quit(me):
"""Forces the quit flag (returned by quitp) on."""
The NAME and VERSION are passed on to the server. The CMDTAB is a
dictionary mapping command names (in lowercase) to command objects.
- If the CMDTAB doesn't have entries for commands HELP and QUIT then defaults
- are provided.
+ If the CMDTAB doesn't have entries for commands `HELP' and `QUIT' then
+ defaults are provided.
TripeService itself is mostly agnostic about the nature of command objects,
but the TripeServiceJob class (below) has some requirements. The built-in
"""
Job handler coroutine.
- A standard TripeService invokes a TripeServiceJob for each incoming job
- request, passing it the jobid, command and arguments, and a command
- object. The command object needs the following attributes.
+ A standard `TripeService' invokes a `TripeServiceJob' for each incoming job
+ request, passing it the jobid, command and arguments, and a command object.
+ The command object needs the following attributes.
usage A usage list (excluding the command name) showing
arguments and options.
run(*ARGS) Function to react to the command with ARGS split into
separate arguments. Invoked in a coroutine. The
- svcinfo function (not the TripeCommandDispatcher
- method) may be used to send INFO lines. The function
- may raise TripeJobError to send a FAIL response back,
- or TripeSyntaxError to send a generic usage error.
- TripeJobCancelled exceptions are trapped silently.
- Other exceptions are translated into a generic
- internal-error message.
+ `svcinfo function (not the `TripeCommandDispatcher'
+ method) may be used to send `INFO' lines. The
+ function may raise `TripeJobError' to send a `FAIL'
+ response back, or `TripeSyntaxError' to send a
+ generic usage error. `TripeJobCancelled' exceptions
+ are trapped silently. Other exceptions are
+ translated into a generic internal-error message.
This class automatically takes care of sending some closing response to the
job, and for informing the service manager that the job is completed.
The job is created with id JID, for service SVC, processing command name
CMD (which the service resolved into the command object COMMAND, or
- None), and with the arguments ARGS.
+ `None'), and with the arguments ARGS.
"""
Coroutine.__init__(me)
me.jid = jid
def svcinfo(*args):
"""
- If invoked from a TripeServiceJob coroutine, sends an INFO line to the
+ If invoked from a TripeServiceJob coroutine, sends an `INFO' line to the
job's sender, automatically using the correct job id.
"""
svcmgr.svcinfo(Coroutine.getcurrent().jid, *args)
svcmgr.running()
svcmgr = TripeServiceManager(None)
-_spawnq = []
def runservices(socket, tab, init = None, setup = None, daemon = False):
"""
Function to start a service provider.
(NAME, VERSION, COMMANDS)
- or a service object (e.g., a TripeService instance).
+ or a service object (e.g., a `TripeService' instance).
COMMANDS is a dictionary mapping command names to tuples
(MIN, MAX, USAGE, FUNC)
- of arguments for a TripeServiceCommand object.
+ of arguments for a `TripeServiceCommand' object.
If DAEMON is true, then the process is forked into the background before we
start. If INIT is given, it is called in the main coroutine, immediately
quit.
"""
- global _spawnq
svcmgr.socket = socket
svcmgr.connect()
svcs = []
M.daemonize()
if init is not None:
init()
- Coroutine(_setupsvc).switch(svcs, setup)
- while not svcmgr.quitp():
- for cr, args, kw in _spawnq:
- cr.switch(*args, **kw)
- _spawnq = []
- M.select()
-
-def spawn(cr, *args, **kw):
- """
- Utility for spawning coroutines.
-
- The coroutine CR is made to be a direct child of the root coroutine, and
- invoked by it with the given arguments.
- """
- cr.parent = rootcr
- _spawnq.append((cr, args, kw))
+ spawn(_setupsvc, svcs, setup)
+ svcmgr.mainloop()
###--------------------------------------------------------------------------
### Utilities for services.
"""
Iterator protocol: return the next option.
- If we've run out, raise StopIteration.
+ If we've run out, raise `StopIteration'.
"""
if len(me.args) == 0 or \
len(me.args[0]) < 2 or \
"""
Return the argument for the most recent option.
- If none is available, raise TripeSyntaxError.
+ If none is available, raise `TripeSyntaxError'.
"""
if len(me.args) == 0:
raise TripeSyntaxError
~mdw / tripe / blobdiff