File: simple.py

package info (click to toggle)
kiwi 1.9.22-1
  • links: PTS, VCS
  • area: main
  • in suites: lenny
  • size: 11,908 kB
  • ctags: 5,549
  • sloc: python: 15,779; ansic: 193; xml: 77; makefile: 57; sh: 17
file content (60 lines) | stat: -rw-r--r-- 1,962 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import gobject

from kiwi.tasklet import Tasklet, WaitForTimeout, WaitForMessages, Message, \
     WaitForTasklet, get_event

## ----------------------------
## And here's an example...
## ----------------------------
class _CountSomeNumbers2(Tasklet):
    '''Counts numbers with at random time spacings'''

    def __init__(self, count, timeout):
        self.count = count
        self.timeout = timeout
        Tasklet.__init__(self)

    def run(self):
        '''Execute the task.'''
        import random
        for i in xrange(self.count):
            print ">> _count_some_numbers2", i
            yield (WaitForTimeout(random.randint(70, self.timeout)),
                   WaitForMessages(accept='quit'))
            event = get_event()
            if isinstance(event, Message) and event.name == 'quit':
                ## this would be the place to do some cleanup.
                return
        raise StopIteration(self.count*2)

def _count_some_numbers1(count):
    '''Counts numbers with at fixed time spacings'''
    timeout = WaitForTimeout(500)
    for i in xrange(count):
        print "_count_some_numbers1", i
        task2 = _CountSomeNumbers2(10, 70)
        yield timeout, task2
        event = get_event()
        if event is timeout:
            print ">>> Got tired of waiting for task!! Canceling!"
            ## send a message asking the tasklet to stop
            yield Message('quit', dest=task2)
        elif isinstance(event, WaitForTasklet):
            print ">>> task returned %r, good task!" % event.retval
            ## restart timeout from scratch, otherwise it keeps
            ## running and we end up giving the next task too little
            ## time.
            timeout.restart()
        else:
            assert False, "strange event"

    raise SystemExit

def _test():
    '''a simple test/example'''
    Tasklet(_count_some_numbers1(5))
    gobject.MainLoop().run()


if __name__ == '__main__':
    _test()