EventLoop.call(log.append, 4) |
EventLoop.call(log.append, 4) |
|
|
class IdleTimer(trellis.Component): |
class IdleTimer(trellis.Component): |
trellis.values( |
trellis.variable.attributes( |
idle_for = NOT_YET, |
|
idle_timeout = 20, |
idle_timeout = 20, |
busy = False, |
busy = False, |
) |
) |
trellis.rules( |
idle_for = trellis.compute( |
idle_for = lambda self: |
lambda self: self.idle_for.begins_with(not self.busy), |
self.idle_for.begins_with(not self.busy) |
initially=NOT_YET |
) |
) |
|
trellis.maintain() # XXX should be perform |
def alarm(self): |
def alarm(self): |
if self.idle_for[self.idle_timeout] and EventLoop.running: |
if self.idle_for[self.idle_timeout] and EventLoop.running: |
log.append(5) |
log.append(5) |
EventLoop.stop() |
EventLoop.stop() |
alarm = trellis.rule(alarm) |
|
|
|
it = IdleTimer() |
it = IdleTimer() |
EventLoop.run() |
EventLoop.run() |
|
|
|
|
|
|
|
|
class TestLinks(unittest.TestCase): |
class TestLinks(unittest.TestCase): |
|
|
def setUp(self): |
def setUp(self): |
self.ctrl.rollback_to(sp) |
self.ctrl.rollback_to(sp) |
self.assertEqual(self.ctrl.queues, {}) |
self.assertEqual(self.ctrl.queues, {}) |
|
|
|
def testNestedReadOnly(self): |
|
log = [] |
|
def aRule(): |
|
log.append(trellis.ctrl.readonly) |
|
return 1 |
|
c1 = trellis.Cell(aRule) |
|
c2 = trellis.Cell(lambda: c1.value * aRule()) |
|
c3 = trellis.ObserverCell(lambda: c2.value) |
|
self.assertEqual(log, [True, True]) |
|
|
|
|
|
|
def testPauseAndCall(self): |
def testPauseAndCall(self): |
log = [] |
log = [] |
class TaskExample(trellis.Component): |
class TaskExample(trellis.Component): |
trellis.values( |
trellis.variable.attributes( |
start = False, |
start = False, |
stop = False |
stop = False |
) |
) |
log.append("waiting to stop") |
log.append("waiting to stop") |
yield activity.Pause |
yield activity.Pause |
|
|
d(activity.task) |
activity.task() |
def demo(self): |
def demo(self): |
yield self.wait_for_start() |
yield self.wait_for_start() |
log.append("starting") |
log.append("starting") |