Ensure that Stream triggering state is handled correctly on bokeh ser…

…ver (#4041)

holoviews/plotting/plot.py

@@ -62,6 +62,7 @@ def __init__(self, renderer=None, root=None, **params):
         self._document = None
         self._root = None
         self._pane = None
+        self._triggering = []
         self.set_root(root)
 
 
@@ -191,21 +192,35 @@ def refresh(self, **kwargs):
             if (curdoc() is not self.document or (state._thread_id is not None and
                 thread_id != state._thread_id)):
                 # If we do not have the Document lock, schedule refresh as callback
+                self._triggering += [s for p in self.traverse(lambda x: x, [Plot])
+                                     for s in p.streams if s._triggering]
                 self.document.add_next_tick_callback(self.refresh)
                 return
 
-        traverse_setter(self, '_force', True)
-        key = self.current_key if self.current_key else self.keys[0]
-        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)
+        # Ensure that server based tick callbacks maintain stream triggering state
+        for s in self._triggering:
+            s._triggering = True
+        try:
 
-        self._trigger_refresh(stream_key)
-        if self.top_level:
-            self.push()
+            traverse_setter(self, '_force', True)
+            key = self.current_key if self.current_key else self.keys[0]
+            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)
+
+            self._trigger_refresh(stream_key)
+            if self.top_level:
+                self.push()
+        except Exception as e:
+            raise e
+        finally:
+            # Reset triggering state
+            for s in self._triggering:
+                s._triggering = False
+            self._triggering = []
 
 
     def _trigger_refresh(self, key):