gocept.async 0.3.3

详情

通常希望异步处理计算。在 lovely.remotetask 出现之前，由于事务集成问题，在 Zope 3 应用程序中实现这一点并不容易。

gocept.async 让任务变得更简单

>>> import gocept.async
>>> @gocept.async.function(service='events')
... def heavy_computing(a, b):
...     print "Computing", a, "+", b, "=", a + b

装饰器 gocept.async.function 精确接受一个参数，即 lovely.remotetask.interfaces.ITaskService 实用类的名称。请注意，gocept.async 本身并不定义任何任务服务。

测试设置

请注意，装饰的函数必须有一个可导入的模块才能使用

>>> import gocept.async.tests
>>> heavy_computing.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.heavy_computing = heavy_computing

在这个测试中，我们定义了一个名为 events 的任务服务

>>> import zope.component
>>> import lovely.remotetask
>>> import lovely.remotetask.interfaces
>>> import lovely.remotetask.processor
>>> sm = zope.component.getSiteManager()
>>> getRootFolder()['tasks'] = tasks = lovely.remotetask.TaskService()
>>> tasks.processorFactory = lovely.remotetask.processor.MultiProcessor
>>> tasks.processorArguments = {'maxThreads': 1}
>>> sm.registerUtility(
...     tasks, lovely.remotetask.interfaces.ITaskService, name='events')

基本知识

当装饰的函数被调用时，它不返回任何内容

>>> heavy_computing(2, 7)

当我们开始处理任务服务时，会调用该函数

>>> gocept.async.tests.process()
Computing 2 + 7 = 9

当用户登录时调用函数，该函数将以该用户身份调用（注意，可能需要手动创建安全代理以在异步函数中启用安全功能。）

>>> @gocept.async.function('events')
... def who_am_i():
...     print gocept.async.task.TaskDescription.get_principal()
...
>>> who_am_i.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.who_am_i = who_am_i
>>> who_am_i()
>>> gocept.async.tests.process()

现在登录

>>> gocept.async.tests.login('zope.user')
>>> who_am_i()
>>> gocept.async.tests.process()
zope.user
>>> gocept.async.tests.logout()

如果在一个已经异步化的过程中调用异步函数，则函数会立即被调用

>>> @gocept.async.function(service='events')
... def call_another():
...     print "Before"
...     heavy_computing(1, 2)
...     print "After"
>>> call_another.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.call_another = call_another
>>> call_another()
>>> gocept.async.tests.process()
Before
Computing 1 + 2 = 3
After

有一个辅助函数可以测试当前是否正在执行异步操作

>>> gocept.async.is_async()
False
>>> @gocept.async.function(service='events')
... def is_async_test():
...     print gocept.async.is_async()
>>> is_async_test.undecorated.__module__ = 'gocept.async.tests'
>>> gocept.async.tests.is_async_test = is_async_test
>>> is_async_test()
>>> gocept.async.tests.process()
True

拆除

>>> sm.registerUtility(
...     tasks, lovely.remotetask.interfaces.ITaskService, name='events')
>>> del gocept.async.tests.heavy_computing
>>> del gocept.async.tests.who_am_i
>>> del gocept.async.tests.call_another
>>> del gocept.async.tests.is_async_test

变更