Ensure multiple callbacks do not bleed wrong plot state

holoviews/plotting/bokeh/callbacks.py

@@ -1,4 +1,3 @@
-import json
 from collections import defaultdict
 
 import param
@@ -10,7 +9,7 @@
 from ..comms import JupyterCommJS
 
 
-def attributes_js(attributes):
+def attributes_js(attributes, handles):
     """
     Generates JS code to look up attributes on JS objects from
     an attributes specification dictionary.
@@ -28,7 +27,16 @@ def attributes_js(attributes):
         obj_name = attrs[0]
         attr_getters = ''.join(["['{attr}']".format(attr=attr)
                                 for attr in attrs[1:]])
-        code += ''.join([data_assign, obj_name, attr_getters, ';\n'])
+        if obj_name not in ['cb_obj', 'cb_data']:
+            assign_str = '{assign}{{id: {obj_name}["id"], value: {obj_name}{attr_getters}}};\n'.format(
+                assign=data_assign, obj_name=obj_name, attr_getters=attr_getters
+            )
+            code += 'if (({obj_name} != undefined) && ({obj_name}["id"] == "{id}")) {{ {assign} }}'.format(
+                obj_name=obj_name, id=handles[obj_name].ref['id'], assign=assign_str
+                )
+        else:
+            assign_str = ''.join([data_assign, obj_name, attr_getters, ';\n'])
+            code += assign_str
     return code
 
 
@@ -76,14 +84,16 @@ class Callback(object):
     js_callback = """
         function on_msg(msg){{
           msg = JSON.parse(msg.content.data);
-          var comm = HoloViewsWidget.comms["{comms_target}"];
-          var comm_state = HoloViewsWidget.comm_state["{comms_target}"];
+          var comms_target = msg["comms_target"]
+          var comm = HoloViewsWidget.comms[comms_target];
+          var comm_state = HoloViewsWidget.comm_state[comms_target];
           if (comm_state.event) {{
             comm.send(comm_state.event);
+            comm_state.blocked = true;
+            comm_state.timeout = Date.now()+{debounce};
           }} else {{
             comm_state.blocked = false;
           }}
-          comm_state.timeout = Date.now();
           comm_state.event = undefined;
           if ((msg.msg_type == "Ready") && msg.content) {{
             console.log("Python callback returned following output:", msg.content);
@@ -92,7 +102,7 @@ class Callback(object):
           }}
         }}
 
-        var argstring = JSON.stringify(data);
+        data['comms_target'] = "{comms_target}";
         if ((window.Jupyter !== undefined) && (Jupyter.notebook.kernel !== undefined)) {{
           var comm_manager = Jupyter.notebook.kernel.comm_manager;
           var comms = HoloViewsWidget.comms["{comms_target}"];
@@ -117,6 +127,8 @@ class Callback(object):
 
         function trigger() {{
             if (comm_state.event != undefined) {{
+               var comms_target = comm_state.event["comms_target"]
+               var comm = HoloViewsWidget.comms[comms_target];
                comm.send(comm_state.event);
             }}
             comm_state.event = undefined;
@@ -125,9 +137,9 @@ class Callback(object):
         timeout = comm_state.timeout + {timeout};
         if ((window.Jupyter == undefined) | (Jupyter.notebook.kernel == undefined)) {{
         }} else if ((comm_state.blocked && (Date.now() < timeout))) {{
-            comm_state.event = argstring;
+            comm_state.event = data;
         }} else {{
-            comm_state.event = argstring;
+            comm_state.event = data;
             setTimeout(trigger, {debounce});
             comm_state.blocked = true;
             comm_state.timeout = Date.now()+{debounce};
@@ -152,39 +164,84 @@ def __init__(self, plot, streams, source, **params):
         self.streams = streams
         self.comm = self._comm_type(plot, on_msg=self.on_msg)
         self.source = source
+        self.handle_ids = defaultdict(list)
 
 
     def initialize(self):
         plots = [self.plot]
         if self.plot.subplots:
             plots += list(self.plot.subplots.values())
 
+        handles = {}
+        for plot in plots:
+            for k, v in plot.handles.items():
+                if k not in handles:
+                    handles[k] = v
+        self.handle_ids.update(self._get_handle_ids(handles))
+
         found = []
         for plot in plots:
             for handle in self.handles:
                 if handle not in plot.handles or handle in found:
                     continue
-                self.set_customjs(plot.handles[handle])
+                self.set_customjs(plot.handles[handle], handles)
                 found.append(handle)
+
         if len(found) != len(self.handles):
             self.warning('Plotting handle for JS callback not found')
 
 
+    def _filter_msg(self, msg, ids):
+        """
+        Filter event values that do not originate from the plotting
+        handles associated with a particular stream using their
+        ids to match them.
+        """
+        filtered_msg = {}
+        for k, v in msg.items():
+            if isinstance(v, dict) and 'id' in v:
+                if v['id'] in ids:
+                    filtered_msg[k] = v['value']
+            else:
+                filtered_msg[k] = v
+        return filtered_msg
+
+
     def on_msg(self, msg):
-        msg = json.loads(msg)
-        msg = self._process_msg(msg)
-        if any(v is None for v in msg.values()):
-            return
         for stream in self.streams:
-            stream.update(trigger=False, **msg)
+            ids = self.handle_ids[stream]
+            filtered_msg = self._filter_msg(msg, ids)
+            processed_msg = self._process_msg(filtered_msg)
+            if not processed_msg:
+                continue
+            stream.update(trigger=False, **processed_msg)
         Stream.trigger(self.streams)
 
 
     def _process_msg(self, msg):
+        """
+        Subclassable method to preprocess JSON message in callback
+        before passing to stream.
+        """
         return msg
 
 
-    def set_customjs(self, handle):
+    def _get_handle_ids(self, handles):
+        """
+        Gather the ids of the plotting handles attached to this callback
+        This allows checking that a stream is not given the state
+        of a plotting handle it wasn't attached to
+        """
+        stream_handle_ids = defaultdict(list)
+        for stream in self.streams:
+            for h in self.handles:
+                if h in handles:
+                    handle_id = handles[h].ref['id']
+                    stream_handle_ids[stream].append(handle_id)
+        return stream_handle_ids
+
+
+    def set_customjs(self, handle, references):
         """
         Generates a CustomJS callback by generating the required JS
         code and gathering all plotting handles and installs it on
@@ -195,24 +252,21 @@ def set_customjs(self, handle):
         self_callback = self.js_callback.format(comms_target=self.comm.target,
                                                 timeout=self.timeout,
                                                 debounce=self.debounce)
-        attributes = attributes_js(self.attributes)
+
+        attributes = attributes_js(self.attributes, references)
         code = 'var data = {};\n' + attributes + self.code + self_callback
 
-        handles = {}
-        subplots = list(self.plot.subplots.values())[::-1] if self.plot.subplots else []
-        plots = [self.plot] + subplots
-        for plot in plots:
-            handles.update({k: v for k, v in plot.handles.items()
-                            if k in self.handles})
         # Set callback
         if id(handle.callback) in self._callbacks:
             cb = self._callbacks[id(handle.callback)]
             if isinstance(cb, type(self)):
                 cb.streams += self.streams
+                for k, v in self.handle_ids.items():
+                    cb.handle_ids[k] += v
             else:
                 handle.callback.code += code
         else:
-            js_callback = CustomJS(args=handles, code=code)
+            js_callback = CustomJS(args=references, code=code)
             self._callbacks[id(js_callback)] = self
             handle.callback = js_callback
 
@@ -249,8 +303,12 @@ class RangeXYCallback(Callback):
     handles = ['x_range', 'y_range']
 
     def _process_msg(self, msg):
-        return {'x_range': (msg['x0'], msg['x1']),
-                'y_range': (msg['y0'], msg['y1'])}
+        data = {}
+        if 'x0' in msg and 'x1' in msg:
+            data['x_range'] = (msg['x0'], msg['x1'])
+        if 'y0' in msg and 'y1' in msg:
+            data['y_range'] = (msg['y0'], msg['y1'])
+        return data
 
 
 class RangeXCallback(Callback):
@@ -261,7 +319,10 @@ class RangeXCallback(Callback):
     handles = ['x_range']
 
     def _process_msg(self, msg):
-        return {'x_range': (msg['x0'], msg['x1'])}
+        if 'x0' in msg and 'x1' in msg:
+            return {'x_range': (msg['x0'], msg['x1'])}
+        else:
+            return {}
 
 
 class RangeYCallback(Callback):
@@ -272,7 +333,10 @@ class RangeYCallback(Callback):
     handles = ['y_range']
 
     def _process_msg(self, msg):
-        return {'y_range': (msg['y0'], msg['y1'])}
+        if 'y0' in msg and 'y1' in msg:
+            return {'y_range': (msg['y0'], msg['y1'])}
+        else:
+            return {}
 
 
 class BoundsCallback(Callback):
@@ -285,7 +349,10 @@ class BoundsCallback(Callback):
     handles = ['box_select']
 
     def _process_msg(self, msg):
-        return {'bounds': (msg['x0'], msg['y0'], msg['x1'], msg['y1'])}
+        if all(c in msg for c in ['x0', 'y0', 'x1', 'y1']):
+            return {'bounds': (msg['x0'], msg['y0'], msg['x1'], msg['y1'])}
+        else:
+            return {}
 
 
 class Selection1DCallback(Callback):
@@ -295,7 +362,10 @@ class Selection1DCallback(Callback):
     handles = ['source']
 
     def _process_msg(self, msg):
-        return {'index': [int(v) for v in msg['index']]}
+        if 'index' in msg:
+            return {'index': [int(v) for v in msg['index']]}
+        else:
+            return {}
 
 
 callbacks = Stream._callbacks['bokeh']

holoviews/plotting/comms.py

@@ -114,11 +114,16 @@ def _handle_msg(self, msg):
             if stdout:
                 stdout = '\n\t'+'\n\t'.join(stdout)
                 error = '\n'.join([stdout, error])
-            msg = {'msg_type': "Error", 'traceback': error}
+            reply = {'msg_type': "Error", 'traceback': error}
         else:
             stdout = '\n\t'+'\n\t'.join(stdout) if stdout else ''
-            msg = {'msg_type': "Ready", 'content': stdout}
-        self.comm.send(json.dumps(msg))
+            reply = {'msg_type': "Ready", 'content': stdout}
+
+        # Returning the comms_target in an ACK message ensures that
+        # the correct comms handle is unblocked
+        if 'comms_target' in msg:
+            reply['comms_target'] = msg.pop('comms_target', None)
+        self.send(json.dumps(reply))
 
 
 class JupyterComm(Comm):
@@ -154,6 +159,10 @@ def init(self):
 
     @classmethod
     def decode(cls, msg):
+        """
+        Decodes messages following Jupyter messaging protocol.
+        If JSON decoding fails data is assumed to be a regular string.
+        """
         return msg['content']['data']
 
 

tests/testcomms.py

@@ -1,3 +1,4 @@
+import json
 from nose.tools import *
 
 from holoviews.element.comparison import ComparisonTestCase
@@ -17,13 +18,33 @@ def test_decode(self):
         msg = 'Test'
         self.assertEqual(Comm.decode(msg), msg)
 
-    def test_on_msg(self):
+    def test_handle_message_error_reply(self):
         def raise_error(msg):
-            if msg == 'Error':
-                raise Exception()
+            raise Exception('Test')
+        def assert_error(msg):
+            decoded = json.loads(msg)
+            self.assertEqual(decoded['msg_type'], "Error")
+            self.assertTrue(decoded['traceback'].endswith('Exception: Test'))
         comm = Comm(None, target='Test', on_msg=raise_error)
-        with self.assertRaises(Exception):
-            comm._handle_msg('Error')
+        comm.send = assert_error
+        comm._handle_msg({})
+
+    def test_handle_message_ready_reply(self):
+        def assert_ready(msg):
+            self.assertEqual(json.loads(msg), {'msg_type': "Ready", 'content': ''})
+        comm = Comm(None, target='Test')
+        comm.send = assert_ready
+        comm._handle_msg({})
+
+    def test_handle_message_ready_reply_with_comms_target(self):
+        def assert_ready(msg):
+            decoded = json.loads(msg)
+            self.assertEqual(decoded, {'msg_type': "Ready", 'content': '',
+                                       'comms_target': 'Testing target'})
+        comm = Comm(None, target='Test')
+        comm.send = assert_ready
+        comm._handle_msg({'comms_target': 'Testing target'})
+
 
 
 class TestJupyterComm(ComparisonTestCase):

tests/testplotinstantiation.py

@@ -246,7 +246,7 @@ def test_stream_callback(self):
         dmap = DynamicMap(lambda x, y: Points([(x, y)]), kdims=[], streams=[PositionXY()])
         plot = bokeh_renderer.get_plot(dmap)
         bokeh_renderer(plot)
-        plot.callbacks[0].on_msg('{"x": 10, "y": -10}')
+        plot.callbacks[0].on_msg({"x": 10, "y": -10})
         data = plot.handles['source'].data
         self.assertEqual(data['x'], np.array([10]))
         self.assertEqual(data['y'], np.array([-10]))