async def handle_control_event(self, message):
        """Handle an event from the Channels layer.

        Channels layer callback, do not call directly.
        """
        cmd = message[WorkerProtocol.COMMAND]
        logger.debug(__("Manager worker got channel command '{}'.", cmd))

        # Prepare settings for use; Django overlaid by state overlaid by
        # anything immediate in the current packet.
        immediates = {}
        if cmd == WorkerProtocol.COMMUNICATE:
            immediates = message.get(WorkerProtocol.COMMUNICATE_SETTINGS, {}) or {}
        override = self.state.settings_override or {}
        override.update(immediates)
        self.settings_actual = self._marshal_settings()
        self.settings_actual.update(override)

        if cmd == WorkerProtocol.COMMUNICATE:
            try:
                await database_sync_to_async(self._data_scan)(**message[WorkerProtocol.COMMUNICATE_EXTRA])
            except Exception:
                logger.exception("Unknown error occured while processing communicate control command.")
                raise
            finally:
                await self.sync_counter.dec('communicate')

        elif cmd == WorkerProtocol.FINISH:
            try:
                data_id = message[WorkerProtocol.DATA_ID]
                data_location = DataLocation.objects.get(data__id=data_id)
                if not getattr(settings, 'FLOW_MANAGER_KEEP_DATA', False):
                    try:
                        def handle_error(func, path, exc_info):
                            """Handle permission errors while removing data directories."""
                            if isinstance(exc_info[1], PermissionError):
                                os.chmod(path, 0o700)
                                shutil.rmtree(path)

                        # Remove secrets directory, but leave the rest of the runtime directory
                        # intact. Runtime directory will be removed during data purge, when the
                        # data object is removed.
                        secrets_dir = os.path.join(
                            self._get_per_data_dir('RUNTIME_DIR', data_location.subpath),
                            ExecutorFiles.SECRETS_DIR
                        )
                        shutil.rmtree(secrets_dir, onerror=handle_error)
                    except OSError:
                        logger.exception("Manager exception while removing data runtime directory.")

                if message[WorkerProtocol.FINISH_SPAWNED]:
                    await database_sync_to_async(self._data_scan)(**message[WorkerProtocol.FINISH_COMMUNICATE_EXTRA])
            except Exception:
                logger.exception(
                    "Unknown error occured while processing finish control command.",
                    extra={'data_id': data_id}
                )
                raise
            finally:
                await self.sync_counter.dec('executor')

        elif cmd == WorkerProtocol.ABORT:
            await self.sync_counter.dec('executor')

        else:
            logger.error(__("Ignoring unknown manager control command '{}'.", cmd))
async def handle_control_event ( self , message ) : cmd = message [ WorkerProtocol . COMMAND ] logger . debug ( __ ( "Manager worker got channel command '{}'." , cmd ) ) immediates = { } if cmd == WorkerProtocol . COMMUNICATE : immediates = message . get ( WorkerProtocol . COMMUNICATE_SETTINGS , { } ) or { } override = self . state . settings_override or { } override . update ( immediates ) self . settings_actual = self . _marshal_settings ( ) self . settings_actual . update ( override ) if cmd == WorkerProtocol . COMMUNICATE : try : await database_sync_to_async ( self . _data_scan ) ( ** message [ WorkerProtocol . COMMUNICATE_EXTRA ] ) except Exception : logger . exception ( "Unknown error occured while processing communicate control command." ) raise finally : await self . sync_counter . dec ( 'communicate' ) elif cmd == WorkerProtocol . FINISH : try : data_id = message [ WorkerProtocol . DATA_ID ] data_location = DataLocation . objects . get ( data__id = data_id ) if not getattr ( settings , 'FLOW_MANAGER_KEEP_DATA' , False ) : try : def handle_error ( func , path , exc_info ) : if isinstance ( exc_info [ 1 ] , PermissionError ) : os . chmod ( path , 0o700 ) shutil . rmtree ( path ) secrets_dir = os . path . join ( self . _get_per_data_dir ( 'RUNTIME_DIR' , data_location . subpath ) , ExecutorFiles . SECRETS_DIR ) shutil . rmtree ( secrets_dir , onerror = handle_error ) except OSError : logger . exception ( "Manager exception while removing data runtime directory." ) if message [ WorkerProtocol . FINISH_SPAWNED ] : await database_sync_to_async ( self . _data_scan ) ( ** message [ WorkerProtocol . FINISH_COMMUNICATE_EXTRA ] ) except Exception : logger . exception ( "Unknown error occured while processing finish control command." , extra = { 'data_id' : data_id } ) raise finally : await self . sync_counter . dec ( 'executor' ) elif cmd == WorkerProtocol . ABORT : await self . sync_counter . dec ( 'executor' ) else : logger . error ( __ ( "Ignoring unknown manager control command '{}'." , cmd ) )
Handle an event from the Channels layer.

        Channels layer callback, do not call directly.