Merge pull request #904 from ioam/linked_streams

Allow multiple streams attached to the same plot model

holoviews/plotting/bokeh/callbacks.py

@@ -133,6 +133,8 @@ class Callback(object):
     # Timeout if a comm message is swallowed
     timeout = 20000
 
+    _callbacks = {}
+
     def __init__(self, plot, streams, source, **params):
         self.plot = plot
         self.streams = streams
@@ -162,7 +164,8 @@ def on_msg(self, msg):
         if any(v is None for v in msg.values()):
             return
         for stream in self.streams:
-            stream.update(**msg)
+            stream.update(trigger=False, **msg)
+        Stream.trigger(self.streams)
 
 
     def _process_msg(self, msg):
@@ -188,7 +191,16 @@ def set_customjs(self, handle):
         for plot in plots:
             handles.update(plot.handles)
         # Set callback
-        handle.callback = CustomJS(args=handles, code=code)
+        if id(handle.callback) in self._callbacks:
+            cb = self._callbacks[id(handle.callback)]
+            if isinstance(cb, type(self)):
+                cb.streams += self.streams
+            else:
+                handle.callback.code += code
+        else:
+            js_callback = CustomJS(args=handles, code=code)
+            self._callbacks[id(js_callback)] = self
+            handle.callback = js_callback
 
 
 

holoviews/plotting/plot.py

@@ -486,7 +486,9 @@ def refresh(self, **kwargs):
         """
         traverse_setter(self, '_force', True)
         key = self.current_key if self.current_key else self.keys[0]
-        stream_params = stream_parameters(self.streams)
+        dim_streams = [stream for stream in self.streams
+                       if any(c in self.dimensions for c in stream.contents)]
+        stream_params = stream_parameters(dim_streams)
         key = tuple(None if d in stream_params else k
                     for d, k in zip(self.dimensions, key))
         stream_key = util.wrap_tuple_streams(key, self.dimensions, self.streams)

holoviews/util.py

@@ -1,10 +1,12 @@
+import inspect
+
 import param
 
 from .core import DynamicMap, ViewableElement
 from .core.operation import ElementOperation
 from .core.util import Aliases
 from .core import util
-
+from .streams import Stream
 
 class Dynamic(param.ParameterizedFunction):
     """
@@ -36,8 +38,12 @@ def __call__(self, map_obj, **params):
             dmap = self._make_dynamic(map_obj, callback)
         if isinstance(self.p.operation, ElementOperation):
             streams = []
-            for s in self.p.streams:
-                stream = s()
+            for stream in self.p.streams:
+                if inspect.isclass(stream) and issubclass(stream, Stream):
+                    stream = stream()
+                elif not isinstance(stream, Stream):
+                    raise ValueError('Stream must only contain Stream '
+                                     'classes or instances')
                 stream.update(**{k: self.p.operation.p.get(k) for k, v in
                                  stream.contents.items()})
                 streams.append(stream)