Index: main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/async.py
===================================================================
--- main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/async.py	(revision 9081)
+++ main/waeup.kofa/branches/uli-zc-async/src/waeup/kofa/async.py	(revision 9082)
@@ -1,24 +1,45 @@
-import datetime
+## $Id$
+##
+## Copyright (C) 2012 Uli Fouquet & Henrik Bettermann
+## This program is free software; you can redistribute it and/or modify
+## it under the terms of the GNU General Public License as published by
+## the Free Software Foundation; either version 2 of the License, or
+## (at your option) any later version.
+##
+## This program is distributed in the hope that it will be useful,
+## but WITHOUT ANY WARRANTY; without even the implied warranty of
+## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+## GNU General Public License for more details.
+##
+## You should have received a copy of the GNU General Public License
+## along with this program; if not, write to the Free Software
+## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+##
+"""Components for asynchronous job (task) handling, mainly based on
+:mod:`zc.async`.
+"""
 import grok
-import os
 import time
+import transaction
 import zc.async.dispatcher
-from cgi import escape
-from pytz import utc
-from zc.async.interfaces import IQueue
+import zc.async.utils
+import zope.component
 from zc.async.job import Job as AsyncJob
 from zc.async.queue import getDefaultQueue
-from waeup.kofa.interfaces import IUniversity
-from hurry.jqueryui import jqueryui, base, blitzer, ui_lightness, humanity
-from hurry.jquery import jquery
-from zc.async.interfaces import COMPLETED, NEW, PENDING, ACTIVE, ASSIGNED
-
+from zc.async.interfaces import IJob, ObjectAdded
+from ZODB.utils import u64
+from zope.component.hooks import getSite
+from zope.event import notify
+from waeup.kofa.interfaces import (
+    IJobManager, IProgressable, IJobContainer, WAEUP_KEY,
+    )
 
 def compute(num):
+    """A convenience function to test asynchronous jobs.
+    """
     start = time.time()
     duration = 60
     end = start + duration
     print "MyJob starts computation at ", start
-    #from zc.async.local import getJob, getQueue
     myjob = zc.async.local.getJob()
     print "MyJob's job: ", zc.async.local.getJob()
@@ -41,209 +62,211 @@
     return num * 2
 
-def get_next_num(context):
-    num = 1
-    print "KEYS: ", [x for x in context['jobs'].keys()]
-    while str(num) in [x for x in context['jobs'].keys()]:
-        num += 1
-    print "NEXT: ", str(num)
-    return str(num)
-
-class Job(AsyncJob):
+#
+# Content components
+#
+class ProgressableJob(AsyncJob):
+    """A job that can indicate its progress via a `percent` attribute.
+    """
+    grok.implements(IJob, IProgressable)
+    grok.provides(IJob)
+
     percent = None
-    pass
-
-grok.templatedir('.') #os.path.dirname(__file__))
-class ASyncView(grok.View):
-    grok.context(IUniversity)
-    grok.name('async')
-    grok.template('async')
-
-    def update(self,  start=None, DELETE=None, REMOVE=None, *args, **kw):
-        humanity.need()
-        jqueryui.need()
-        # the image was created via http://ajaxload.info/
-        self.spinner_url = self.url(
-            grok.getSite()) + '/@@/waeup.kofa.browser/square_circles_black.gif'
-        print args, kw
-        self.dispatcher = zc.async.dispatcher.get()
-        site = grok.getSite()
-        newq = site.__parent__._p_jar.root()['zc.async'].items()[0][1]
-        print "NEW: ", newq, list(newq)
-
-
-        print "DISP LEN: ", dir(self.dispatcher), self.dispatcher.getStatusInfo(), self.dispatcher.getStatistics(), self.dispatcher.jobs.items() # self.dispatcher.getJobInfo()
-        self.queue = IQueue(self.context)
-        self.dispatcher2 = self.queue.dispatchers[zc.async.instanceuuid.UUID]
-        print "JOBS: ", self.dispatcher.jobs.keys(), self.dispatcher.jobs.values()
-        print "DISPATCHERS: ", self.dispatcher, self.dispatcher2
-        print "QUEUES: ", self.dispatcher.queues
-        print "QUEUE1: ", list(self.dispatcher.queues[''])
-        print "QUEUE: ", list(self.queue)
-        #print "DISPATCHERS: ", self.queue.dispatchers.items()
-        print "DEFAULT: ", getDefaultQueue(self.context), len(getDefaultQueue(self.context))
-        print "DELETE: ", DELETE
-        if DELETE is not None:
-            if DELETE in self.context['jobs']:
-                job = self.context['jobs'][DELETE]
-                from zc.async.interfaces import COMPLETED, NEW, PENDING
-                if job.status in (COMPLETED, NEW):
-                    del self.context['jobs'][DELETE]
-                    print "DELETED %s" % job
+
+
+def get_job_id(persisted_job):
+    """Get the object id of an already persisted job.
+
+    The `persisted_job` must provide a `_p_oid` attribute.
+    """
+    job_id = u64(persisted_job._p_oid)
+    return "%s" % job_id
+
+class JobContainer(grok.Container):
+    """A container for :class:`IKofa` jobs.
+    """
+    grok.implements(IJobContainer)
+
+
+class JobManager(grok.GlobalUtility):
+    """A manager for asynchronous running jobs (tasks).
+
+    Registered as a global utility for the
+    `waeup.kofa.interfaces.IJobManager` interface.
+
+    This is the central location for managing asynchronous running
+    jobs/tasks.
+
+    It works roughly like this: for usual tasks it looks up some
+    JobContainer installed in a ZODB database root (the installation
+    can happen during startup; see the respective installer classes
+    and functions in this module) and then interacts with this
+    JobContainer.
+
+    The optional `site` parameter for most methods in here serves for
+    finding the databases' roots. It is sufficient to pass any
+    persisted object (or more precisely: some object with a valid
+    ``_p_jar__`` attribte). As long as some site was already set (for
+    instance during regular requests), the site is looked up
+    automatically and you don't have to pass the `site` parameter
+    then. So, in most cases you won't have to give a `site` parameter.
+    """
+    grok.implements(IJobManager)
+    grok.provides(IJobManager)
+
+    def _get_site(self, site):
+        if site is None:
+            site = getSite()
+            if site is None:
+                raise LookupError('cannot find a valid site')
+        return site
+
+    def _get_jobs_container(self, site):
+        # in fact we need _some_ persisted object, not necessarily a site
+        return site._p_jar.root()[WAEUP_KEY]
+
+    def put(self, job, site=None):
+        """Start the `job` and store it in local `site`.
+
+        The `job` must be an `IJob` instance.
+
+        It will be put into the default queue and then stored in local
+        site. The status of the job can be seen immediately in
+        `job.status`.
+
+        Please specify special treatments like `begin_after` or
+        `begin_by` by setting the respectives attributes of the job
+        itself.
+        """
+        site = self._get_site(site)
+        container = self._get_jobs_container(site)
+        transaction.begin()
+        queue = getDefaultQueue(site)
+        new_job = queue.put(job)
+        job_id = get_job_id(new_job)
+        container[job_id] = new_job
+        transaction.commit()
+        return job_id
+
+    def get(self, job_id, site=None):
+        """Get the job with `job_id` from local `site`.
+
+        If `job_id` cannot be found, ``None`` is returned. This
+        suitable e.g. when used with a traverser.
+        """
+        site = self._get_site(site)
+        container = self._get_jobs_container(site)
+        return container.get(job_id, None)
+
+    def jobs(self, site=None):
+        """Get all stored jobs as an iterable.
+
+        Result provides tuples (JOB_ID, JOB_OBJECT).
+        """
+        site = self._get_site(site)
+        container = self._get_jobs_container(site)
+        for job_id, job in container.items():
+            yield (job_id, job)
+
+    def remove(self, job_id, site=None):
+        """Remove job with `job_id` from local job container.
+
+        If no such job can be found this is silently ignored.
+
+        Please note: removing a job from the job container does not
+        mean to stop its execution (if it wasn't started yet or is
+        currently running).
+        """
+        site = self._get_site(site)
+        container = self._get_jobs_container(site)
+        if job_id in container.keys():
+            del container[job_id]
+        return
+
+    def start_test_job(self):
+        """Start a test job.
+
+        A method for testing the general asynchronous functionality of
+        waeup.kofa. The test job started here executes the local
+        :func:`compute` function with ``23`` as argument.
+        """
+        job = AsyncJob(compute, 23)
+        job_id = self.put(job)
+        return job_id
+
+class JobManagerTraverser(grok.Traverser):
+    """A traverser for the global ``IJobManager``.
+
+    Looks up jobs by job_id and returns the respective job if it
+    can be found.
+    """
+    grok.context(IJobManager)
+
+    def traverse(self, name):
+        # ``None`` if name cannot be found.
+        return self.context.get(name)
+
+##
+## configuration (permissions, subscribers, etc.) ...
+##
+
+class ViewJobs(grok.Permission):
+    grok.name('waeup.viewJobs')
+
+class ManageJobs(grok.Permission):
+    grok.name('waeup.manageJobs')
+
+class JobContainerInstaller(object):
+    """Install a JobContainer in root of given DB.
+
+    Instances of this installer can be called when a Zope instance
+    comes up (i.e. when an IDatabaseOpenedEvent was triggered).
+
+    It looks for some database named as in `db_name` and installs a
+    job container in the root of this database (``None`` by default)
+    if it does not exist already.
+    """
+    def __init__(self, db_name=None):
+        # This IDatabaseOpenedEvent will be from zope.app.appsetup if that
+        # package is around
+        zope.component.adapter(zc.async.interfaces.IDatabaseOpenedEvent)(self)
+        self.db_name = db_name
+        return
+
+    def __call__(self, ev):
+        db = ev.database
+        tm = transaction.TransactionManager()
+        conn = db.open(transaction_manager=tm)
+        tm.begin()
+        try:
+            try:
+                root = conn.root()
+                if WAEUP_KEY in root:
+                    return
+                if self.db_name is not None:
+                    other = conn.get_connection(self.db_name)
+                    container = other.root()[WAEUP_KEY] = JobContainer()
+                    other.add(container)
                 else:
-                    if job in self.queue:
-                        print "JOB IN ", self.queue, "!!!!!!!!!!!!!!!!!!"
-                        print "QUEUE BEFORE: ", list(self.queue)
-                        try:
-                            self.queue.remove(job)
-                        except:
-                            "COULDN'T REMOVE ANYWAY!!"
-                        print "QUEUE AFTER: ", list(self.queue)
-                        self.queue.pull()
-                        print "QUEUE AFTER2: ", list(self.queue)
-                    #job_queue = job.queue
-                    #agent = job.agent
-                    #print "JOB AGENT: ", agent, dir(agent)
-                    #print "JOB AGENT QUEUE: ", agent.queue, list(agent.queue)
-                    #print "JOB QUEUE: ", job_queue, list(job_queue)
-                    #if job in list(job_queue) or True:
-                    #    agent.remove(job)
-                    #    #job.queue.remove(job)
-                    #    print "REMOVED %s from queue %s" % (
-                    #        job, job_queue)
-                    else:
-                        print "CANT FIND JOB in QUEUE"
-
-        #import pdb; pdb.set_trace()
-        if start:
-            start_time = datetime.datetime.utcnow()
-            start_time += datetime.timedelta(seconds=10)
-            start_time = utc.localize(start_time)
-            print "START"
-            print "STARTTIME: ", start_time
-            import transaction
-            transaction.begin()
-            job = self.queue.put(Job(compute, 21),
-                                 begin_after=start_time)
-            job.annotations['percent'] = 0.0
-            print " started: ", job.status, type(job)
-            if 'jobs' not in self.context:
-                self.context['jobs'] = grok.Container()
-            job_id = get_next_num(self.context)
-            print "JOB ID: ", job_id, dir(job)
-            self.context['jobs'][job_id] = job
-            transaction.commit()
-            time.sleep(0.5)
-            self.redirect(self.url())
-
-    def job_status(self):
-        if 'jobs' not in self.context:
-            return ''
-        result = ""
-        #for num, job in enumerate(self.queue):
-        for num, job in self.context['jobs'].items():
-            result += '<div><form>'
-            if job.status in ('completed-status', 'new-status'):
-                result += '<a href="?DELETE=%s">delete</a>' %(
-                    num)
-            else: # job.status == 'pendig-status' or (
-                # job.status == 'active-status'):
-                result += '<a href="?DELETE=%s">remove from queue</a>' % (
-                    num)
-            result += '<input type="checkbox" name="remove" />%s %s, %s, %r</form></div>' % (
-                num,
-                escape('%r' % job),
-                job.status,
-                dict(job.annotations))
-            print "job in queue: ", job, type(job)
-        return result
-
-    def NOrender(self):
-        print "RENDER: ", self.namespace().items(), type(self.namespace)
-        return grok.PageTemplate('Hallo!')
-        #print "RENDER: ", self.static['square_circles_black.gif']
-        #return r'''
-        return '''
-<html>
-<head>
-<title>Async testing</title>
-<script type="text/javascript">
-$(document).ready(function() {
-  var myLoadingDiv = $("#progressbar");
-  var mySpinner = $("#loadspinner");
-  myLoadingDiv.progressbar({disabled:true});
-  function update() {
-  $.getJSON("status?",  {jobid: "4",}, function(data) {
-    if (data == null) {
-      myLoadingDiv.progressbar("disable");
-      myLoadingDiv.progressbar("value", 0);
-      $("#msg").html("Syncing...");
-      clearInterval(intervalID);
-      //alert('No Data'):
-      }
-    else {
-      myLoadingDiv.progressbar("enable");
-      myLoadingDiv.progressbar("value", data.percent);
-      $("#msg").html(data.msg + ' (' + data.status + ') ');
-      //alert(data.msg);
-      if (data.percent == 100) {
-        clearInterval(intervalID);
-        }
-      }
-  });}
-  intervalID = setInterval(update, 500);
-  //update();
-});
-</script>
-</head>
-<body>
-<h1>Async stuff</h1>
-<div>Dispatcher: %s</div>
-<div>Queue: %s</div>
-<div>Jobs: %s</div>
-<form method="POST">
-  <input type="submit" name="reload" value="reload" />
-  <input type="submit" name="start" value="start up" />
-</form>
-<div id="progressbar"
-     class="ui-progressbar ui-widget ui-widget-content ui-corner-all">
-       <div id="msg">Hi there</div>
-</div>
-<div id="loadspinner">
-</div>
-</body>
-</htlm>
-''' % (
-   escape('%r' % self.dispatcher),
-   escape('%r' % self.queue),
-   self.job_status(),
-   )
-
-
-class ASyncJSON(grok.JSON):
-    grok.context(IUniversity)
-
-    def status(self, jobid):
-        print "request for ", jobid
-        msg = 'please wait...'
-        job = self.context['jobs'].get(jobid, None)
-        if not job:
-            return
-        percent = job.annotations.get('percent', None)
-        if not percent:
-            return
-        percent = int(percent+0.5)
-        print "percent: ", percent
-        if percent >= 100:
-            percent = 100
-            msg = 'done.'
-        status_map = {
-            NEW: 'new',
-            COMPLETED: 'completed',
-            PENDING: 'pending',
-            ACTIVE: 'active',
-            ASSIGNED: 'assigned',
-            }
-        return {'msg': msg,
-                'percent': percent,
-                'status': status_map.get(job.status, 'unknown'),
-                }
+                    container = JobContainer()
+                root[WAEUP_KEY] = container
+                notify(ObjectAdded(container, root, WAEUP_KEY))
+                tm.commit()
+                zc.async.utils.log.info('job container added')
+            except:
+                tm.abort()
+                raise
+        finally:
+            conn.close()
+        return
+
+#: Can be used as event subscriber from ZCML; installs a job container
+#: in default database (named ``''``) root.
+job_container_installer = JobContainerInstaller()
+
+#: An installer instance that installs a job container in a ZODB
+#: called ``async`` - this name is used in several components of the
+#: `zc.async` package we don't want to reimplement here.
+#:
+#: To use this installer as an event subscriber by ZCML, make sure the
+#: instance provides also a ZODB called `async`.
+multidb_job_container_installer = JobContainerInstaller(db_name='async')
