Begin consuming messages.

        :param string instance: (Optional) instance to use in the WebSocket URL
        :param string processor: (Optional) processor to use in the WebSocket URL

def open(self, callback, instance=None, processor=None):
        """
        Begin consuming messages.

        :param string instance: (Optional) instance to use in the WebSocket URL
        :param string processor: (Optional) processor to use in the WebSocket URL
        """
        assert not self._closed

        ws_url = self._client.ws_root
        if instance:
            ws_url += '/' + instance
            if processor:
                ws_url += '/' + processor

        self._callback = callback
        self._websocket = websocket.WebSocketApp(
            ws_url,
            on_open=self._on_websocket_open,
            on_message=self._on_websocket_message,
            on_error=self._on_websocket_error,
            subprotocols=['protobuf'],
            header=[
                '{}: {}'.format(k, self._client.session.headers[k])
                for k in self._client.session.headers
            ],
        )
        self._consumer = threading.Thread(target=self._websocket.run_forever)

        # Running this as a daemon thread improves possibilities
        # for consumers of our API to control shutdown.
        # (example: can just use time.sleep on the main thread instead of blocking on the future)
        self._consumer.daemon = True

        self._consumer.start()