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

def makeService(self, options):
        """Construct a server using MLLPFactory.

        :rtype: :py:class:`twisted.application.internet.StreamServerEndpointService`
        """
        from twisted.internet import reactor
        from txHL7.mllp import IHL7Receiver, MLLPFactory

        receiver_name = options['receiver']
        receiver_class = reflect.namedClass(receiver_name)
        verifyClass(IHL7Receiver, receiver_class)
        factory = MLLPFactory(receiver_class())
        multi_service = MultiService()

        for port_number in PORTS:
            port = "tcp:interface={0}:port={1}".format(HOST, port_number,)
            endpoint = endpoints.serverFromString(reactor, port)
            server = internet.StreamServerEndpointService(endpoint, factory)
            server.setName(u"mllp-{0}-{1}".format(receiver_name, port_number))
            multi_service.addService(server)
        return multi_service

def do_start(self):
        """
        make this transport begin listening on the specified interface and UDP port
        interface must be an IP address
        """
        # save a TorConfig so we can later use it to send messages
        self.torconfig = txtorcon.TorConfig(control=self.tor.protocol)
        yield self.torconfig.post_bootstrap

        hs_strings = []
        if len(self.onion_unix_socket) == 0:
            local_socket_endpoint_desc = "tcp:interface=%s:%s" % (self.onion_tcp_interface_ip, self.onion_tcp_port)
        else:
            local_socket_endpoint_desc = "unix:%s" % self.onion_unix_socket
        onion_service_endpoint = endpoints.serverFromString(self.reactor, local_socket_endpoint_desc)
        datagram_proxy_factory = OnionDatagramProxyFactory(received_handler=lambda x: self.datagram_received(x))
        yield onion_service_endpoint.listen(datagram_proxy_factory)
        if len(self.onion_unix_socket) == 0:
            hs_strings.append("%s %s:%s" % (self.onion_port, self.onion_tcp_interface_ip, self.onion_tcp_port))
        else:
            hs_strings.append("%s unix:%s" % (self.onion_port, self.onion_unix_socket))
        hs = txtorcon.torconfig.EphemeralHiddenService(hs_strings, key_blob_or_type=self.onion_key)
        yield hs.add_to_tor(self.tor.protocol)

def _start_onion_service(self, factory):

        def progress(percent, tag, message):
            bar = int(percent / 10)
            log.debug('[%s%s] %s' % ('#' * bar, '.' * (10 - bar), message))

        def setup_complete(port):
            port = txtorcon.IHiddenService(port)
            self.uri = "http://%s" % (port.getHost().onion_uri)
            log.info('I have set up a hidden service, advertised at: %s'
                     % self.uri)
            log.info('locally listening on %s' % port.local_address.getHost())

        def setup_failed(args):
            log.error('onion service setup FAILED: %r' % args)

        endpoint = endpoints.serverFromString(reactor, 'onion:80')
        txtorcon.IProgressProvider(endpoint).add_progress_listener(progress)
        d = endpoint.listen(factory)
        d.addCallback(setup_complete)
        d.addErrback(setup_failed)
        return d

def startService(self):
        """Construct server and bind."""
        from twisted.internet import reactor
        service.Service.startService(self)

        def connected(port):
            self._port = port

        # noinspection PyUnresolvedReferences
        def failure(err):
            log.err(err, _why='Could not bind to port')
            reactor.stop()

        factory = self._factory_klass()
        server = endpoints.serverFromString(reactor, self._endpoint)

        d = server.listen(factory)
        d.addCallbacks(connected, failure)
        return d

def main():
    # Setup the blockchain
    blockchain = LevelDBBlockchain(settings.LEVELDB_PATH)
    Blockchain.RegisterBlockchain(blockchain)
    dbloop = task.LoopingCall(Blockchain.Default().PersistBlocks)
    dbloop.start(.1)
    NodeLeader.Instance().Start()

    # Disable smart contract events for external smart contracts
    settings.set_log_smart_contract_events(False)

    # Start a thread with custom code
    d = threading.Thread(target=custom_background_code)
    d.setDaemon(True)  # daemonizing the thread will kill it when the main thread is quit
    d.start()

    # Hook up Klein API to Twisted reactor
    endpoint_description = "tcp:port=%s:interface=localhost" % API_PORT
    endpoint = endpoints.serverFromString(reactor, endpoint_description)
    endpoint.listen(Site(app.resource()))

    # Run all the things (blocking call)
    logger.info("Everything setup and running. Waiting for events...")
    reactor.run()
    logger.info("Shutting down.")

def makeService(self, options):
        """
        Called by Twisted after having parsed the command-line options.
        :param options: ``usage.Options`` instance
        :return: the server instance
        """
        # Configuration is mandatory
        if options['config'] is None:
            print 'You need to specify a configuration file via `twistd ldap-proxy -c config.ini`.'
            sys.exit(1)

        config = load_config(options['config'])
        factory = ProxyServerFactory(config)

        endpoint_string = serverFromString(reactor, config['ldap-proxy']['endpoint'])
        return internet.StreamServerEndpointService(endpoint_string, factory)

def parseStreamServer(self, reactor, description, **options):

        # The present endpoint plugin is intended to be used as in the
        # following for running a streaming protocol over WebSocket over
        # an underlying stream transport.
        #
        # endpoint = serverFromString(reactor,
        # "autobahn:tcp\:9000\:interface\=0.0.0.0:url=ws\://localhost\:9000:compress=false"
        #
        # This will result in `parseStreamServer` to be called will
        #
        # description == tcp:9000:interface=0.0.0.0
        #
        # and
        #
        # options == {'url': 'ws://localhost:9000', 'compress': 'false'}
        #
        # Essentially, we are using the `\:` escape to coerce the endpoint descriptor
        # of the underlying stream transport into one (first) positional argument.
        #
        # Note that the `\:` within "url" is another form of escaping!
        #
        opts = _parseOptions(options)
        endpoint = serverFromString(reactor, description)
        return AutobahnServerEndpoint(reactor, endpoint, opts)

def run(self, *args, **kwargs):
    """Start all the servers in one place."""
    endpoint_description = 'tcp:port={}:interface={}'.format(
        self.options.web_port,
        self.options.web_interface
    )
    endpoint = endpoints.serverFromString(reactor, endpoint_description)
    endpoint.listen(
        Site(
            klein.resource()
        )
    )
    if self.options.proxy:
        self.factory = ProxyFactory()
        self.factory.server = self
    self.interface = self.options.interface
    self.port = self.options.port
    return super(GameServer, self).run(*args, **kwargs)

def service(description, factory, reactor=None):
    """
    Return the service corresponding to a description.

    @param description: The description of the listening port, in the syntax
        described by L{twisted.internet.endpoints.serverFromString}.
    @type description: C{str}

    @param factory: The protocol factory which will build protocols for
        connections to this service.
    @type factory: L{twisted.internet.interfaces.IProtocolFactory}

    @rtype: C{twisted.application.service.IService}
    @return: the service corresponding to a description of a reliable stream
        server.

    @see: L{twisted.internet.endpoints.serverFromString}
    """
    if reactor is None:
        from twisted.internet import reactor

    svc = StreamServerEndpointService(
        endpoints.serverFromString(reactor, description), factory)
    svc._raiseSynchronously = True
    return svc

def listen(description, factory):
    """
    Listen on a port corresponding to a description.

    @param description: The description of the connecting port, in the syntax
        described by L{twisted.internet.endpoints.serverFromString}.
    @type description: L{str}

    @param factory: The protocol factory which will build protocols on
        connection.
    @type factory: L{twisted.internet.interfaces.IProtocolFactory}

    @rtype: L{twisted.internet.interfaces.IListeningPort}
    @return: the port corresponding to a description of a reliable virtual
        circuit server.

    @see: L{twisted.internet.endpoints.serverFromString}
    """
    from twisted.internet import reactor
    name, args, kw = endpoints._parseServer(description, factory)
    return getattr(reactor, 'listen' + name)(*args, **kw)

def test_ssl(self):
        """
        When passed an SSL strports description, L{endpoints.serverFromString}
        returns a L{SSL4ServerEndpoint} instance initialized with the values
        from the string.
        """
        reactor = object()
        server = endpoints.serverFromString(
            reactor,
            "ssl:1234:backlog=12:privateKey=%s:"
            "certKey=%s:sslmethod=TLSv1_METHOD:interface=10.0.0.1"
            % (escapedPEMPathName, escapedPEMPathName))
        self.assertIsInstance(server, endpoints.SSL4ServerEndpoint)
        self.assertIs(server._reactor, reactor)
        self.assertEqual(server._port, 1234)
        self.assertEqual(server._backlog, 12)
        self.assertEqual(server._interface, "10.0.0.1")
        self.assertEqual(server._sslContextFactory.method, TLSv1_METHOD)
        ctx = server._sslContextFactory.getContext()
        self.assertIsInstance(ctx, ContextType)

def test_sslChainLoads(self):
        """
        Specifying a chain file loads the contained certificates in the right
        order.
        """
        server = endpoints.serverFromString(
            object(),
            self.SSL_CHAIN_TEMPLATE % (escapedPEMPathName,
                                       escapedChainPathName,)
        )
        # Test chain file is just a concatenation of thing1.pem and thing2.pem
        # so we can check that loading has succeeded and order has been
        # preserved.
        expectedChainCerts = [
            Certificate.loadPEM(casPath.child("thing%d.pem" % (n,))
                                .getContent())
            for n in [1, 2]
        ]
        cf = server._sslContextFactory
        self.assertEqual(cf.extraCertChain[0].digest('sha1'),
                         expectedChainCerts[0].digest('sha1'))
        self.assertEqual(cf.extraCertChain[1].digest('sha1'),
                         expectedChainCerts[1].digest('sha1'))

def test_sslChainFileMustContainCert(self):
        """
        If C{extraCertChain} is passed, it has to contain at least one valid
        certificate in PEM format.
        """
        fp = FilePath(self.mktemp())
        fp.create().close()
        # The endpoint string is the same as in the valid case except for
        # a different chain file.  We use an empty temp file which obviously
        # will never contain any certificates.
        with self.assertRaises(ValueError) as caught:
            endpoints.serverFromString(
                object(),
                self.SSL_CHAIN_TEMPLATE % (
                    escapedPEMPathName,
                    endpoints.quoteStringArgument(fp.path),
                )
            )

        # The raised exception should list what file it is attempting to find
        # the chain in.
        self.assertEqual(str(caught.exception),
                         ("Specified chain file '%s' doesn't contain any valid"
                          " certificates in PEM format.") % (fp.path,))

def test_unix(self):
        """
        When passed a UNIX strports description, L{endpoint.serverFromString}
        returns a L{UNIXServerEndpoint} instance initialized with the values
        from the string.
        """
        reactor = object()
        endpoint = endpoints.serverFromString(
            reactor,
            "unix:/var/foo/bar:backlog=7:mode=0123:lockfile=1")
        self.assertIsInstance(endpoint, endpoints.UNIXServerEndpoint)
        self.assertIs(endpoint._reactor, reactor)
        self.assertEqual(endpoint._address, "/var/foo/bar")
        self.assertEqual(endpoint._backlog, 7)
        self.assertEqual(endpoint._mode, 0o123)
        self.assertTrue(endpoint._wantPID)

def _parse(reactor, directory, pemdir, *args, **kwargs):
    """
    Parse a txacme endpoint description.

    :param reactor: The Twisted reactor.
    :param directory: ``twisted.python.url.URL`` for the ACME directory to use
        for issuing certs.
    :param str pemdir: The path to the certificate directory to use.
    """
    def colon_join(items):
        return ':'.join([item.replace(':', '\\:') for item in items])
    sub = colon_join(list(args) + ['='.join(item) for item in kwargs.items()])
    pem_path = FilePath(pemdir).asTextMode()
    acme_key = load_or_create_client_key(pem_path)
    return AutoTLSEndpoint(
        reactor=reactor,
        directory=directory,
        client_creator=partial(Client.from_url, key=acme_key, alg=RS256),
        cert_store=DirectoryStore(pem_path),
        cert_mapping=HostDirectoryMap(pem_path),
        sub_endpoint=serverFromString(reactor, sub))

def listen(self, reactor, endpoint_description):
        """
        Run the server, i.e. start listening for requests on the given host and
        port.

        :param reactor: The ``IReactorTCP`` to use.
        :param endpoint_description:
            The Twisted description for the endpoint to listen on.
        :return:
            A deferred that returns an object that provides ``IListeningPort``.
        """
        endpoint = serverFromString(reactor, endpoint_description)
        return endpoint.listen(Site(self.app.resource()))

def test_onion_datagram_proxy():
    received_buffer = []
    received_d = defer.Deferred()

    def received(data):
        received_buffer.append(data)
        received_d.callback(None)

    received_size = 10
    proxy_factory = OnionDatagramProxyFactory(received)
    protocol = proxy_factory.buildProtocol(123)
    packet = b"A" * received_size
    protocol.stringReceived(packet)
    assert received_buffer[0] == packet

    service_port = yield txtorcon.util.available_tcp_port(reactor)
    service_endpoint_desc = "tcp:interface=127.0.0.1:%s" % service_port
    service_endpoint = endpoints.serverFromString(reactor, service_endpoint_desc)
    yield service_endpoint.listen(proxy_factory)

    client_endpoint_desc = "tcp:127.0.0.1:%s" % service_port
    client_endpoint = endpoints.clientFromString(reactor, client_endpoint_desc)
    client_protocol = Int32StringReceiver()
    yield endpoints.connectProtocol(client_endpoint, client_protocol)
    client_protocol.sendString(packet)
    print "BEFORE CLOSE"
    client_protocol.transport.loseConnection()
    yield received_d
    assert received_buffer[0] == packet

def setupservice(self):
        # TODO: Remove this line when Python Klein pull request #103 is
        # released
        # NOTE: Docker 1.9 will fail without this line. Docker 1.10 will
        # fail as it no longer includes the Host as part of the http header.
        # Therefore, we need to remove this line altogether.
        # 4/6/16 Removing this line as it's causing problems for testers on
        # Docker 1.10. If you're running 1.9, you can apply the Klein fix
        # here https://github.com/twisted/klein.git to fix.
        UNIXAddress.port = 0
        UNIXAddress.host = b"127.0.0.1"

        # Turnoff use of parameterized hpe.conf and use bind mounted
        # configuration file
        # CONFIG = ['--config-file', self._config_file]
        CONFIG = ['--config-file', CONFIG_FILE]

        # Setup the default, hpe3parconfig, and hpelefthandconfig
        # configuration objects.
        try:
            hpedefaultconfig = getdefaultconfig(CONFIG)
        except Exception as ex:
            msg = (_('hpe3pardocker setupservice failed, error is: %s'),
                   six.text_type(ex))
            LOG.error(msg)
            raise exception.HPEPluginStartPluginException(reason=msg)

        # Set Logging level
        logging_level = hpedefaultconfig.logging
        setup_logging('hpe_storage_api', logging_level)

        self._create_listening_directory(PLUGIN_PATH.parent())
        endpoint = serverFromString(self._reactor, "unix:{}:mode=600".
                                    format(PLUGIN_PATH.path))
        servicename = StreamServerEndpointService(endpoint, Site(
            VolumePlugin(self._reactor, hpedefaultconfig).app.resource()))
        return servicename

def start_tor(site, hs_public_port, hsdir):
    # set up HS server
    hs_endpoint = endpoints.serverFromString(reactor,
        "onion:"+str(hs_public_port)+":hiddenServiceDir="+hsdir)
    d = hs_endpoint.listen(site)
    #add chain of callbacks for actions after Tor is set up correctly.
    d.addCallback(listening)
    d.addErrback(setup_failed)
    return d

def startService(self):
        factory = Factory.forProtocol(EndpointForwardingProtocol)
        factory.service = self
        self._endpoint = serverFromString(self._reactor, self._endpointDescriptor)
        self._endpointPort = yield self._endpoint.listen(factory)

def test_quoteStringArgument(self):
        """
        L{endpoints.quoteStringArgument} should quote backslashes and colons
        for interpolation into L{endpoints.serverFromString} and
        L{endpoints.clientFactory} arguments.
        """
        self.assertEqual(endpoints.quoteStringArgument("some : stuff \x5c"),
                         "some \x5c: stuff \x5c\x5c")

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 test_sslDHparameters(self):
        """
        If C{dhParameters} are specified, they are passed as
        L{DiffieHellmanParameters} into L{CertificateOptions}.
        """
        fileName = 'someFile'
        reactor = object()
        server = endpoints.serverFromString(
            reactor,
            "ssl:4321:privateKey={0}:certKey={1}:dhParameters={2}"
            .format(escapedPEMPathName, escapedPEMPathName, fileName)
        )
        cf = server._sslContextFactory
        self.assertIsInstance(cf.dhParameters, DiffieHellmanParameters)
        self.assertEqual(FilePath(fileName), cf.dhParameters._dhFile)

def test_unknownType(self):
        """
        L{endpoints.serverFromString} raises C{ValueError} when given an
        unknown endpoint type.
        """
        value = self.assertRaises(
            # faster-than-light communication not supported
            ValueError, endpoints.serverFromString, None,
            "ftl:andromeda/carcosa/hali/2387")
        self.assertEqual(
            str(value),
            "Unknown endpoint type: 'ftl'")

def test_stringDescription(self):
        """
        L{serverFromString} returns a L{TCP6ServerEndpoint} instance with a
        'tcp6' endpoint string description.
        """
        ep = endpoints.serverFromString(
            MemoryReactor(), "tcp6:8080:backlog=12:interface=\:\:1")
        self.assertIsInstance(ep, endpoints.TCP6ServerEndpoint)
        self.assertIsInstance(ep._reactor, MemoryReactor)
        self.assertEqual(ep._port, 8080)
        self.assertEqual(ep._backlog, 12)
        self.assertEqual(ep._interface, '::1')

def test_stringDescription(self):
        """
        L{serverFromString} returns a L{StandardIOEndpoint} instance with a
        'stdio' endpoint string description.
        """
        ep = endpoints.serverFromString(MemoryReactor(), "stdio:")
        self.assertIsInstance(ep, endpoints.StandardIOEndpoint)
        self.assertIsInstance(ep._reactor, MemoryReactor)

def addEndpoint(self, service, description):
        """
        Add an endpoint to a service.

        @type service: L{bytes}
        @param service: A service, either C{b'smtp'} or C{b'pop3'}.

        @type description: L{bytes}
        @param description: An endpoint description string or a TCP port
            number.
        """
        from twisted.internet import reactor
        self[service].append(endpoints.serverFromString(reactor, description))

def run(self, host=None, port=None, logFile=None,
            endpoint_description=None):
        """
        Run a minimal twisted.web server on the specified C{port}, bound to the
        interface specified by C{host} and logging to C{logFile}.

        This function will run the default reactor for your platform and so
        will block the main thread of your application.  It should be the last
        thing your klein application does.

        @param host: The hostname or IP address to bind the listening socket
            to.  "0.0.0.0" will allow you to listen on all interfaces, and
            "127.0.0.1" will allow you to listen on just the loopback interface.
        @type host: str

        @param port: The TCP port to accept HTTP requests on.
        @type port: int

        @param logFile: The file object to log to, by default C{sys.stdout}
        @type logFile: file object

        @param endpoint_description: specification of endpoint. Must contain
            protocol, port and interface. May contain other optional arguments,
             e.g. to use SSL: "ssl:443:privateKey=key.pem:certKey=crt.pem"
        @type endpoint_description: str
        """
        if logFile is None:
            logFile = sys.stdout

        log.startLogging(logFile)

        if not endpoint_description:
            endpoint_description = "tcp:port={0}:interface={1}".format(port,
                                                                       host)

        endpoint = endpoints.serverFromString(reactor, endpoint_description)
        endpoint.listen(Site(self.resource()))
        reactor.run()

def startService(self):
        self.endpoint = yield endpoints.serverFromString(reactor, self.description)
        self.endpoint.listen(server.Site(self.resource))

def runWithoutLog():
    site = Site(testServer, logPath=b'/tmp/log')
    endpoints.serverFromString(reactor, 'tcp:8081').listen(site)
    reactor.run()

def _create_responder(self):
        action = start_action(action_type=u'integration:create_responder')
        with action.context():
            responder = TLSSNI01Responder()
            host_map = responder.wrap_host_map({})
            site = Site(Resource())
            endpoint = TLSEndpoint(
                endpoint=serverFromString(reactor, self.ENDPOINT),
                contextFactory=SNIMap(host_map))
            return (
                DeferredContext(endpoint.listen(site))
                .addCallback(lambda port: self.addCleanup(port.stopListening))
                .addCallback(lambda _: responder)
                .addActionFinish())

def create_ldap_server():
    "Returns a context manager that represents the LDAP server."

    db = yield _create_db()
    factory = _LDAPServerFactory(db)
    factory.debug = True

    # We just pick an arbitrary port to listen on.
    serverEndpointStr = "tcp:0"
    e = serverFromString(reactor, serverEndpointStr)
    listener = yield e.listen(factory)

    defer.returnValue(_LdapServer(listener))