Ensure multiple callbacks do not bleed wrong plot state

holoviews/plotting/bokeh/callbacks.py

@@ -10,7 +10,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 +28,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 +85,20 @@ 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}"];
+          if ("comms_target" in msg) {{
+            comms_target = msg["comms_target"]
+          }} else {{
+            comms_target = "{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,6 +107,7 @@ class Callback(object):
           }}
         }}
 
+        data['comms_target'] = "{comms_target}";
         var argstring = JSON.stringify(data);
         if ((window.Jupyter !== undefined) && (Jupyter.notebook.kernel !== undefined)) {{
           var comm_manager = Jupyter.notebook.kernel.comm_manager;
@@ -151,6 +167,7 @@ def __init__(self, plot, streams, source, **params):
         self.plot = plot
         self.streams = streams
         self.comm = self._comm_type(plot, on_msg=self.on_msg)
+        self.stream_handles = defaultdict(list)
         self.source = source
 
 
@@ -171,12 +188,23 @@ def initialize(self):
 
 
     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 each stream check whether plot state is meant for it
+        # by checking that the IDs match the IDs of the stream's plot
+        # handles, dispatch only the part of the message meant for
+        # a particular stream
         for stream in self.streams:
-            stream.update(trigger=False, **msg)
+            ids = self.stream_handles[stream]
+            sanitized_msg = {}
+            for k, v in msg.items():
+                if isinstance(v, dict) and 'id' in v:
+                    if v['id'] in ids:
+                        sanitized_msg[k] = v['value']
+                else:
+                    sanitized_msg[k] = v
+            processed_msg = self._process_msg(sanitized_msg)
+            if not processed_msg:
+                continue
+            stream.update(trigger=False, **processed_msg)
         Stream.trigger(self.streams)
 
 
@@ -195,23 +223,38 @@ 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)
-        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})
+
+        attributes = attributes_js(self.attributes, handles)
+        code = 'var data = {};\n' + attributes + self.code + self_callback
+
+        # 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)
+
         # 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 stream_handle_ids.items():
+                    cb.stream_handles[k] += v
             else:
                 handle.callback.code += code
         else:
+            self.stream_handles.update(stream_handle_ids)
             js_callback = CustomJS(args=handles, code=code)
             self._callbacks[id(js_callback)] = self
             handle.callback = js_callback
@@ -249,8 +292,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 +308,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 +322,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 +338,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 +351,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,7 +159,16 @@ def init(self):
 
     @classmethod
     def decode(cls, msg):
-        return msg['content']['data']
+        """
+        Decodes messages following Jupyter messaging protocol.
+        If JSON decoding fails data is assumed to be a regular string.
+        """
+        data = msg['content']['data']
+        try:
+            data = json.loads(data)
+        except ValueError:
+            pass
+        return data
 
 
     def send(self, 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]))