-
Notifications
You must be signed in to change notification settings - Fork 1
/
element.py
62 lines (45 loc) · 1.29 KB
/
element.py
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
61
62
# -*- coding: utf-8 -*-
"""
Building block
"""
class Element(object):
def __init__(self):
self.sinks = []
def put(self, data):
for sink in self.sinks:
sink.put(data)
def start(self):
pass
def stop(self):
pass
def link(self, sink):
if hasattr(sink, 'put') and callable(sink.put):
self.sinks.append(sink)
else:
raise ValueError('Not implement put() method')
def unlink(self, sink):
self.sinks.remove(sink)
def pipeline(self, *args):
source = self
for sink in args:
source.link(sink)
source = sink
return self
def pipeline_start(self):
def recursive_start_sink(s):
# start downstream first
if hasattr(s, 'sinks'):
for sink in s.sinks:
recursive_start_sink(sink)
s.start()
recursive_start_sink(self)
recursive_start = pipeline_start
def pipeline_stop(self):
def recursive_stop_sink(s):
# stop upstream first
s.stop()
if hasattr(s, 'sinks'):
for sink in s.sinks:
recursive_stop_sink(sink)
recursive_stop_sink(self)
recursive_stop = pipeline_stop