How to use the portend.occupied function in portend

To help you get started, we’ve selected a few portend examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github coady / lupyne / tests / test_server.py View on Github external
def start(self, port, *args, **config):
        config.update(self.config)
        config['server.socket_port'] = port
        portend.free('localhost', port)
        server = self[port] = subprocess.Popen((sys.executable, '-m', self.module, '-c', json.dumps(config)) + args)
        portend.occupied('localhost', port)
        server.started = time.time()
        assert not server.poll()
        return clients.Resource('http://localhost:{}'.format(port))
github cherrypy / cherrypy / cherrypy / process / servers.py View on Github external
def wait(self):
        """Wait until the HTTP server is ready to receive requests."""
        while not getattr(self.httpserver, 'ready', False):
            if self.interrupt:
                raise self.interrupt
            time.sleep(.1)

        # Wait for port to be occupied
        if not os.environ.get('LISTEN_PID', None):
            # Wait for port to be occupied if not running via socket-activation
            # (for socket-activation the port will be managed by systemd )
            if isinstance(self.bind_addr, tuple):
                with _safe_wait(*self.bound_addr):
                    portend.occupied(*self.bound_addr, timeout=Timeouts.occupied)
github Southpaw-TACTIC / TACTIC / 3rd_party / python2 / site-packages / cherrypy / process / servers.py View on Github external
if self.interrupt:
                raise self.interrupt
            time.sleep(.1)

        # bypass check when LISTEN_PID is set
        if os.environ.get('LISTEN_PID', None):
            return

        # bypass check when running via socket-activation
        # (for socket-activation the port will be managed by systemd)
        if not isinstance(self.bind_addr, tuple):
            return

        # wait for port to be occupied
        with _safe_wait(*self.bound_addr):
            portend.occupied(*self.bound_addr, timeout=Timeouts.occupied)

portend

TCP port monitoring and discovery

MIT
Latest version published 1 year ago

Package Health Score

55 / 100
Full package analysis