Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_cherrypy_server(host="127.0.0.1", port=9008):
"""
Runs a CherryPy server on Python 2.x.
"""
import cherrypy
from ws4py.server.cherrypyserver import WebSocketPlugin, WebSocketTool
from ws4py.websocket import EchoWebSocket
cherrypy.config.update({'server.socket_host': host,
'server.socket_port': port,
'engine.autoreload.on': False,
'log.screen': False})
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
class Root(object):
@cherrypy.expose
def index(self):
pass
config = {
'/': {
'tools.encode.on': False,
'tools.websocket.on': True,
'tools.websocket.handler_cls': EchoWebSocket
}
}
logger = logging.getLogger('autobahn_testsuite')
logger.warning("Serving CherryPy server on %s:%s" % (host, port))
print(a, b)
def tast(self, _client, a=5, b=1, c=3):
"""
@type _client: CommHandler
"""
time.sleep(a)
_client.onTest(5,6)
#_client.otherClients.onTest(5,6) #todo: not implemented respond on client
#_client.allClients.onTest(6,7)
return {1:1,2:2}
cherrypy.config.update({'server.socket_host': '0.0.0.0',
'server.socket_port': 8888,
'engine.autoreload_on': False})
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
config = {
'/': {
'tools.websocket.on': True,
'tools.websocket.handler_cls': ClientHandler
}
}
cherrypy.quickstart(Root(), '/', config)
'/': {
'tools.staticdir.on': True,
'tools.staticdir.dir': os.path.join(
os.path.dirname(os.path.abspath(__file__)), "static")
}
}
if args.auth is not None:
conf['/'].update({
'tools.auth_basic.on': True,
'tools.auth_basic.realm': 'pysenteishon',
'tools.auth_basic.checkpassword': validate_password,
})
# Websocket server support
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
conf.update({
'/slides': {
'tools.websocket.on': True,
'tools.websocket.handler_cls': WebSocketHandler
}
})
cherrypy.quickstart(
PySenteishon(plugins={
plugins.ON_TAP_KEY: [plugins.Screenshot],
}), '/', conf
)
if args.version:
print(VERSION)
def start():
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
cherrypy.tree.mount(
WS_API(),
'/ws',
config={
'/': {
'tools.websocket.on': True,
'tools.websocket.handler_cls': NWebSocket,
'tools.sessions.on': True,
'tools.sessions.timeout': session_timeout
}
def StartAgent(trainer=None,port=8765):
global workout
workout = trainer
cherrypy.config.update({'server.socket_port': port})
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.tools.websocket = WebSocketTool()
cherrypy.config.update({'engine.autoreload.on': False})
config = {'/ws': {'tools.websocket.on': True,
'tools.websocket.handler_cls': AgentServer}}
cherrypy.quickstart(Root(), '/', config)
# encoding: utf-8
from __future__ import absolute_import, print_function, unicode_literals
try:
import cherrypy
from ws4py.server.cherrypyserver import WebSocketPlugin, WebSocketTool
from ws4py.websocket import WebSocket
except ImportError:
raise
cherrypy.tools.websocket = WebSocketTool()
WebSocketPlugin(cherrypy.engine).subscribe()
class SynchronizingWebsocket(WebSocket):
'''
Class to handle requests sent to this websocket connection.
Each instance of this class represents a Salt websocket connection.
Waits to receive a ``ready`` message from the client.
Calls send on it's end of the pipe to signal to the sender on receipt
of ``ready``.
This class also kicks off initial information probing jobs when clients
initially connect. These jobs help gather information about minions, jobs,
and documentation.
'''
def __init__(self, *args, **kwargs): # pylint: disable=E1002
super(SynchronizingWebsocket, self).__init__(*args, **kwargs)
else:
logging.basicConfig(level=args.loglevel[0])
if args.quiet_cp:
cherrypy.log.error_file = ""
cherrypy.log.access_file = ""
logging.getLogger("cherrypy.error").setLevel(logging.CRITICAL)
if not args.quiet:
print "----"
print "CSS-TS endpoint URL: ws://%s:%d/ts" % (args.ts_addr, args.ts_port)
print "CSS-WC endpoint bound to host %s on port %d" % (args.wc_addr, args.wc_port)
print "Content ID:",CONTENT_ID
print "----"
WebSocketPlugin(cherrypy.engine).subscribe()
cherrypy.config.update({"server.socket_host":args.ts_addr})
cherrypy.config.update({"server.socket_port":args.ts_port})
cherrypy.config.update({"engine.autoreload.on":False})
wallClock = SysClock(tickRate=1000000000)
tsServer = TSServer(contentId=CONTENT_ID, wallClock=wallClock, maxConnectionsAllowed=3)
class Root(object):
@cherrypy.expose
def ts(self):
pass
cherrypy.tree.mount(Root(), "/", config={"/ts": {'tools.dvb_ts.on': True,
'tools.dvb_ts.handler_cls': tsServer.handler}})
cherrypy.engine.start()