diff --git a/holoviews/core/util.py b/holoviews/core/util.py
index 56d514be14..a4a1ca8698 100644
--- a/holoviews/core/util.py
+++ b/holoviews/core/util.py
@@ -954,7 +954,7 @@ def get_dynamic_item(map_obj, dimensions, key):
     and a corresponding key. The dimensions must be a subset
     of the map_obj key dimensions.
     """
-    if key == () and not dimensions:
+    if key == () and not dimensions or not map_obj.kdims:
         return key, map_obj[()]
     elif isinstance(key, tuple):
         dims = {d.name: k for d, k in zip(dimensions, key)
diff --git a/holoviews/plotting/bokeh/__init__.py b/holoviews/plotting/bokeh/__init__.py
index 19edd9d645..110b9a0057 100644
--- a/holoviews/plotting/bokeh/__init__.py
+++ b/holoviews/plotting/bokeh/__init__.py
@@ -17,7 +17,7 @@
 from ..plot import PlotSelector
 
 from .annotation import TextPlot, LineAnnotationPlot, SplinePlot
-from .callbacks import Callbacks # noqa (API import)
+from .callbacks import Callback # noqa (API import)
 from .element import OverlayPlot, BokehMPLWrapper
 from .chart import (PointPlot, CurvePlot, SpreadPlot, ErrorPlot, HistogramPlot,
                     SideHistogramPlot, BoxPlot, BarPlot, SpikesPlot,
diff --git a/holoviews/plotting/bokeh/callbacks.py b/holoviews/plotting/bokeh/callbacks.py
index 9b764e7a03..900b66f6aa 100644
--- a/holoviews/plotting/bokeh/callbacks.py
+++ b/holoviews/plotting/bokeh/callbacks.py
@@ -1,446 +1,242 @@
+import json
 from collections import defaultdict
 
-import numpy as np
 import param
+import numpy as np
+from bokeh.models import CustomJS
 
-from ...core.data import ArrayColumns, PandasInterface
-from .util import compute_static_patch, models_to_json
-
-from bokeh.models import CustomJS, TapTool, ColumnDataSource
-from bokeh.core.json_encoder import serialize_json
-from bokeh.io import _CommsHandle
-from bokeh.util.notebook import get_comms
+from ...streams import (Stream, PositionXY, RangeXY, Selection1D, RangeX,
+                        RangeY, PositionX, PositionY, Bounds)
+from ..comms import JupyterCommJS
 
 
-class Callback(param.ParameterizedFunction):
-    """
-    Callback functions provide an easy way to interactively modify
-    the plot based on changes to the bokeh axis ranges, data sources,
-    tools or widgets.
-
-    The data sent to the Python callback from javascript is defined
-    by the plot attributes and callback_obj attributes and optionally
-    custom javascript code.
-
-    The user should define any plot_attributes and cb_attributes
-    he wants access to, which will be supplied in the form of a
-    dictionary to the call method. The call method can then apply
-    any processing depending on the callback data and return the
-    modified bokeh plot objects.
+def attributes_js(attributes):
     """
+    Generates JS code to look up attributes on JS objects from
+    an attributes specification dictionary.
 
-    apply_on_update = param.Boolean(default=False, doc="""
-        Whether the callback is applied when the plot is updated""")
-
-    callback_obj = param.Parameter(doc="""
-        Bokeh PlotObject the callback is applied to.""")
-
-    cb_attributes = param.List(default=[], doc="""
-        Callback attributes returned to the Python callback.""")
-
-    code = param.String(default="", doc="""
-        Custom javascript code executed on the callback. The code
-        has access to the plot, source and cb_obj and may modify
-        the data javascript object sent back to Python.""")
-
-    current_data = param.Dict(default={}, doc="""
-        A dictionary of the last data supplied to the callback.""")
-
-    delay = param.Number(default=200, doc="""
-        Delay when initiating callback events in milliseconds.""")
-
-    initialize_cb = param.Boolean(default=True, doc="""
-        Whether the callback should be initialized when it's first
-        added to a plot""")
-
-    plot_attributes = param.Dict(default={}, doc="""
-        Plot attributes returned to the Python callback.""")
-
-    plots = param.List(default=[], doc="""
-        The HoloViews plot object the callback applies to.""")
-
-    streams = param.List(default=[], doc="""
-        List of streams attached to this callback.""")
-
-    reinitialize = param.Boolean(default=False, doc="""
-        Whether the Callback should be reinitialized per plot instance""")
-
-    skip_unchanged = param.Boolean(default=False, doc="""
-        Avoid running the callback if the callback data is unchanged.
-        Useful for avoiding infinite loops.""")
-
-    timeout = param.Number(default=2500, doc="""
-        Callback error timeout in milliseconds.""")
-
-    JS_callback = """
-        function callback(msg){
-          if (msg.msg_type == "execute_result") {
-            if (msg.content.data['text/plain'] === "'Complete'") {
-               if (HoloViewsWidget._queued.length) {
-                   execute_callback();
-               } else {
-                   HoloViewsWidget._blocked = false;
-               }
-               HoloViewsWidget._timeout = Date.now();
-            }
-          } else {
-            console.log("Python callback returned unexpected message:", msg)
-          }
-        }
-        callbacks = {iopub: {output: callback}};
-    var data = {};
-    """
+    Example:
 
-    IPython_callback = """
-        function execute_callback() {{
-           data = HoloViewsWidget._queued.pop(0);
-           var argstring = JSON.stringify(data);
-           argstring = argstring.replace('true', 'True').replace('false','False');
-           var kernel = IPython.notebook.kernel;
-           var cmd = "Callbacks.callbacks[{callback_id}].update(" + argstring + ")";
-           var pyimport = "from holoviews.plotting.bokeh import Callbacks;";
-           kernel.execute(pyimport + cmd, callbacks, {{silent : false}});
-        }}
+    Input  : {'x': 'cb_data.geometry.x'}
 
-        if (!HoloViewsWidget._queued) {{
-            HoloViewsWidget._queued = [];
-            HoloViewsWidget._blocked = false;
-            HoloViewsWidget._timeout = Date.now();
-        }}
+    Output : data['x'] = cb_data['geometry']['x']
+    """
+    code = ''
+    for key, attr_path in attributes.items():
+        data_assign = "data['{key}'] = ".format(key=key)
+        attrs = attr_path.split('.')
+        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'])
+    return code
+
+
+class Callback(object):
+    """
+    Provides a baseclass to define callbacks, which return data from
+    bokeh models such as the plot ranges or various tools. The callback
+    then makes this data available to any streams attached to it.
+
+    The defintion of a callback consists of a number of components:
+
+    * handles    :  The handles define which plotting handles the
+                    callback will be attached on, e.g. this could be
+                    the x_range, y_range, a plotting tool or any other
+                    bokeh object that allows callbacks.
+
+    * attributes :  The attributes define which attributes to send
+                    back to Python. They are defined as a dictionary
+                    mapping between the name under which the variable
+                    is made available to Python and the specification
+                    of the attribute. The specification should start
+                    with the variable name that is to be accessed and
+                    the location of the attribute separated by periods.
+                    All plotting handles such as tools, the x_range,
+                    y_range and (data)source can be addressed in this
+                    way, e.g. to get the start of the x_range as 'x'
+                    you can supply {'x': 'x_range.attributes.start'}.
+                    Additionally certain handles additionally make the
+                    cb_data and cb_obj variables available containing
+                    additional information about the event.
+
+    * code        : Defines any additional JS code to be executed,
+                    which can modify the data object that is sent to
+                    the backend.
+
+    The callback can also define a _process_msg method, which can
+    modify the data sent by the callback before it is passed to the
+    streams.
+    """
 
-        timeout = HoloViewsWidget._timeout + {timeout};
-        if ((typeof _ === "undefined")  || _.isEmpty(data)) {{
-        }} else if ((HoloViewsWidget._blocked && (Date.now() < timeout))) {{
-            HoloViewsWidget._queued = [data];
-        }} else {{
-            HoloViewsWidget._queued = [data];
-            setTimeout(execute_callback, {delay});
-            HoloViewsWidget._blocked = true;
-            HoloViewsWidget._timeout = Date.now();
+    code = ""
+
+    attributes = {}
+
+    js_callback = """
+        var argstring = JSON.stringify(data);
+        if ((window.Jupyter !== undefined) && (Jupyter.notebook.kernel !== undefined)) {{
+          var comm_manager = Jupyter.notebook.kernel.comm_manager;
+          var comms = HoloViewsWidget.comms["{comms_target}"];
+          if (comms && ("{comms_target}" in comms)) {{
+            comm = comms["{comms_target}"];
+          }} else {{
+            comm = comm_manager.new_comm("{comms_target}", {{}}, {{}}, {{}});
+
+          comm_manager["{comms_target}"] = comm;
+            HoloViewsWidget.comms["{comms_target}"] = comm;
+          }}
+          comm_manager["{comms_target}"] = comm;
+          comm.send(argstring)
         }}
     """
 
-    def initialize(self, data):
-        """
-        Initialize is called when the callback is added to a new plot
-        and the initialize option is enabled. May avoid repeat
-        initialization by setting initialize_cb parameter to False
-        inside this method.
-        """
+    # The plotting handle(s) to attach the JS callback on
+    handles = []
 
+    _comm_type = JupyterCommJS
 
-    def __call__(self, data):
-        """
-        The call method can modify any bokeh plot object
-        depending on the supplied data dictionary. It
-        should return the modified plot objects as a list.
-        """
-        return []
+    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.source = source
 
 
-    def update(self, data, chained=False):
-        """
-        The update method is called by the javascript callback
-        with the supplied data and will return the json serialized
-        string representation of the changes to the Bokeh plot.
-        When chained=True it will return a list of the plot objects
-        to be updated, allowing chaining of callback operations.
-        """
-        if self.skip_unchanged and self.current_data == data:
-            return [] if chained else "{}"
-        self.current_data = data
+    def initialize(self):
+        plots = [self.plot]
+        if self.plot.subplots:
+            plots += list(self.plot.subplots.values())
+
+        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])
+                found.append(handle)
+        if len(found) != len(self.handles):
+            self.warning('Plotting handle for JS callback not found')
 
-        objects = self(data)
 
+    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:
-            objects += stream.update(data, True)
+            stream.update(**msg)
 
-        if chained:
-            return objects
-        else:
-            return self.serialize(objects)
 
+    def _process_msg(self, msg):
+        return msg
 
-    def serialize(self, models):
+
+    def set_customjs(self, handle):
         """
-        Serializes any Bokeh plot objects passed to it as a list.
+        Generates a CustomJS callback by generating the required JS
+        code and gathering all plotting handles and installs it on
+        the requested callback handle.
         """
-        for plot in self.plots:
-            msg = compute_static_patch(plot.document, models)
-            plot.comm.send(serialize_json(msg))
-        return 'Complete'
-
 
+        # Generate callback JS code to get all the requested data
+        self_callback = self.js_callback.format(comms_target=self.comm.target)
+        attributes = attributes_js(self.attributes)
+        code = 'var data = {};\n' + attributes + self.code + self_callback
 
-class DownsampleCallback(Callback):
-    """
-    DownsampleCallbacks can downsample the data before it is
-    plotted and can therefore provide major speed optimizations.
-    """
+        handles = dict(self.plot.handles)
+        plots = [self.plot] + (self.plot.subplots.values()[::-1] if self.plot.subplots else [])
+        for plot in plots:
+            handles.update(plot.handles)
+        # Set callback
+        handle.callback = CustomJS(args=handles, code=code)
 
-    apply_on_update = param.Boolean(default=True, doc="""
-        Callback should always be applied after each update to
-        downsample the data before it is displayed.""")
 
-    reinitialize = param.Boolean(default=True, doc="""
-        DownsampleColumns should be reinitialized per plot object""")
 
+class PositionXYCallback(Callback):
 
+    attributes = {'x': 'cb_data.geometry.x', 'y': 'cb_data.geometry.y'}
 
-class DownsampleImage(DownsampleCallback):
-    """
-    Downsamples any Image plot to the specified
-    max_width and max_height by slicing the
-    Image to the specified x_range and y_range
-    and then finding step values matching the
-    constraints.
-    """
+    handles = ['hover']
 
-    max_width = param.Integer(default=250, doc="""
-        Maximum plot width in pixels after slicing and downsampling.""")
 
-    max_height = param.Integer(default=250, doc="""
-        Maximum plot height in pixels after slicing and downsampling.""")
+class PositionXCallback(Callback):
 
-    plot_attributes = param.Dict(default={'x_range': ['start', 'end'],
-                                          'y_range': ['start', 'end']})
+    attributes = {'x': 'cb_data.geometry.x'}
 
-    def __call__(self, data):
-        xstart, xend = data['x_range']
-        ystart, yend = data['y_range']
+    handles = ['hover']
 
-        ranges = self.plots[0].current_ranges
-        element = self.plots[0].current_frame
 
-        # Slice Element to match selected ranges
-        xdim, ydim = element.dimensions('key', True)
-        sliced = element.select(**{xdim: (xstart, xend),
-                                   ydim: (ystart, yend)})
+class PositionYCallback(Callback):
 
-        # Get dimensions of sliced element
-        shape = sliced.data.shape
-        max_shape = (self.max_height, self.max_width)
+    attributes = {'y': 'cb_data.geometry.y'}
 
-        #Find minimum downsampling to fit requirement
-        steps = []
-        for s, max_s in zip(shape, max_shape):
-            step = 1
-            while s/step > max_s: step += 1
-            steps.append(step)
-        resampled = sliced.clone(sliced.data[::steps[0], ::steps[1]])
+    handles = ['hover']
 
-        # Update data source
-        new_data = self.plots[0].get_data(resampled, ranges)[0]
-        source = self.plots[0].handles['source']
-        source.data.update(new_data)
-        return [source]
 
+class RangeXYCallback(Callback):
 
+    attributes = {'x0': 'x_range.attributes.start',
+                  'x1': 'x_range.attributes.end',
+                  'y0': 'y_range.attributes.start',
+                  'y1': 'y_range.attributes.end'}
 
-class DownsampleColumns(DownsampleCallback):
-    """
-    Downsamples any column based Element by randomizing
-    the rows and updating the ColumnDataSource with
-    up to max_samples.
-    """
+    handles = ['x_range', 'y_range']
 
-    compute_ranges = param.Boolean(default=False, doc="""
-        Whether the ranges are recomputed for the sliced region""")
+    def _process_msg(self, msg):
+        return {'x_range': (msg['x0'], msg['x1']),
+                'y_range': (msg['y0'], msg['y1'])}
 
-    max_samples = param.Integer(default=800, doc="""
-        Maximum number of samples to display at the same time.""")
 
-    random_seed = param.Integer(default=42, doc="""
-        Seed used to initialize randomization.""")
+class RangeXCallback(Callback):
 
-    plot_attributes = param.Dict(default={'x_range': ['start', 'end'],
-                                          'y_range': ['start', 'end']})
+    attributes = {'x0': 'x_range.attributes.start',
+                  'x1': 'x_range.attributes.end'}
 
+    handles = ['x_range']
 
-    def initialize(self, data):
-        self.prng = np.random.RandomState(self.random_seed)
+    def _process_msg(self, msg):
+        return {'x_range': (msg['x0'], msg['x1'])}
 
-    def __call__(self, data):
-        xstart, xend = data['x_range']
-        ystart, yend = data['y_range']
 
-        plot = self.plots[0]
-        element = plot.current_frame
-        if element.interface not in [ArrayColumns, PandasInterface]:
-            element = plot.current_frame.clone(datatype=['array'])
+class RangeYCallback(Callback):
 
-        # Slice element to current ranges
-        xdim, ydim = element.dimensions(label=True)[0:2]
-        sliced = element.select(**{xdim: (xstart, xend),
-                                   ydim: (ystart, yend)})
+    attributes = {'y0': 'y_range.attributes.start',
+                  'y1': 'y_range.attributes.end'}
 
-        if self.compute_ranges:
-            ranges = {d: element.range(d) for d in element.dimensions()}
-        else:
-            ranges  = plot.current_ranges
+    handles = ['y_range']
 
-        if len(sliced) > self.max_samples:
-            length = len(sliced)
-            if element.interface is PandasInterface:
-                data = sliced.data.sample(self.max_samples,
-                                          random_state=self.prng)
-            else:
-                inds = self.prng.choice(length, self.max_samples, False)
-                data = element.data[inds, :]
-            sliced = element.clone(data)
+    def _process_msg(self, msg):
+        return {'y_range': (msg['y0'], msg['y1'])}
 
-        # Update data source
-        new_data = plot.get_data(sliced, ranges)[0]
-        source = plot.handles['source']
-        source.data.update(new_data)
-        return [source]
 
+class BoundsCallback(Callback):
 
-class Callbacks(param.Parameterized):
-    """
-    Callbacks allows defining a number of callbacks to be applied
-    to a plot. Callbacks should
-    """
+    attributes = {'x0': 'cb_data.geometry.x0',
+                  'x1': 'cb_data.geometry.x1',
+                  'y0': 'cb_data.geometry.y0',
+                  'y1': 'cb_data.geometry.y1'}
 
-    selection = param.ClassSelector(class_=(CustomJS, Callback, list), doc="""
-        Callback that gets triggered when user applies a selection to a
-        data source.""")
+    handles = ['box_select']
 
-    ranges = param.ClassSelector(class_=(CustomJS, Callback, list), doc="""
-        Callback applied to plot x_range and y_range, data will
-        supply 'x_range' and 'y_range' lists of the form [low, high].""")
+    def _process_msg(self, msg):
+        return {'bounds': (msg['x0'], msg['y0'], msg['x1'], msg['y1'])}
 
-    x_range = param.ClassSelector(class_=(CustomJS, Callback, list), doc="""
-        Callback applied to plot x_range, data will supply
-        'x_range' as a list of the form [low, high].""")
 
-    y_range = param.ClassSelector(class_=(CustomJS, Callback, list), doc="""
-        Callback applied to plot x_range, data will supply
-        'y_range' as a list of the form [low, high].""")
+class Selection1DCallback(Callback):
 
-    tap = param.ClassSelector(class_=(CustomJS, Callback, list), doc="""
-        Callback that gets triggered when user clicks on a glyph.""")
+    attributes = {'index': 'source.selected.1d.indices'}
 
-    callbacks = {}
+    handles = ['source']
 
-    plot_callbacks = defaultdict(list)
 
-    def initialize_callback(self, cb_obj, plot, pycallback):
-        """
-        Initialize the callback with the appropriate data
-        and javascript, execute once and return bokeh CustomJS
-        object to be installed on the appropriate plot object.
-        """
-        if pycallback.reinitialize:
-            pycallback = pycallback.instance(plots=[])
-        pycallback.callback_obj = cb_obj
-        pycallback.plots.append(plot)
-
-        # Register the callback to allow calling it from JS
-        cb_id = id(pycallback)
-        self.callbacks[cb_id] = pycallback
-        self.plot_callbacks[id(cb_obj)].append(pycallback)
+callbacks = Stream._callbacks['bokeh']
 
-        # Generate callback JS code to get all the requested data
-        self_callback = Callback.IPython_callback.format(callback_id=cb_id,
-                                                         timeout=pycallback.timeout,
-                                                         delay=pycallback.delay)
-        code = ''
-        for k, v in pycallback.plot_attributes.items():
-            format_kwargs = dict(key=repr(k), attrs=repr(v))
-            if v is None:
-                code += "data[{key}] = plot.get({key});\n".format(**format_kwargs)
-            else:
-                code += "data[{key}] = {attrs}.map(function(attr) {{" \
-                        "  return plot.get({key}).get(attr)" \
-                        "}})\n".format(**format_kwargs)
-        if pycallback.cb_attributes:
-            code += "data['cb_obj'] = {attrs}.map(function(attr) {{"\
-                    "  return cb_obj.get(attr)}});\n".format(attrs=repr(pycallback.cb_attributes))
-
-        data = self._get_data(pycallback, plot)
-        code = Callback.JS_callback + code + pycallback.code + self_callback
-
-        # Generate CustomJS object
-        customjs = CustomJS(args=plot.handles, code=code)
-
-        # Get initial callback data and call to initialize
-        if pycallback.initialize_cb:
-            pycallback.initialize(data)
-
-        return customjs, pycallback
-
-
-    def _get_data(self, pycallback, plot):
-        data = {}
-        plot_data = models_to_json([plot.state])[0]
-        for k, v in pycallback.plot_attributes.items():
-            if v is None:
-                data[k] = plot_data.get(k)
-            else:
-                obj = getattr(plot.state, k)
-                obj_data = models_to_json([obj])[0]
-                data[k] = [obj_data.get(attr, obj_data.get('data', {}).get(attr))
-                           for attr in v]
-        if pycallback.cb_attributes:
-            cb_data = models_to_json([pycallback.callback_obj])[0]
-            data['cb_obj'] = [cb_data.get(attr) for attr in pycallback.cb_attributes]
-        return data
-
-
-    def _chain_callbacks(self, plot, cb_obj, callbacks):
-        """
-        Initializes new callbacks and chains them to
-        existing callbacks, allowing multiple callbacks
-        on the same plot object.
-        """
-        other_callbacks = self.plot_callbacks[id(cb_obj)]
-        chain_callback = other_callbacks[-1] if other_callbacks else None
-        if not isinstance(callbacks, list): callbacks = [callbacks]
-        for callback in callbacks:
-            if isinstance(callback, Callback):
-                jscb, pycb = self.initialize_callback(cb_obj, plot, callback)
-                if chain_callback and pycb is not chain_callback:
-                    chain_callback.streams.append(pycb)
-                    chain_callback = pycb
-                else:
-                    cb_obj.callback = jscb
-                    chain_callback = pycb
-            else:
-                cb_obj.callback = callback
-
-    @property
-    def downsample(self):
-        return any(isinstance(v, DownsampleCallback)
-                   for _ , v in self.get_param_values())
-
-
-    def __call__(self, plot):
-        """
-        Initialize callbacks, chaining them as necessary
-        and setting them on the appropriate plot object.
-        """
-        # Initialize range callbacks
-        xrange_cb = self.ranges if self.ranges else self.x_range
-        yrange_cb = self.ranges if self.ranges else self.y_range
-        if xrange_cb:
-            self._chain_callbacks(plot, plot.state.x_range, xrange_cb)
-        if yrange_cb:
-            self._chain_callbacks(plot, plot.state.y_range, yrange_cb)
-
-        if self.tap:
-            for tool in plot.state.select(type=TapTool):
-                self._chain_callbacks(plot, tool, self.tap)
-
-        if self.selection:
-            for tool in plot.state.select(type=(ColumnDataSource)):
-                self._chain_callbacks(plot, tool, self.selection)
-
-    def update(self, plot):
-        """
-        Allows updating the callbacks before data is sent to frontend.
-        """
-        for cb in self.callbacks.values():
-            if cb.apply_on_update and plot in cb.plots:
-                data = self._get_data(cb, plot)
-                cb(data)
+callbacks[PositionXY] = PositionXYCallback
+callbacks[PositionX] = PositionXCallback
+callbacks[PositionY] = PositionYCallback
+callbacks[RangeXY] = RangeXYCallback
+callbacks[RangeX] = RangeXCallback
+callbacks[RangeY] = RangeYCallback
+callbacks[Bounds] = BoundsCallback
+callbacks[Selection1D] = Selection1DCallback
diff --git a/holoviews/plotting/bokeh/element.py b/holoviews/plotting/bokeh/element.py
index 4a0148d4da..88dbd6e5da 100644
--- a/holoviews/plotting/bokeh/element.py
+++ b/holoviews/plotting/bokeh/element.py
@@ -1,4 +1,5 @@
 from io import BytesIO
+from itertools import groupby
 
 import numpy as np
 import bokeh
@@ -14,6 +15,7 @@
 except ImportError:
     LogColorMapper, ColorBar = None, None
 from bokeh.models import LogTicker, BasicTicker
+from bokeh.plotting.helpers import _known_tools as known_tools
 
 try:
     from bokeh import mpl
@@ -26,9 +28,9 @@
 from ...core.options import abbreviated_exception
 from ...core import util
 from ...element import RGB
+from ...streams import Stream
 from ..plot import GenericElementPlot, GenericOverlayPlot
 from ..util import dynamic_update
-from .callbacks import Callbacks
 from .plot import BokehPlot
 from .util import (mpl_to_bokeh, convert_datetime, update_plot,
                    bokeh_version, mplcmap_to_palette)
@@ -55,10 +57,6 @@
 
 class ElementPlot(BokehPlot, GenericElementPlot):
 
-    callbacks = param.ClassSelector(class_=Callbacks, doc="""
-        Callbacks object defining any javascript callbacks applied
-        to the plot.""")
-
     bgcolor = param.Parameter(default='white', doc="""
         Background color of the plot.""")
 
@@ -168,18 +166,56 @@ def __init__(self, element, plot=None, **params):
         self.current_ranges = None
         super(ElementPlot, self).__init__(element, **params)
         self.handles = {} if plot is None else self.handles['plot']
-        element_ids = self.hmap.traverse(lambda x: id(x), [Element])
-        self.static = len(set(element_ids)) == 1 and len(self.keys) == len(self.hmap)
+        self.static = len(self.hmap) == 1 and len(self.keys) == len(self.hmap)
+        self.callbacks = self._construct_callbacks()
 
 
-    def _init_tools(self, element):
+    def _construct_callbacks(self):
+        """
+        Initializes any callbacks for streams which have defined
+        the plotted object as a source.
+        """
+        if not self.static or isinstance(self.hmap, DynamicMap):
+            source = self.hmap
+        else:
+            source = self.hmap.last
+        streams = Stream.registry.get(id(source), [])
+        registry = Stream._callbacks['bokeh']
+        callbacks = {(registry[type(stream)], stream) for stream in streams
+                     if type(stream) in registry and streams}
+        cbs = []
+        for cb, group in groupby(sorted(callbacks), lambda x: x[0]):
+            cb_streams = [s for _, s in group]
+            cbs.append(cb(self, cb_streams, source))
+        return cbs
+
+
+    def _init_tools(self, element, callbacks=[]):
         """
         Processes the list of tools to be supplied to the plot.
         """
-        tools = self.default_tools + self.tools
+        if self.batched:
+            dims = self.hmap.last.kdims
+        else:
+            dims = list(self.overlay_dims.keys())
+        dims += element.dimensions()
+        tooltips = [(d.pprint_label, '@'+util.dimension_sanitizer(d.name))
+                    for d in dims]
+
+        callbacks = callbacks+self.callbacks
+        cb_tools = []
+        for cb in callbacks:
+            for handle in cb.handles:
+                if handle and handle in known_tools:
+                    if handle == 'hover':
+                        tool = HoverTool(tooltips=tooltips)
+                    else:
+                        tool = known_tools[handle]()
+                    cb_tools.append(tool)
+                    self.handles[handle] = tool
+
+        tools = cb_tools + self.default_tools + self.tools
         if 'hover' in tools:
-            tooltips = [(d.pprint_label, '@'+util.dimension_sanitizer(d.name))
-                        for d in element.dimensions()]
             tools[tools.index('hover')] = HoverTool(tooltips=tooltips)
         return tools
 
@@ -347,6 +383,8 @@ def _init_axes(self, plot):
             plot.above = plot.below
             plot.below = []
             plot.xaxis[:] = plot.above
+        self.handles['xaxis'] = plot.xaxis[0]
+        self.handles['x_range'] = plot.x_range
 
         if self.yaxis is None:
             plot.yaxis.visible = False
@@ -354,6 +392,8 @@ def _init_axes(self, plot):
             plot.right = plot.left
             plot.left = []
             plot.yaxis[:] = plot.right
+        self.handles['yaxis'] = plot.yaxis[0]
+        self.handles['y_range'] = plot.y_range
 
 
     def _axis_properties(self, axis, key, plot, dimension,
@@ -512,7 +552,7 @@ def initialize_plot(self, ranges=None, plot=None, plots=None, source=None):
         self.handles['plot'] = plot
 
         # Get data and initialize data source
-        empty = self.callbacks and self.callbacks.downsample
+        empty = False
         if self.batched:
             data, mapping = self.get_batched_data(element, ranges, empty)
         else:
@@ -533,9 +573,11 @@ def initialize_plot(self, ranges=None, plot=None, plots=None, source=None):
             self._update_glyph(glyph, properties, mapping)
         if not self.overlaid:
             self._update_plot(key, plot, style_element)
-        if self.callbacks:
-            self.callbacks(self)
-            self.callbacks.update(self)
+
+        if not self.batched:
+            for cb in self.callbacks:
+                cb.initialize()
+
         if not self.overlaid:
             self._process_legend()
         self.drawn = True
@@ -571,7 +613,7 @@ def update_frame(self, key, ranges=None, plot=None, element=None, empty=False):
 
         plot = self.handles['plot']
         source = self.handles['source']
-        empty = (self.callbacks and self.callbacks.downsample) or empty
+        empty = False
         if self.batched:
             data, mapping = self.get_batched_data(element, ranges, empty)
         else:
@@ -585,8 +627,6 @@ def update_frame(self, key, ranges=None, plot=None, element=None, empty=False):
         if not self.overlaid:
             self._update_ranges(style_element, ranges)
             self._update_plot(key, plot, style_element)
-        if self.callbacks:
-            self.callbacks.update(self)
 
 
     @property
@@ -933,14 +973,14 @@ def _init_tools(self, element):
         tools = []
         hover = False
         for key, subplot in self.subplots.items():
-                el = element.get(key)
-                if el is not None:
-                    el_tools = subplot._init_tools(el)
-                    el_tools = [t for t in el_tools
-                                if not (isinstance(t, HoverTool) and hover)]
-                    tools += el_tools
-                    if any(isinstance(t, HoverTool) for t in el_tools):
-                        hover = True
+            el = element.get(key)
+            if el is not None:
+                el_tools = subplot._init_tools(el, self.callbacks)
+                el_tools = [t for t in el_tools
+                            if not (isinstance(t, HoverTool) and hover)]
+                tools += el_tools
+                if any(isinstance(t, HoverTool) for t in el_tools):
+                    hover = True
         return list(set(tools))
 
 
@@ -978,6 +1018,9 @@ def initialize_plot(self, ranges=None, plot=None, plots=None):
             self._process_legend()
         self.drawn = True
 
+        for cb in self.callbacks:
+            cb.initialize()
+
         return self.handles['plot']
 
 
diff --git a/holoviews/plotting/bokeh/path.py b/holoviews/plotting/bokeh/path.py
index 09930009ce..ada06529d2 100644
--- a/holoviews/plotting/bokeh/path.py
+++ b/holoviews/plotting/bokeh/path.py
@@ -49,25 +49,6 @@ class PolygonPlot(ColorbarPlot, PathPlot):
     style_opts = ['color', 'cmap', 'palette'] + line_properties + fill_properties
     _plot_methods = dict(single='patches', batched='patches')
 
-    def _init_tools(self, element):
-        """
-        Processes the list of tools to be supplied to the plot.
-        """
-        tools = self.default_tools + self.tools
-        if 'hover' not in tools:
-            return tools
-        tools.pop(tools.index('hover'))
-        if self.batched:
-            dims = self.hmap.last.kdims
-        else:
-            dims = list(self.overlay_dims.keys())
-        dims += element.vdims
-        tooltips = [(d.pprint_label, '@'+util.dimension_sanitizer(d.name))
-                    for d in dims]
-        tools.append(HoverTool(tooltips=tooltips))
-        return tools
-
-
     def get_data(self, element, ranges=None, empty=False):
         xs = [] if empty else [path[:, 0] for path in element.data]
         ys = [] if empty else [path[:, 1] for path in element.data]
diff --git a/holoviews/plotting/bokeh/plot.py b/holoviews/plotting/bokeh/plot.py
index cfc509fb86..98df9e30ce 100644
--- a/holoviews/plotting/bokeh/plot.py
+++ b/holoviews/plotting/bokeh/plot.py
@@ -48,6 +48,7 @@ class BokehPlot(DimensionedPlot):
     def document(self):
         return self._document
 
+
     @document.setter
     def document(self, doc):
         self._document = doc
@@ -55,11 +56,13 @@ def document(self, doc):
             for plot in self.subplots.values():
                 plot.document = doc
 
+
     def __init__(self, *args, **params):
         super(BokehPlot, self).__init__(*args, **params)
         self._document = None
         self.root = None
 
+
     def get_data(self, element, ranges=None, empty=False):
         """
         Returns the data from an element in the appropriate format for
@@ -70,14 +73,6 @@ def get_data(self, element, ranges=None, empty=False):
         raise NotImplementedError
 
 
-    def set_document(self, document):
-        """
-        Sets the current document on all subplots.
-        """
-        for plot in self.traverse(lambda x: x):
-            plot.document = document
-
-
     def set_root(self, root):
         """
         Sets the current document on all subplots.
diff --git a/holoviews/plotting/bokeh/util.py b/holoviews/plotting/bokeh/util.py
index 3339a7682b..37b1d3feb0 100644
--- a/holoviews/plotting/bokeh/util.py
+++ b/holoviews/plotting/bokeh/util.py
@@ -16,7 +16,7 @@
 from bokeh.core.json_encoder import serialize_json # noqa (API import)
 from bokeh.document import Document
 from bokeh.models.plots import Plot
-from bokeh.models import GlyphRenderer
+from bokeh.models import GlyphRenderer, Model, HasProps
 from bokeh.models.widgets import DataTable, Tabs
 from bokeh.plotting import Figure
 if bokeh_version >= '0.12':
@@ -146,36 +146,6 @@ def convert_datetime(time):
     return time.astype('datetime64[s]').astype(float)*1000
 
 
-def models_to_json(models):
-    """
-    Convert list of bokeh models into json to update plot(s).
-    """
-    json_data, ids = [], []
-    for plotobj in models:
-        if plotobj.ref['id'] in ids:
-            continue
-        else:
-            ids.append(plotobj.ref['id'])
-        json = plotobj.to_json(False)
-        json.pop('tool_events', None)
-        json.pop('renderers', None)
-        json_data.append({'id': plotobj.ref['id'],
-                          'type': plotobj.ref['type'],
-                          'data': json})
-    return json_data
-
-
-def refs(json):
-    """
-    Finds all the references to other objects in the json
-    representation of a bokeh Document.
-    """
-    result = {}
-    for obj in json['roots']['references']:
-        result[obj['id']] = obj
-    return result
-
-
 def get_ids(obj):
     """
     Returns a list of all ids in the supplied object.  Useful for
@@ -193,6 +163,40 @@ def get_ids(obj):
     return list(itertools.chain(*ids))
 
 
+def replace_models(obj):
+    """
+    Recursively processes references, replacing Models with there .ref
+    values and HasProps objects with their property values.
+    """
+    if isinstance(obj, Model):
+        return obj.ref
+    elif isinstance(obj, HasProps):
+        return obj.properties_with_values(include_defaults=False)
+    elif isinstance(obj, dict):
+        return {k: replace_models(v) for k, v in obj.items()}
+    elif isinstance(obj, list):
+        return [replace_models(v) for v in obj]
+    else:
+        return obj
+
+
+def to_references(doc):
+    """
+    Convert the document to a dictionary of references. Avoids
+    unnecessary JSON serialization/deserialization within Python and
+    the corresponding performance penalty.
+    """
+    root_ids = []
+    for r in doc._roots:
+        root_ids.append(r._id)
+
+    references = {}
+    for obj in doc._references_json(doc._all_models.values()):
+        obj = replace_models(obj)
+        references[obj['id']] = obj
+    return references
+
+
 def compute_static_patch(document, models):
     """
     Computes a patch to update an existing document without
@@ -215,7 +219,7 @@ def compute_static_patch(document, models):
     ensure that only the references between objects are sent without
     duplicating any of the data.
     """
-    references = refs(document.to_json())
+    references = to_references(document)
     model_ids = [m.ref['id'] for m in models]
 
     requested_updates = []
diff --git a/holoviews/plotting/comms.py b/holoviews/plotting/comms.py
index 2057bd0d56..a4711f064a 100644
--- a/holoviews/plotting/comms.py
+++ b/holoviews/plotting/comms.py
@@ -1,11 +1,10 @@
 import uuid
 
-import param
 from ipykernel.comm import Comm as IPyComm
 from IPython import get_ipython
 
 
-class Comm(param.Parameterized):
+class Comm(object):
     """
     Comm encompasses any uni- or bi-directional connection between
     a python process and a frontend allowing passing of messages
@@ -119,7 +118,7 @@ def send(self, data):
 
 
 
-class JupyterCommJS(Comm):
+class JupyterCommJS(JupyterComm):
     """
     JupyterCommJS provides a comms channel for the Jupyter notebook,
     which is initialized on the frontend. This allows sending events
@@ -149,7 +148,7 @@ def __init__(self, plot, target=None, on_msg=None):
         """
         Initializes a Comms object
         """
-        super(JupyterComm, self).__init__(plot, target, on_msg)
+        super(JupyterCommJS, self).__init__(plot, target, on_msg)
         self.manager = get_ipython().kernel.comm_manager
         self.manager.register_target(self.target, self._handle_open)
 
diff --git a/holoviews/plotting/widgets/widgets.js b/holoviews/plotting/widgets/widgets.js
index 0d9add46e5..2878fe0b09 100644
--- a/holoviews/plotting/widgets/widgets.js
+++ b/holoviews/plotting/widgets/widgets.js
@@ -1,6 +1,8 @@
 function HoloViewsWidget(){
 }
 
+HoloViewsWidget.comms = {};
+
 HoloViewsWidget.prototype.init_slider = function(init_val){
 	if(this.load_json) {
 		this.from_json()
diff --git a/holoviews/streams.py b/holoviews/streams.py
index 6e66cab908..53f18df704 100644
--- a/holoviews/streams.py
+++ b/holoviews/streams.py
@@ -5,8 +5,7 @@
 """
 
 import param
-import uuid
-from collections import OrderedDict
+from collections import defaultdict
 from .core import util
 
 
@@ -74,8 +73,12 @@ class Stream(param.Parameterized):
     the parameter dictionary when the trigger classmethod is called.
     """
 
-    # Mapping from uuid to stream instance
-    registry = OrderedDict()
+    # Mapping from a source id to a list of streams
+    registry = defaultdict(list)
+
+    # Mapping to define callbacks by backend and Stream type.
+    # e.g. Stream._callbacks['bokeh'][Stream] = Callback
+    _callbacks = defaultdict(dict)
 
     @classmethod
     def trigger(cls, streams):
@@ -104,14 +107,6 @@ def trigger(cls, streams):
             subscriber(**dict(union))
 
 
-    @classmethod
-    def find(cls, obj):
-        """
-        Return a set of streams from the registry with a given source.
-        """
-        return set(v for v in cls.registry.values() if v.source is obj)
-
-
     def __init__(self, preprocessors=[], source=None, subscribers=[], **params):
         """
         Mapping allows multiple streams with similar event state to be
@@ -121,14 +116,25 @@ def __init__(self, preprocessors=[], source=None, subscribers=[], **params):
         datastructure that the stream receives events from, as supported
         by the plotting backend.
         """
-        self.source = source
+        self._source = source
         self.subscribers = subscribers
         self.preprocessors = preprocessors
         self._hidden_subscribers = []
 
-        self.uuid = uuid.uuid4().hex
         super(Stream, self).__init__(**params)
-        self.registry[self.uuid] = self
+        if source:
+            self.registry[id(source)].append(self)
+
+    @property
+    def source(self):
+        return self._source
+
+    @source.setter
+    def source(self, source):
+        if self._source:
+            raise Exception('source has already been defined on stream.')
+        self._source = source
+        self.registry[id(source)].append(self)
 
 
     @property
@@ -184,10 +190,6 @@ class PositionX(Stream):
     x = param.Number(default=0, doc="""
            Position along the x-axis in data coordinates""", constant=True)
 
-    def __init__(self, preprocessors=[], source=None, subscribers=[], **params):
-        super(PositionX, self).__init__(preprocessors=preprocessors, source=source,
-                                        subscribers=subscribers, **params)
-
 
 class PositionY(Stream):
     """
@@ -200,10 +202,6 @@ class PositionY(Stream):
     y = param.Number(default=0, doc="""
            Position along the y-axis in data coordinates""", constant=True)
 
-    def __init__(self, preprocessors=[], source=None, subscribers=[], **params):
-        super(PositionY, self).__init__(preprocessors=preprocessors, source=source,
-                                        subscribers=subscribers, **params)
-
 
 class PositionXY(Stream):
     """
@@ -213,18 +211,61 @@ class PositionXY(Stream):
     position of the mouse/trackpad cursor.
     """
 
-
     x = param.Number(default=0, doc="""
            Position along the x-axis in data coordinates""", constant=True)
 
     y = param.Number(default=0, doc="""
            Position along the y-axis in data coordinates""", constant=True)
 
-    def __init__(self, preprocessors=[], source=None, subscribers=[], **params):
-        super(PositionXY, self).__init__(preprocessors=preprocessors, source=source,
-                                        subscribers=subscribers, **params)
 
+class RangeXY(Stream):
+    """
+    Axis ranges along x- and y-axis in data coordinates.
+    """
+
+    x_range = param.NumericTuple(default=(0, 1), constant=True, doc="""
+      Range of the x-axis of a plot in data coordinates""")
+
+    y_range = param.NumericTuple(default=(0, 1), constant=True, doc="""
+      Range of the y-axis of a plot in data coordinates""")
+
+
+class RangeX(Stream):
+    """
+    Axis range along x-axis in data coordinates.
+    """
+
+    x_range = param.NumericTuple(default=(0, 1), constant=True, doc="""
+      Range of the x-axis of a plot in data coordinates""")
+
+
+class RangeY(Stream):
+    """
+    Axis range along y-axis in data coordinates.
+    """
+
+    y_range = param.NumericTuple(default=(0, 1), constant=True, doc="""
+      Range of the y-axis of a plot in data coordinates""")
+
+
+class Bounds(Stream):
+    """
+    A stream representing the bounds of a box selection as an
+    tuple of the left, bottom, right and top coordinates.
+    """
+
+    bounds = param.NumericTuple(default=(0, 0, 1, 1), constant=True,
+                                doc="""
+        Bounds defined as (left, bottom, top, right) tuple.""")
+
+
+class Selection1D(Stream):
+    """
+    A stream representing a 1D selection of objects by their index.
+    """
 
+    index = param.List(default=[], doc="""
+        Indices into a 1D datastructure.""")
 
 
 class ParamValues(Stream):
diff --git a/tests/testplotinstantiation.py b/tests/testplotinstantiation.py
index 12293f8e35..b12bcd15d2 100644
--- a/tests/testplotinstantiation.py
+++ b/tests/testplotinstantiation.py
@@ -13,13 +13,13 @@
                                Scatter3D)
 from holoviews.element.comparison import ComparisonTestCase
 from holoviews.streams import PositionXY
+from holoviews.plotting import comms
 
 # Standardize backend due to random inconsistencies
 try:
     from matplotlib import pyplot
     pyplot.switch_backend('agg')
     from holoviews.plotting.mpl import OverlayPlot
-    from holoviews.plotting.comms import Comm
     mpl_renderer = Store.renderers['matplotlib']
 except:
     mpl_renderer = None
@@ -27,6 +27,7 @@
 try:
     import holoviews.plotting.bokeh
     bokeh_renderer = Store.renderers['bokeh']
+    from holoviews.plotting.bokeh.callbacks import Callback
     from bokeh.models.mappers import LinearColorMapper, LogColorMapper
 except:
     bokeh_renderer = None
@@ -46,7 +47,7 @@ def setUp(self):
         if mpl_renderer is None:
             raise SkipTest("Matplotlib required to test plot instantiation")
         self.default_comm, _ = mpl_renderer.comms['default']
-        mpl_renderer.comms['default'] = (Comm, '')
+        mpl_renderer.comms['default'] = (comms.Comm, '')
 
     def teardown(self):
         mpl_renderer.comms['default'] = (self.default_comm, '')
@@ -92,13 +93,17 @@ class TestBokehPlotInstantiation(ComparisonTestCase):
 
     def setUp(self):
         self.previous_backend = Store.current_backend
-        Store.current_backend = 'bokeh'
-
         if not bokeh_renderer:
             raise SkipTest("Bokeh required to test plot instantiation")
+        Store.current_backend = 'bokeh'
+        Callback._comm_type = comms.Comm
+        self.default_comm, _ = bokeh_renderer.comms['default']
+        bokeh_renderer.comms['default'] = (comms.Comm, '')
 
     def teardown(self):
         Store.current_backend = self.previous_backend
+        Callback._comm_type = comms.JupyterCommJS
+        mpl_renderer.comms['default'] = (self.default_comm, '')
 
     def test_batched_plot(self):
         overlay = NdOverlay({i: Points(np.arange(i)) for i in range(1, 100)})
@@ -141,6 +146,16 @@ def test_spikes_colormapping(self):
         self._test_colormapping(spikes, 1)
 
 
+    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}')
+        data = plot.handles['source'].data
+        self.assertEqual(data['x'], np.array([10]))
+        self.assertEqual(data['y'], np.array([-10]))
+
+
 class TestPlotlyPlotInstantiation(ComparisonTestCase):
 
     def setUp(self):