Merge pull request #1097 from pypeclub/feature/ws_tool_in_avalon

Websocket server tool in avalon

pype/lib/applications.py

@@ -101,6 +101,9 @@ def __init__(self):
 
  def refresh(self):
  """Refresh applications from settings."""
+ self.applications.clear()
+ self.tools.clear()
+
  settings = get_system_settings()
 
  hosts_definitions = settings["applications"]

pype/modules/__init__.py

@@ -40,7 +40,7 @@
 from .muster import MusterModule
 from .deadline import DeadlineModule
 from .standalonepublish_action import StandAlonePublishAction
-from .websocket_server import WebsocketModule
+from .webserver import WebServerModule
 from .sync_server import SyncServer
 
 
@@ -82,6 +82,6 @@
  "DeadlineModule",
  "StandAlonePublishAction",
 
- "WebsocketModule",
+ "WebServerModule",
  "SyncServer"
 )

pype/modules/rest_api/rest_api.py

@@ -107,7 +107,6 @@ def initialize(self, modules_settings):
 
  self.rest_api_url = None
  self.rest_api_thread = None
- self.resources_url = None
 
  def register_callback(
  self, path, callback, url_prefix="", methods=[], strict_match=False
@@ -189,11 +188,9 @@ def tray_init(self):
  self.rest_api_url = "http://localhost:{}".format(port)
  self.rest_api_thread = RestApiThread(self, port)
  self.register_statics("/res", resources.RESOURCES_DIR)
- self.resources_url = "{}/res".format(self.rest_api_url)
 
  # Set rest api environments
  os.environ["PYPE_REST_API_URL"] = self.rest_api_url
- os.environ["PYPE_STATICS_SERVER"] = self.resources_url
 
  def tray_start(self):
  RestApiFactory.prepare_registered()

pype/modules/webserver/__init__.py

@@ -0,0 +1,6 @@
+from .webserver_module import WebServerModule
+
+
+__all__ = (
+ "WebServerModule",
+)

pype/modules/webserver/server.py

@@ -0,0 +1,142 @@
+import threading
+import asyncio
+
+from aiohttp import web
+
+from pype.lib import PypeLogger
+
+log = PypeLogger.get_logger("WebServer")
+
+
+class WebServerManager:
+ """Manger that care about web server thread."""
+ def __init__(self, module):
+ self.module = module
+
+ self.client = None
+ self.handlers = {}
+ self.on_stop_callbacks = []
+
+ self.app = web.Application()
+
+ # add route with multiple methods for single "external app"
+
+ self.webserver_thread = WebServerThread(self, self.module.port)
+
+ def add_route(self, *args, **kwargs):
+ self.app.router.add_route(*args, **kwargs)
+
+ def add_static(self, *args, **kwargs):
+ self.app.router.add_static(*args, **kwargs)
+
+ def start_server(self):
+ if self.webserver_thread and not self.webserver_thread.is_alive():
+ self.webserver_thread.start()
+
+ def stop_server(self):
+ if not self.is_running:
+ return
+ try:
+ log.debug("Stopping Web server")
+ self.webserver_thread.is_running = False
+ self.webserver_thread.stop()
+
+ except Exception:
+ log.warning(
+ "Error has happened during Killing Web server",
+ exc_info=True
+ )
+
+ @property
+ def is_running(self):
+ if not self.webserver_thread:
+ return False
+ return self.webserver_thread.is_running
+
+ def thread_stopped(self):
+ for callback in self.on_stop_callbacks:
+ callback()
+
+
+class WebServerThread(threading.Thread):
+ """ Listener for requests in thread."""
+ def __init__(self, manager, port):
+ super(WebServerThread, self).__init__()
+
+ self.is_running = False
+ self.port = port
+ self.manager = manager
+ self.loop = None
+ self.runner = None
+ self.site = None
+ self.tasks = []
+
+ def run(self):
+ self.is_running = True
+
+ try:
+ log.info("Starting WebServer server")
+ self.loop = asyncio.new_event_loop() # create new loop for thread
+ asyncio.set_event_loop(self.loop)
+
+ self.loop.run_until_complete(self.start_server())
+
+ log.debug(
+ "Running Web server on URL: \"localhost:{}\"".format(self.port)
+ )
+
+ asyncio.ensure_future(self.check_shutdown(), loop=self.loop)
+ self.loop.run_forever()
+
+ except Exception:
+ log.warning(
+ "Web Server service has failed", exc_info=True
+ )
+ finally:
+ self.loop.close() # optional
+
+ self.is_running = False
+ self.manager.thread_stopped()
+ log.info("Web server stopped")
+
+ async def start_server(self):
+ """ Starts runner and TCPsite """
+ self.runner = web.AppRunner(self.manager.app)
+ await self.runner.setup()
+ self.site = web.TCPSite(self.runner, 'localhost', self.port)
+ await self.site.start()
+
+ def stop(self):
+ """Sets is_running flag to false, 'check_shutdown' shuts server down"""
+ self.is_running = False
+
+ async def check_shutdown(self):
+ """ Future that is running and checks if server should be running
+ periodically.
+ """
+ while self.is_running:
+ while self.tasks:
+ task = self.tasks.pop(0)
+ log.debug("waiting for task {}".format(task))
+ await task
+ log.debug("returned value {}".format(task.result))
+
+ await asyncio.sleep(0.5)
+
+ log.debug("Starting shutdown")
+ await self.site.stop()
+ log.debug("Site stopped")
+ await self.runner.cleanup()
+ log.debug("Runner stopped")
+ tasks = [
+ task
+ for task in asyncio.all_tasks()
+ if task is not asyncio.current_task()
+ ]
+ list(map(lambda task: task.cancel(), tasks)) # cancel all the tasks
+ results = await asyncio.gather(*tasks, return_exceptions=True)
+ log.debug(f'Finished awaiting cancelled tasks, results: {results}...')
+ await self.loop.shutdown_asyncgens()
+ # to really make sure everything else has time to stop
+ await asyncio.sleep(0.07)
+ self.loop.stop()

pype/modules/webserver/webserver_module.py

@@ -0,0 +1,55 @@
+import os
+from pype import resources
+from .. import PypeModule, ITrayService
+
+
+class WebServerModule(PypeModule, ITrayService):
+ name = "webserver"
+ label = "WebServer"
+
+ def initialize(self, module_settings):
+ self.enabled = True
+ self.server_manager = None
+
+ # TODO find free port
+ self.port = 8098
+
+ def connect_with_modules(self, *_a, **_kw):
+ return
+
+ def tray_init(self):
+ self.create_server_manager()
+ self._add_resources_statics()
+
+ def tray_start(self):
+ self.start_server()
+
+ def tray_exit(self):
+ self.stop_server()
+
+ def _add_resources_statics(self):
+ static_prefix = "/res"
+ self.server_manager.add_static(static_prefix, resources.RESOURCES_DIR)
+
+ os.environ["PYPE_STATICS_SERVER"] = "http://localhost:{}{}".format(
+ self.port, static_prefix
+ )
+
+ def start_server(self):
+ if self.server_manager:
+ self.server_manager.start_server()
+
+ def stop_server(self):
+ if self.server_manager:
+ self.server_manager.stop_server()
+
+ def create_server_manager(self):
+ if self.server_manager:
+ return
+
+ from .server import WebServerManager
+
+ self.server_manager = WebServerManager(self)
+ self.server_manager.on_stop_callbacks.append(
+ self.set_service_failed_icon
+ )