Python twisted.internet.endpoints 模块，TCP4ServerEndpoint() 实例源码

def createServerEndpoint(self, reactor, factory, **listenArgs):
        """
        Create an L{TCP4ServerEndpoint} and return the values needed to verify
        its behaviour.

        @param reactor: A fake L{IReactorTCP} that L{TCP4ServerEndpoint} can
            call L{IReactorTCP.listenTCP} on.
        @param factory: The thing that we expect to be passed to our
            L{IStreamServerEndpoint.listen} implementation.
        @param listenArgs: Optional dictionary of arguments to
            L{IReactorTCP.listenTCP}.
        """
        address = IPv4Address("TCP", "0.0.0.0", 0)

        if listenArgs is None:
            listenArgs = {}

        return (endpoints.TCP4ServerEndpoint(reactor,
                                             address.port,
                                             **listenArgs),
                (address.port, factory,
                 listenArgs.get('backlog', 50),
                 listenArgs.get('interface', '')),
                address)

def connectableEndpoint(debug=False):
    """
    Create an endpoint that can be fired on demand.

    @param debug: A flag; whether to dump output from the established
        connection to stdout.
    @type debug: L{bool}

    @return: A client endpoint, and an object that will cause one of the
        L{Deferred}s returned by that client endpoint.
    @rtype: 2-L{tuple} of (L{IStreamClientEndpoint}, L{ConnectionCompleter})
    """
    reactor = MemoryReactorClock()
    clientEndpoint = TCP4ClientEndpoint(reactor, "0.0.0.0", 4321)
    serverEndpoint = TCP4ServerEndpoint(reactor, 4321)
    serverEndpoint.listen(Factory.forProtocol(Protocol))
    return clientEndpoint, ConnectionCompleter(reactor)

def test_service(self):
        """
        L{strports.service} returns a L{StreamServerEndpointService}
        constructed with an endpoint produced from
        L{endpoint.serverFromString}, using the same syntax.
        """
        reactor = object() # the cake is a lie
        aFactory = Factory()
        aGoodPort = 1337
        svc = strports.service(
            'tcp:' + str(aGoodPort), aFactory, reactor=reactor)
        self.assertIsInstance(svc, internet.StreamServerEndpointService)

        # See twisted.application.test.test_internet.EndpointServiceTests.
        # test_synchronousRaiseRaisesSynchronously
        self.assertTrue(svc._raiseSynchronously)
        self.assertIsInstance(svc.endpoint, TCP4ServerEndpoint)
        # Maybe we should implement equality for endpoints.
        self.assertEqual(svc.endpoint._port, aGoodPort)
        self.assertIs(svc.factory, aFactory)
        self.assertIs(svc.endpoint._reactor, reactor)

def test_start_up_binds_first_of_real_endpoint_options(self):
        service = RegionService(sentinel.advertiser)

        # endpoint_1.listen(...) will bind to a random high-numbered port.
        endpoint_1 = TCP4ServerEndpoint(reactor, 0)
        # endpoint_2.listen(...), if attempted, will crash because only root
        # (or a user with explicit capabilities) can do stuff like that. It's
        # a reasonable assumption that the user running these tests is not
        # root, but we'll check the port number later too to be sure.
        endpoint_2 = TCP4ServerEndpoint(reactor, 1)

        service.endpoints = [[endpoint_1, endpoint_2]]

        yield service.startService()
        self.addCleanup(wait_for_reactor(service.stopService))

        # A single port has been bound.
        self.assertThat(service.ports, MatchesAll(
            HasLength(1), AllMatch(IsInstance(tcp.Port))))

        # The port is not listening on port 1; i.e. a belt-n-braces check that
        # endpoint_2 was not used.
        [port] = service.ports
        self.assertThat(port.getHost().port, Not(Equals(1)))

def _init_manhole(self, cfg):
        port = cfg['port']
        user, passwd = cfg['user'], cfg['passwd']
        sshFactory = manhole.getManholeFactory(
            {'core': self}, user, passwd)
        endpoint = TCP4ServerEndpoint(reactor, port)
        endpoint.listen(sshFactory)

        log.info('Started manhole in PORT {0!s}'.format(port))

def main():
    try:
        # Create end point
        endpoint = TCP4ServerEndpoint(
            reactor=reactor,
            port=8000,
            interface='127.0.0.1',
        )

        # Start listening
        endpoint.listen(Factory.forProtocol(EchoProtocol))

        # Run reactor
        reactor.run()

    # If have `KeyboardInterrupt`
    except KeyboardInterrupt:
        # Stop gracefully
        pass


# Trace calls in this module.
#
# Calling this function is needed because at the point `trace_calls_in_specs`
# is called, this module is being initialized, therefore callables defined
# after the call point are not accessible to `trace_calls_in_specs`.
#

def startService(self):
        """
        Register ourselves with the database and establish all outgoing
        connections to other servers in the cluster.
        """
        @inlineCallbacks
        def startup(txn):
            endpoint = TCP4ServerEndpoint(self.reactor, self.ampPort)
            # If this fails, the failure mode is going to be ugly, just like
            # all conflicted-port failures.  But, at least it won't proceed.
            self._listeningPort = yield endpoint.listen(self.peerFactory())
            self.ampPort = self._listeningPort.getHost().port
            yield Lock.exclusive(NodeInfo.table).on(txn)
            nodes = yield self.activeNodes(txn)
            selves = [node for node in nodes
                      if ((node.hostname == self.hostname) and
                          (node.port == self.ampPort))]
            if selves:
                self.thisProcess = selves[0]
                nodes.remove(self.thisProcess)
                yield self.thisProcess.update(pid=self.pid,
                                              time=datetime.now())
            else:
                self.thisProcess = yield NodeInfo.create(
                    txn, hostname=self.hostname, port=self.ampPort,
                    pid=self.pid, time=datetime.now()
                )

            for node in nodes:
                self._startConnectingTo(node)

        self._startingUp = inTransaction(self.transactionFactory, startup)

        @self._startingUp.addBoth
        def done(result):
            self._startingUp = None
            super(PeerConnectionPool, self).startService()
            self._lostWorkCheckLoop()
            return result

def start(self):
        log.debug('starting')

        # setup a basic web server for test control
        self.control_endpoint = endpoints.TCP4ServerEndpoint(reactor, 18880)
        self.control_endpoint.listen(self.get_test_control_site())

        # TODO tmp: populate some devices and logical devices
        # reactor.callLater(0, self._tmp_populate_stuff)
        log.info('started')

def init_rest_service(port):
    hc = HealthCheck()
    endpoint = endpoints.TCP4ServerEndpoint(reactor, port)
    endpoint.listen(hc.get_site())

def server(self, reactor):
        """
        Create a server-side TCP endpoint.
        """
        return TCP4ServerEndpoint(reactor, 0, interface=self.interface)

def test_tcp(self):
        """
        When passed a TCP strports description, L{endpoints.serverFromString}
        returns a L{TCP4ServerEndpoint} instance initialized with the values
        from the string.
        """
        reactor = object()
        server = endpoints.serverFromString(
            reactor, "tcp:1234:backlog=12:interface=10.0.0.1")
        self.assertIsInstance(server, endpoints.TCP4ServerEndpoint)
        self.assertIs(server._reactor, reactor)
        self.assertEqual(server._port, 1234)
        self.assertEqual(server._backlog, 12)
        self.assertEqual(server._interface, "10.0.0.1")

def _endpointTest(self, service):
        """
        Use L{Options} to parse a single service configuration parameter and
        verify that an endpoint of the correct type is added to the list for
        that service.
        """
        options = Options()
        options.parseOptions(['--' + service, 'tcp:1234'])
        self.assertEqual(len(options[service]), 1)
        self.assertIsInstance(
            options[service][0], endpoints.TCP4ServerEndpoint)

def test_protoDefaults(self):
        """
        POP3 and SMTP each listen on a TCP4ServerEndpoint by default.
        """
        options = Options()
        options.parseOptions([])

        self.assertEqual(len(options['pop3']), 1)
        self.assertIsInstance(
            options['pop3'][0], endpoints.TCP4ServerEndpoint)

        self.assertEqual(len(options['smtp']), 1)
        self.assertIsInstance(
            options['smtp'][0], endpoints.TCP4ServerEndpoint)

def start_fake_riak_server(self, stats):
        def cb(listener):
            self.addCleanup(listener.stopListening)
            return listener

        data = static.Data(json.dumps(stats).encode(), 'application/json')

        data.isLeaf = True
        site = server.Site(data)
        endpoint = endpoints.TCP4ServerEndpoint(reactor, 0)
        return endpoint.listen(site).addCallback(cb)

def start_riemann_server(self):
        factory = FakeRiemannServerFactory()
        self.addCleanup(factory.stop_listening)
        return factory.start_listening(TCP4ServerEndpoint(reactor, 0))

def setUp(self):
        super(TestNettestTimeout, self).setUp()
        from twisted.internet.protocol import Protocol, Factory
        from twisted.internet.endpoints import TCP4ServerEndpoint

        class DummyProtocol(Protocol):

            def dataReceived(self, data):
                pass

        class DummyFactory(Factory):

            def __init__(self):
                self.protocols = []

            def buildProtocol(self, addr):
                proto = DummyProtocol()
                self.protocols.append(proto)
                return proto

            def stopFactory(self):
                for proto in self.protocols:
                    proto.transport.loseConnection()

        self.factory = DummyFactory()
        endpoint = TCP4ServerEndpoint(reactor, 8007)
        self.port = yield endpoint.listen(self.factory)

        config.advanced.measurement_timeout = 2

def _start_server(self, launch_tor):
        self._ui.notify_bootstrap(
            notifications.UnmessageNotification('Configuring local server'))

        endpoint = TCP4ServerEndpoint(self._twisted_reactor,
                                      self._port_local_server,
                                      interface=self._ip_local_server)
        self._twisted_server_endpoint = endpoint

        d = Deferred()

        def endpoint_listening(port):
            self._ui.notify_bootstrap(
                notifications.UnmessageNotification('Running local server'))

            if self._local_mode:
                d.callback(None)
            else:
                d_tor = self._start_tor(launch_tor)
                d_tor.addCallbacks(d.callback, d.errback)

        self._twisted_factory = _ConversationFactory(
            peer=self,
            connection_made=self._add_intro_manager)

        d_server = endpoint.listen(self._twisted_factory)
        d_server.addCallbacks(endpoint_listening, d.errback)

        def run_reactor():
            self._ui.notify_bootstrap(
                notifications.UnmessageNotification('Running reactor'))

            # TODO improve the way the reactor is run
            self._twisted_reactor.run(installSignalHandlers=0)
        thread.start_new_thread(run_reactor, ())

        return d

def make_endpoint(self):
        return TCP4ServerEndpoint(reactor, 0, interface="localhost")

def setup_server():
    config = yaml.load(open('./config/config.yml', 'r'))
    server_port = 8080
    if 'rpc' in config:
        server_port = config['rpc']['server_port']

    site = server.Site(RPCInterface())
    endpoint = endpoints.TCP4ServerEndpoint(reactor, server_port)
    endpoint.listen(site)

def startService(self):
        if HAS_WEB_UI:
            webdir = os.path.abspath(
                pkg_resources.resource_filename('leap.bitmask_js', 'public'))
            log.debug('webdir: %s' % webdir)
        else:
            log.warn('bitmask_js not found, serving bitmask.core ui')
            webdir = os.path.abspath(
                pkg_resources.resource_filename(
                    'leap.bitmask.core.web', 'static'))
            jspath = os.path.join(
                here(), '..', '..', '..',
                'ui', 'app', 'lib', 'bitmask.js')
            jsapi = File(os.path.abspath(jspath))

        api = Api(CommandDispatcher(self._core), self._core.global_tokens)

        root = File(webdir)
        root.putChild(u'API', api)

        # XXX remove it we don't bring session tokens again
        # protected_api = protectedResourceFactory(
        # api, self._core.global_tokens, self.API_WHITELIST)
        # root.putChild(u'API', protected_api)

        if not HAS_WEB_UI:
            root.putChild('bitmask.js', jsapi)

        factory = Site(root)
        self.site = factory

        if self.onion and _has_txtorcon():
            self._start_onion_service(factory)
        else:
            interface = '127.0.0.1'
            endpoint = endpoints.TCP4ServerEndpoint(
                reactor, self.port, interface=interface)
            self.uri = 'https://%s:%s' % (interface, self.port)
            endpoint.listen(factory)

        # TODO this should be set in a callback to the listen call
        self.running = True