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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
"""
Rhizi websocket web API
"""
import json
import logging
from socketio.mixins import BroadcastMixin
from socketio.namespace import BaseNamespace
import traceback
from model.graph import Attr_Diff, Topo_Diff
from rz_kernel import RZDoc_Exception__not_found
log = logging.getLogger('rhizi')
class WebSocket_Graph_NS(BaseNamespace, BroadcastMixin):
"""
Rhizi '/graph' websocket namespace
"""
def __init__(self, *args, **kw):
super(WebSocket_Graph_NS, self).__init__(*args, **kw)
def __context__common(self, json_dict):
"""
build a common rquest context to pass along with a kernel diff commit
"""
rzdoc_name = json_dict['rzdoc_name']
return {'rzdoc_name': rzdoc_name}
def _log_conn(self, prefix_msg):
rmt_addr = self.environ['REMOTE_ADDR']
rmt_port = self.environ['REMOTE_PORT']
sid = self.environ['socketio'].sessid
log.info('ws: %s: sid: %s, remote-socket: %s:%s' % (prefix_msg, sid, rmt_addr, rmt_port))
def _on_rzdoc_subscribe_common(self, data_dict, is_subscribe=None):
rzdoc_name_raw = data_dict['rzdoc_name']
# FIXME: non-flask dep. sanitization
# rzdoc_name = sanitize_input__rzdoc_name(rzdoc_name_raw)
rzdoc_name = rzdoc_name_raw
rmt_addr = self.environ['REMOTE_ADDR']
rmt_port = self.environ['REMOTE_PORT']
remote_socket_addr = (rmt_addr, rmt_port)
socket = self.socket
kernel = self.request.kernel
msg_name = 'rzdoc_subscribe' if is_subscribe else 'rzdoc_unsubscribe'
try:
if is_subscribe:
kernel.rzdoc__reader_subscribe(remote_socket_addr=remote_socket_addr,
rzdoc_name=rzdoc_name,
socket=socket)
self.ack(msg_name)
else:
kernel.rzdoc__reader_unsubscribe(remote_socket_addr=remote_socket_addr,
rzdoc_name=rzdoc_name,
socket=socket)
self.ack(msg_name)
except RZDoc_Exception__not_found:
self.nak(msg_name)
def ack(self, acked_msg_name):
self.emit('ack', acked_msg_name)
def nak(self, acked_msg_name):
self.emit('nak', acked_msg_name)
def multicast_msg(self, msg_name, *args):
self.socket.server.log_multicast(msg_name)
try:
super(WebSocket_Graph_NS, self).broadcast_event_not_me(msg_name, *args)
except Exception as e:
log.error(e.message)
log.error(traceback.print_exc())
def on_diff_commit__topo(self, json_data):
# FIXME: sanitize input
json_dict = json.loads(json_data)
topo_diff = Topo_Diff.from_json_dict(json_dict['topo_diff'])
log.info('ws: rx: topo diff: ' + str(topo_diff))
ctx = self.__context__common(json_dict)
kernel = self.request.kernel
topo_diff, commit_ret = kernel.diff_commit__topo(topo_diff, ctx)
# handle serialization
topo_diff_dict = topo_diff.to_json_dict()
assert Topo_Diff.Commit_Result_Type == type(commit_ret)
return self.multicast_msg('diff_commit__topo', topo_diff_dict, commit_ret)
def on_diff_commit__attr(self, json_data):
# FIXME: sanitize input
json_dict = json.loads(json_data)
attr_diff = Attr_Diff.from_json_dict(json_dict['attr_diff'])
log.info('ws: rx: attr diff: ' + str(attr_diff))
ctx = self.__context__common(json_dict)
kernel = self.request.kernel
attr_diff, commit_ret = kernel.diff_commit__attr(attr_diff, ctx)
# [!] note: here we actually send the attr_diff twice, but in the future
# commit_ret may not be the same
return self.multicast_msg('diff_commit__attr', attr_diff, commit_ret)
def on_rzdoc_subscribe(self, data_dict):
return self._on_rzdoc_subscribe_common(data_dict, is_subscribe=True)
def on_rzdoc_unsubscribe(self, data_dict):
return self._on_rzdoc_subscribe_common(data_dict, is_subscribe=False)
def recv_connect(self):
self._log_conn('conn open')
super(WebSocket_Graph_NS, self).recv_connect() # super called despite being empty
def recv_disconnect(self):
self._log_conn('conn close')
super(WebSocket_Graph_NS, self).recv_disconnect()
|
