def __stream(self, endpoint, listener, params={}, run_async=False, timeout=__DEFAULT_STREAM_TIMEOUT, reconnect_async=False, reconnect_async_wait_sec=__DEFAULT_STREAM_RECONNECT_WAIT_SEC):
        """
        Internal streaming API helper.

        Returns a handle to the open connection that the user can close if they
        wish to terminate it.
        """

        # Check if we have to redirect
        instance = self.instance()
        if "streaming_api" in instance["urls"] and instance["urls"]["streaming_api"] != self.api_base_url:
            # This is probably a websockets URL, which is really for the browser, but requests can't handle it
            # So we do this below to turn it into an HTTPS or HTTP URL
            parse = urlparse(instance["urls"]["streaming_api"])
            if parse.scheme == 'wss':
                url = "https://" + parse.netloc
            elif parse.scheme == 'ws':
                url = "http://" + parse.netloc
            else:
                raise MastodonAPIError(
                        "Could not parse streaming api location returned from server: {}.".format(
                            instance["urls"]["streaming_api"]))
        else:
            url = self.api_base_url

        # The streaming server can't handle two slashes in a path, so remove trailing slashes
        if url[-1] == '/':
            url = url[:-1]
        
        # Connect function (called and then potentially passed to async handler)
        def connect_func():
            headers = {"Authorization": "Bearer " + self.access_token}
            connection = self.session.get(url + endpoint, headers = headers, data = params, stream = True,
                                  timeout=(self.request_timeout, timeout))

            if connection.status_code != 200:
                raise MastodonNetworkError("Could not connect to streaming server: %s" % connection.reason)
            return connection
        connection = None
        
        # Async stream handler
        class __stream_handle():
            def __init__(self, connection, connect_func, reconnect_async, reconnect_async_wait_sec):
                self.closed = False
                self.running = True
                self.connection = connection
                self.connect_func = connect_func
                self.reconnect_async = reconnect_async
                self.reconnect_async_wait_sec = reconnect_async_wait_sec
                self.reconnecting = False
                
            def close(self):
                self.closed = True
                self.connection.close()

            def is_alive(self):
                return self._thread.is_alive()

            def is_receiving(self):
                if self.closed or not self.running or self.reconnecting or not self.is_alive():
                    return False
                else:
                    return True

            def _threadproc(self):
                self._thread = threading.current_thread()
                
                # Run until closed or until error if not autoreconnecting
                while self.running:
                    if not self.connection is None:
                        with closing(self.connection) as r:
                            try:
                                listener.handle_stream(r)
                            except (AttributeError, MastodonMalformedEventError, MastodonNetworkError) as e:
                                if not (self.closed or self.reconnect_async):
                                    raise e
                                else:
                                    if self.closed:
                                        self.running = False

                    # Reconnect loop. Try immediately once, then with delays on error.
                    if (self.reconnect_async and not self.closed) or self.connection is None:
                        self.reconnecting = True
                        connect_success = False
                        while not connect_success:
                            connect_success = True
                            try:
                                self.connection = self.connect_func()
                                if self.connection.status_code != 200:
                                    time.sleep(self.reconnect_async_wait_sec)
                                    connect_success = False
                                    exception = MastodonNetworkError("Could not connect to server.")
                                    listener.on_abort(exception)
                            except:
                                time.sleep(self.reconnect_async_wait_sec)
                                connect_success = False
                        self.reconnecting = False
                    else:
                        self.running = False
                return 0

        if run_async:
            handle = __stream_handle(connection, connect_func, reconnect_async, reconnect_async_wait_sec)
            t = threading.Thread(args=(), target=handle._threadproc)
            t.daemon = True
            t.start()
            return handle
        else:
            # Blocking, never returns (can only leave via exception)
            connection = connect_func()            
            with closing(connection) as r:
                listener.handle_stream(r)
Internal streaming API helper.

        Returns a handle to the open connection that the user can close if they
        wish to terminate it.