(self)
Worker startup procedure
Worker startup procedure
Worker startup procedure
async def start_worker(self):
        """Worker startup procedure"""
        # Ensure an event loop exists
        get_event_loop()

        self._worker_tasks = set()

        # Start monitoring for errors on the error queue
        error_monitor_task = asyncio.ensure_future(self.error_monitor())
        self._error_monitor_task = error_monitor_task
        self._worker_tasks.add(self._error_monitor_task)

        # Features setup & logging
        if not self.api_registry.all() and Feature.RPCS in self.features:
            logger.info("Disabling serving of RPCs as no APIs have been registered")
            self.features.remove(Feature.RPCS)

        logger.info(
            LBullets(
                f"Enabled features ({len(self.features)})", items=[f.value for f in self.features]
            )
        )

        disabled_features = set(ALL_FEATURES) - set(self.features)
        logger.info(
            LBullets(
                f"Disabled features ({len(disabled_features)})",
                items=[f.value for f in disabled_features],
            )
        )

        # Api logging
        logger.info(
            LBullets(
                "APIs in registry ({})".format(len(self.api_registry.all())),
                items=self.api_registry.names(),
            )
        )

        # Push all registered APIs into the global schema
        for api in self.api_registry.all():
            await self.schema.add_api(api)

        # We're running as a worker now (e.g. lightbus run), so
        # do the lazy loading immediately
        await self.lazy_load_now()

        # Setup schema monitoring
        monitor_task = asyncio.ensure_future(
            queue_exception_checker(self.schema.monitor(), self.error_queue)
        )

        logger.info("Executing before_worker_start & on_start hooks...")
        await self.hook_registry.execute("before_worker_start")
        logger.info("Execution of before_worker_start & on_start hooks was successful")

        # Setup RPC consumption
        if Feature.RPCS in self.features:
            consume_rpc_task = asyncio.ensure_future(
                queue_exception_checker(self.consume_rpcs(), self.error_queue)
            )
        else:
            consume_rpc_task = None

        # Start off any registered event listeners
        if Feature.EVENTS in self.features:
            await self.event_client.start_registered_listeners()

        # Start off any background tasks
        if Feature.TASKS in self.features:
            for coroutine in self._background_coroutines:
                task = asyncio.ensure_future(queue_exception_checker(coroutine, self.error_queue))
                self._background_tasks.append(task)

        self._worker_tasks.add(consume_rpc_task)
        self._worker_tasks.add(monitor_task)
async def start_worker ( self ) : # Ensure an event loop exists get_event_loop ( ) self . _worker_tasks = set ( ) # Start monitoring for errors on the error queue error_monitor_task = asyncio . ensure_future ( self . error_monitor ( ) ) self . _error_monitor_task = error_monitor_task self . _worker_tasks . add ( self . _error_monitor_task ) # Features setup & logging if not self . api_registry . all ( ) and Feature . RPCS in self . features : logger . info ( "Disabling serving of RPCs as no APIs have been registered" ) self . features . remove ( Feature . RPCS ) logger . info ( LBullets ( f"Enabled features ({len(self.features)})" , items = [ f . value for f in self . features ] ) ) disabled_features = set ( ALL_FEATURES ) - set ( self . features ) logger . info ( LBullets ( f"Disabled features ({len(disabled_features)})" , items = [ f . value for f in disabled_features ] , ) ) # Api logging logger . info ( LBullets ( "APIs in registry ({})" . format ( len ( self . api_registry . all ( ) ) ) , items = self . api_registry . names ( ) , ) ) # Push all registered APIs into the global schema for api in self . api_registry . all ( ) : await self . schema . add_api ( api ) # We're running as a worker now (e.g. lightbus run), so # do the lazy loading immediately await self . lazy_load_now ( ) # Setup schema monitoring monitor_task = asyncio . ensure_future ( queue_exception_checker ( self . schema . monitor ( ) , self . error_queue ) ) logger . info ( "Executing before_worker_start & on_start hooks..." ) await self . hook_registry . execute ( "before_worker_start" ) logger . info ( "Execution of before_worker_start & on_start hooks was successful" ) # Setup RPC consumption if Feature . RPCS in self . features : consume_rpc_task = asyncio . ensure_future ( queue_exception_checker ( self . consume_rpcs ( ) , self . error_queue ) ) else : consume_rpc_task = None # Start off any registered event listeners if Feature . EVENTS in self . features : await self . event_client . start_registered_listeners ( ) # Start off any background tasks if Feature . TASKS in self . features : for coroutine in self . _background_coroutines : task = asyncio . ensure_future ( queue_exception_checker ( coroutine , self . error_queue ) ) self . _background_tasks . append ( task ) self . _worker_tasks . add ( consume_rpc_task ) self . _worker_tasks . add ( monitor_task )