treestream/example.py at master · GambitResearch/treestream · GitHub

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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
from __future__ import print_function

from treestream import RedisTreeWriter

from threading import Thread, Lock
from time import sleep
from random import random, randrange

from treestream import (
	RedisTreeReader,
	RedisTreeWriter,
)

print_lock = Lock()

def start_thread(func):
	thread = Thread(target=func)
	thread.setDaemon(True)
	thread.start()
	return thread


def tprint(*args):
	with print_lock:
		print(*args)


class MyReader(RedisTreeReader):
	def  __init__(self, reader_name, **kwargs):
		super(MyReader, self).__init__(**kwargs)
		self.reader_name = reader_name

	def handle_update(self, timestamp, tree_path, old_value, value):
		tprint(self.reader_name, tree_path, '->', value)

	def handle_delete(self, timestamp, tree_path, old_value):
		tprint(self.reader_name, tree_path, '-> None')

	def handle_delete_subtree(self, timestamp, tree_path, deleted_subtree):
		pass

	def handle_sync_lost(self):
		tprint(self.reader_name, 'sync-lost')

	def handle_synced(self, tree):
		tprint(self.reader_name, 'synced')
		with print_lock:
			tree.pretty_print(depth=4)


def writer1():
	'''updates the value of a leaf node every ~100ms'''
	wr = RedisTreeWriter()
	start_thread(wr.run)

	wr.delete(()) # clear the tree

	def loop():
		while True:
			tree_path = [str(randrange(3)) for _ in range(3)]
			value = str(randrange(1000,10000))
			wr.update(tree_path, value)
			sleep(random() * 0.1)
	start_thread(loop)
	return wr


def writer2():
	'''deletes a subtree every ~5s'''
	wr = RedisTreeWriter()
	start_thread(wr.run)

	def loop():
		while True:
			tree_path_prefix = [str(randrange(3)) for _ in range(randrange(4))]
			tprint('DELETING', tree_path_prefix)
			wr.delete(tree_path_prefix)
			sleep(random() * 5.0)
	start_thread(loop)
	return wr


def reader1():
	rd = MyReader('reader1')
	start_thread(rd.run)
	return rd


def reader2():
	rd = MyReader('reader2')
	start_thread(rd.run)
	return rd


def main():
	wr1 = writer1()
	wr2 = writer2()
	rd1 = reader1()
	sleep(5.0)
	tprint('reader2 starting')
	rd2 = reader2() # start later, so that we get a sync first
	sleep(5.0)
	wr1.stop()
	wr2.stop()
	rd1.stop()
	rd2.stop()

if __name__ == '__main__':
	main()