| Server IP : 212.252.79.165 / Your IP : 216.73.216.45 [ Web Server : Apache System : Linux 212-252-79-165.cprapid.com 5.15.0-153-generic #163-Ubuntu SMP Thu Aug 7 16:37:18 UTC 2025 x86_64 User : cehaburo ( 1001) PHP Version : 8.1.33 Disable Function : exec,passthru,shell_exec,system Domains : 48 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/application/ |
Upload File : |
# -*- test-case-name: twisted.test.test_strports -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Construct listening port services from a simple string description.
@see: L{twisted.internet.endpoints.serverFromString}
@see: L{twisted.internet.endpoints.clientFromString}
"""
from typing import Optional, cast
from twisted.application.internet import StreamServerEndpointService
from twisted.internet import endpoints, interfaces
def _getReactor() -> interfaces.IReactorCore:
from twisted.internet import reactor
return cast(interfaces.IReactorCore, reactor)
def service(
description: str,
factory: interfaces.IProtocolFactory,
reactor: Optional[interfaces.IReactorCore] = None,
) -> StreamServerEndpointService:
"""
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:
reactor = _getReactor()
svc = StreamServerEndpointService(
endpoints.serverFromString(reactor, description), factory
)
svc._raiseSynchronously = True
return svc
def listen(
description: str, factory: interfaces.IProtocolFactory
) -> interfaces.IListeningPort:
"""
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 cast(
interfaces.IListeningPort, getattr(reactor, "listen" + name)(*args, **kw)
)
__all__ = ["service", "listen"]