Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def testUnixSocketBinding(self):
self.assertEqual(
build(unix_socket="/tmp/daphne.sock"), ["unix:/tmp/daphne.sock"]
)
def testBasics(self):
self.assertEqual(build(), [], msg="Empty list returned when no kwargs given")
def testFileDescriptorBinding(self):
self.assertEqual(build(file_descriptor=5), ["fd:fileno=5"])
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",
]
"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
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,
# )
# 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
"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 "",