How to use the daphne.endpoints.build_endpoint_description_strings function in daphne

To help you get started, we’ve selected a few daphne 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 django / daphne / tests / test_cli.py View on Github external
def testUnixSocketBinding(self):
        self.assertEqual(
            build(unix_socket="/tmp/daphne.sock"), ["unix:/tmp/daphne.sock"]
        )
github django / daphne / tests / test_cli.py View on Github external
def testBasics(self):
        self.assertEqual(build(), [], msg="Empty list returned when no kwargs given")
github django / daphne / tests / test_cli.py View on Github external
def testFileDescriptorBinding(self):
        self.assertEqual(build(file_descriptor=5), ["fd:fileno=5"])
github django / daphne / tests / test_cli.py View on Github external
def testMultipleEnpoints(self):
        self.assertEqual(
            sorted(
                build(
                    file_descriptor=123,
                    unix_socket="/tmp/daphne.sock",
                    port=8080,
                    host="10.0.0.1",
                )
            ),
            sorted(
                [
                    "tcp:port=8080:interface=10.0.0.1",
                    "unix:/tmp/daphne.sock",
                    "fd:fileno=123",
                ]
github django / channels / channels / management / commands / runserver.py View on Github external
"version": self.get_version(),
                "channels_version": __version__,
                "settings": settings.SETTINGS_MODULE,
                "protocol": self.protocol,
                "addr": "[%s]" % self.addr if self._raw_ipv6 else self.addr,
                "port": self.port,
                "quit_command": quit_command,
            }
        )

        # Launch server in 'main' thread. Signals are disabled as it's still
        # actually a subthread under the autoreloader.
        logger.debug("Daphne running, listening on %s:%s", self.addr, self.port)

        # build the endpoint description string from host/port options
        endpoints = build_endpoint_description_strings(host=self.addr, port=self.port)
        try:
            self.server_cls(
                application=self.get_application(options),
                endpoints=endpoints,
                signal_handlers=not options["use_reloader"],
                action_logger=self.log_action,
                http_timeout=self.http_timeout,
                root_path=getattr(settings, "FORCE_SCRIPT_NAME", "") or "",
                websocket_handshake_timeout=self.websocket_handshake_timeout,
            ).run()
            logger.debug("Daphne exited")
        except KeyboardInterrupt:
            shutdown_message = options.get("shutdown_message", "")
            if shutdown_message:
                self.stdout.write(shutdown_message)
            return
github django / daphne / daphne / cli.py View on Github external
args.host,
                args.port is not None,
                args.unix_socket,
                args.file_descriptor is not None,
                args.socket_strings,
            ]
        ):
            # no advanced binding options passed, patch in defaults
            args.host = DEFAULT_HOST
            args.port = DEFAULT_PORT
        elif args.host and args.port is None:
            args.port = DEFAULT_PORT
        elif args.port is not None and not args.host:
            args.host = DEFAULT_HOST
        # Build endpoint description strings from (optional) cli arguments
        endpoints = build_endpoint_description_strings(
            host=args.host,
            port=args.port,
            unix_socket=args.unix_socket,
            file_descriptor=args.file_descriptor,
        )
        endpoints = sorted(args.socket_strings + endpoints)
        # Start the server
        logger.info("Starting server at %s" % (", ".join(endpoints),))
        self.server = self.server_class(
            application=application,
            endpoints=endpoints,
            http_timeout=args.http_timeout,
            ping_interval=args.ping_interval,
            ping_timeout=args.ping_timeout,
            websocket_timeout=args.websocket_timeout,
            websocket_connect_timeout=args.websocket_connect_timeout,
github Splawik / pytigon / schlib / schdjangoext / server.py View on Github external
#    )
    #    for _ in range(4):
    #        worker = WorkerThread(channel_layer)
    #        worker.daemon = True
    #        worker.start()
    try:
        from daphne.server import Server
        from daphne.endpoints import build_endpoint_description_strings
        from channels.routing import get_default_application

        django.setup()

        #application = django.core.handlers.wsgi.WSGIHandler()

        print(addr, port)
        endpoints = build_endpoint_description_strings(host=addr, port=int(port))

        server = Server(
            #channel_layer=channel_layer,
            get_default_application(),
            endpoints=endpoints,
            #host=addr,
            #port=int(port),
            signal_handlers=False,
            action_logger=log_action,
            http_timeout=60,
        )
        server.run()
    except KeyboardInterrupt:
        return
github oTree-org / otree-core / otree / management / commands / runserver.py View on Github external
"Open your browser to http://%(addr)s:%(port)s/\n"
            "To quit the server, press Control+C.\n"
        ) % {
            "addr": addr,
            "port": self.port,
        })

        # silence the lines like:
        # 2018-01-10 18:51:18,092 - INFO - worker - Listening on channels
        # http.request, otree.create_session, websocket.connect,
        # websocket.disconnect, websocket.receive
        daphne_logger = logging.getLogger('django.channels')
        original_log_level = daphne_logger.level
        daphne_logger.level = logging.WARNING

        endpoints = build_endpoint_description_strings(host=self.addr, port=self.port)
        application = self.get_application(options)

        # silence the lines like:
        # INFO HTTP/2 support not enabled (install the http2 and tls Twisted extras)
        # INFO Configuring endpoint tcp:port=8000:interface=127.0.0.1
        # INFO Listening on TCP address 127.0.0.1:8000
        logging.getLogger('daphne.server').level = logging.WARNING

        try:
            self.server_cls(
                application=application,
                endpoints=endpoints,
                signal_handlers=not options["use_reloader"],
                action_logger=self.log_action,
                http_timeout=self.http_timeout,
                root_path=getattr(settings, "FORCE_SCRIPT_NAME", "") or "",