Create a new stream subscription.

        :param str stream: The name of the stream.
        :param on_data: Function that gets called with  :class:`.StreamData`
                        updates.
        :param float timeout: The amount of seconds to wait for the request
                              to complete.
        :return: Future that can be used to manage the background websocket
                 subscription
        :rtype: .WebSocketSubscriptionFuture

def create_stream_subscription(self, stream, on_data, timeout=60):
        """
        Create a new stream subscription.

        :param str stream: The name of the stream.
        :param on_data: Function that gets called with  :class:`.StreamData`
                        updates.
        :param float timeout: The amount of seconds to wait for the request
                              to complete.
        :return: Future that can be used to manage the background websocket
                 subscription
        :rtype: .WebSocketSubscriptionFuture
        """
        options = rest_pb2.StreamSubscribeRequest()
        options.stream = stream

        manager = WebSocketSubscriptionManager(
            self._client, resource='stream', options=options)

        # Represent subscription as a future
        subscription = WebSocketSubscriptionFuture(manager)

        wrapped_callback = functools.partial(
            _wrap_callback_parse_stream_data, subscription, on_data)

        manager.open(wrapped_callback, instance=self._instance)

        # Wait until a reply or exception is received
        subscription.reply(timeout=timeout)

        return subscription