iframe-proxy | Sunbelt Computer Software

This repository was archived by the owner on Mar 23, 2026. It is now read-only.
187 lines (148 loc) · 5.85 KB
import logging
import threading
from localstack.utils.net import is_port_open
from localstack.utils.sync import poll_condition
from localstack.utils.threads import FuncThread, start_thread
LOG = logging.getLogger(__name__)
class StopServer(Exception):
class Server(abc.ABC):
    A Server implements the lifecycle of a server running in a thread.
    def __init__(self, port: int, host: str = "localhost") -> None:
        super().__init__()
        self._thread: FuncThread | None = None
        self._lifecycle_lock = threading.RLock()
        self._stopped = threading.Event()
        self._started = threading.Event()
        self._host = host
        self._port = port
    @property
    def host(self):
        return self._host
    @property
    def port(self):
        return self._port
    @property
    def protocol(self):
        return "http"
    @property
    def url(self):
        return f"{self.protocol}://{self.host}:{self.port}"
    def get_error(self) -> Exception | None:
        """
        If the thread running the server returned with an Exception, then this function will return that exception.
        """
        if not self._started.is_set():
            return None
        future = self._thread.result_future
        if future.done():
            return future.exception()
        return None
    def wait_is_up(self, timeout: float = None) -> bool:
        """
        Waits until the server is started and is_up returns true.
        :param timeout: the time in seconds to wait before returning false. If timeout is None, then wait indefinitely.
        :returns: true if the server is up, false if not or the timeout was reached while waiting.
        """
        # first wait until the started event was called
        self._started.wait(timeout=timeout)
        # then poll the health check
        return poll_condition(self.is_up, timeout=timeout)
    def is_running(self) -> bool:
        """
        Checks whether the thread holding the server is running. The server may be running but not healthy (
        is_running == True, is_up == False).
        :returns: true if the server thread is running
        """
        if not self._started.is_set():
            return False
        if self._stopped.is_set():
            return False
        return self._thread.running
    def is_up(self) -> bool:
        """
        Checks whether the server is up by executing the health check function.
        :returns: false if the server has not been started or if the health check failed, true otherwise
        """
        if not self._started.is_set():
            return False
        try:
            return True if self.health() else False
        except Exception:
            return False
    def shutdown(self) -> None:
        """
        Attempts to shut down the server by calling the internal do_shutdown method. It only does this if the server
        has been started. Repeated calls to this function have no effect.
        :raises RuntimeError: shutdown was called before start
        """
        with self._lifecycle_lock:
            if not self._started.is_set():
                raise RuntimeError("cannot shutdown server before it is started")
            if self._stopped.is_set():
                return
            self._thread.stop()
            self._stopped.set()
            self.do_shutdown()
    def start(self) -> bool:
        """
        Starts the server by calling the internal do_run method in a new thread, and then returns True. Repeated
        calls to this function have no effect but return False.
        :return: True if the server was started in this call, False if the server was already started previously
        """
        with self._lifecycle_lock:
            if self._started.is_set():
                return False
            self._thread = self.do_start_thread()
            self._started.set()
            return True
    def join(self, timeout=None):
        """
        Waits for the given amount of time until the thread running the server returns. If the server hasn't started
        yet, it first waits for the server to start.
        :params: the time in seconds to wait. If None then wait indefinitely.
        :raises TimeoutError: If the server didn't shut down before the given timeout.
        """
        if not self._started.is_set():
            raise RuntimeError("cannot join server before it is started")
        if not self._started.wait(timeout):
            raise TimeoutError
        try:
            self._thread.result_future.result(timeout)
        except TimeoutError:
            raise
        except Exception:
            # Future.result() will re-raise the exception that was raised in the thread
            return
    def health(self):
        """
        Runs a health check on the server. The default implementation performs is_port_open on the server URL.
        """
        return is_port_open(self.url)
    def do_start_thread(self) -> FuncThread:
        """
        Creates and starts the thread running the server. By default, it calls the do_run method in a FuncThread, but
        can be overridden to if the subclass wants to return its own thread.
        """
        def _run(*_):
            try:
                return self.do_run()
            except StopServer:
                LOG.debug("stopping server %s", self.url)
            finally:
                self._stopped.set()
        return start_thread(_run, name=f"server-{self.__class__.__name__}")
    def do_run(self):
        """
        Runs the server (blocking method). (Needs to be overridden by subclasses of do_start_thread is not overridden).
        :raises StopServer: can be raised by the subclass to indicate the server should be stopped.
        """
        pass
    def do_shutdown(self):
        """
        Called when shutdown() is performed. (Should be overridden by subclasses).
        """
        pass
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Uh oh!

FilesExpand file tree

serving.py

Latest commit

History

serving.py

File metadata and controls
