twisted/application/strports.py

# -*- 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}
"""

import warnings

from twisted.internet import endpoints
from twisted.python.deprecate import deprecatedModuleAttribute
from twisted.python.versions import Version
from twisted.application.internet import StreamServerEndpointService


def parse(description, factory, default='tcp'):
    """
    This function is deprecated as of Twisted 10.2.

    @see: L{twisted.internet.endpoints.server}
    """
    return endpoints._parseServer(description, factory, default)

deprecatedModuleAttribute(
    Version("Twisted", 10, 2, 0),
    "in favor of twisted.internet.endpoints.serverFromString",
    __name__, "parse")


_DEFAULT = object()

def service(description, factory, default=_DEFAULT, 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.server}.

    @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}

    @type default: C{str} or C{None}

    @param default: Do not use this parameter. It has been deprecated since
        Twisted 10.2.0.

    @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
    if default is _DEFAULT:
        default = None
    else:
        message = "The 'default' parameter was deprecated in Twisted 10.2.0."
        if default is not None:
            message += (
                "  Use qualified endpoint descriptions; for example, "
                "'tcp:%s'." % (description,))
        warnings.warn(
            message=message, category=DeprecationWarning, stacklevel=2)
    svc = StreamServerEndpointService(
        endpoints._serverFromStringLegacy(reactor, description, default),
        factory)
    svc._raiseSynchronously = True
    return svc


def listen(description, factory, default=None):
    """Listen on a port corresponding to a description

    @type description: C{str}
    @type factory: L{twisted.internet.interfaces.IProtocolFactory}
    @type default: C{str} or C{None}
    @rtype: C{twisted.internet.interfaces.IListeningPort}
    @return: the port corresponding to a description of a reliable
    virtual circuit server.

    See the documentation of the C{parse} function for description
    of the semantics of the arguments.
    """
    from twisted.internet import reactor
    name, args, kw = parse(description, factory, default)
    return getattr(reactor, 'listen'+name)(*args, **kw)


__all__ = ['parse', 'service', 'listen']