core upgrade

- Core upgraded
- max 2 concurrent messages at a time

Author: susrisha

requirements.txt

@@ -1,6 +1,6 @@
 fastapi==0.88.0
 pydantic==1.10.4
-python-ms-core==0.0.21
+python-ms-core==0.0.22
 uvicorn==0.20.0
 html_testRunner==1.2.1
 python-osw-validation==0.2.4

src/main.py

@@ -9,6 +9,8 @@
 
 prefix_router = APIRouter(prefix='/health')
 
+# Have a reference to validator in the app object
+app.validator = None
 
 @lru_cache()
 def get_settings():
@@ -18,7 +20,8 @@ def get_settings():
 @app.on_event('startup')
 async def startup_event(settings: Settings = Depends(get_settings)) -> None:
     try:
-        OSWValidator()
+        # OSWValidator()
+        app.validator = OSWValidator()
     except:
         print('\n\n\x1b[31m Application startup failed due to missing or invalid .env file \x1b[0m')
         print('\x1b[31m Please provide the valid .env file and .env file should contains following parameters\x1b[0m')
@@ -34,6 +37,11 @@ async def startup_event(settings: Settings = Depends(get_settings)) -> None:
             child.kill()
         parent.kill()
 
+@app.on_event('shutdown')
+async def shutdown_event() -> None:
+    print('Shutting down the application')
+    if app.validator:
+        app.validator.stop_listening()
 
 @app.get('/', status_code=status.HTTP_200_OK)
 @prefix_router.get('/', status_code=status.HTTP_200_OK)

src/osw_validator.py

@@ -31,7 +31,8 @@ def __init__(self):
         self.logger = self.core.get_logger()
         self.storage_client = self.core.get_storage_client()
         self.auth = self.core.get_authorizer(config=options)
-        self.start_listening()
+        self.listener_thread = threading.Thread(target=self.start_listening)
+        self.listener_thread.start()
 
     def start_listening(self):
         def process(message) -> None:
@@ -103,3 +104,6 @@ def has_permission(self, roles: List[str], queue_message: Upload) -> bool:
         except Exception as error:
             print('Error validating the request authorization:', error)
             return False
+
+    def stop_listening(self):
+        self.listener_thread.join(timeout=0) # Stop the thread during shutdown.Its still an attempt. Not sure if this will work.