Repository: spotify/luigi
File: luigi/scheduler.py (Python)

Code before:
        worker_id = kwargs['worker']
        batched_params, _ = self._state.get_batcher(worker_id, family)
        if batched_params:
            unbatched_params = {
                param: value
                for param, value in six.iteritems(params)
                if param not in batched_params
            }
        else:
            unbatched_params = params
        self._email_batcher.add_scheduling_fail(task_name, family, unbatched_params, expl, owners)

    @rpc_method()
    def add_worker(self, worker, info, **kwargs):
        self._state.get_worker(worker).add_info(info)

    @rpc_method()
    def disable_worker(self, worker):
        self._state.disable_workers({worker})

    @rpc_method()
    def set_worker_processes(self, worker, n):
        self._state.get_worker(worker).add_rpc_message('set_worker_processes', n=n)

    @rpc_method()
    def send_scheduler_message(self, worker, task, message):
        if not self._config.send_messages:
            return {"messageId": None}

        message_id = str(uuid.uuid4())
        self._state.get_worker(worker).add_rpc_message('dispatch_scheduler_message', task_id=task,
                                                       message_id=message_id, message=message)

        return {"messageId": message_id}

    @rpc_method()
    def add_scheduler_message_response(self, task_id, message_id, response):
        if self._state.has_task(task_id):
            task = self._state.get_task(task_id)
            task.scheduler_message_responses[message_id] = response

    @rpc_method()
    def get_scheduler_message_response(self, task_id, message_id):
        response = None
        if self._state.has_task(task_id):
            task = self._state.get_task(task_id)
            response = task.scheduler_message_responses.pop(message_id, None)
        return {"taskId": task_id, "response": response}

    @rpc_method()
    def is_pause_enabled(self):

Review comment (suggestion):
The parameter name `message` maybe be should be renamed to `content` or `payload` throughout to make it easier to follow. I  at least got confused that `message` became `content` in the scheduler.

Code after:
    def announce_scheduling_failure(self, task_name, family, params, expl, owners, **kwargs):
        if not self._config.batch_emails:
            return
        worker_id = kwargs['worker']
        batched_params, _ = self._state.get_batcher(worker_id, family)
        if batched_params:
            unbatched_params = {
                param: value
                for param, value in six.iteritems(params)
                if param not in batched_params
            }
        else:
            unbatched_params = params
        self._email_batcher.add_scheduling_fail(task_name, family, unbatched_params, expl, owners)

    @rpc_method()
    def add_worker(self, worker, info, **kwargs):
        self._state.get_worker(worker).add_info(info)

    @rpc_method()
    def disable_worker(self, worker):
        self._state.disable_workers({worker})

    @rpc_method()
    def set_worker_processes(self, worker, n):
        self._state.get_worker(worker).add_rpc_message('set_worker_processes', n=n)

    @rpc_method()
    def send_scheduler_message(self, worker, task, content):
        if not self._config.send_messages:
            return {"message_id": None}

        message_id = str(uuid.uuid4())
        self._state.get_worker(worker).add_rpc_message('dispatch_scheduler_message', task_id=task,
                                                       message_id=message_id, content=content)

        return {"message_id": message_id}

    @rpc_method()
    def add_scheduler_message_response(self, task_id, message_id, response):
        if self._state.has_task(task_id):
            task = self._state.get_task(task_id)
            task.scheduler_message_responses[message_id] = response

    @rpc_method()
    def get_scheduler_message_response(self, task_id, message_id):
        response = None
        if self._state.has_task(task_id):
            task = self._state.get_task(task_id)
            response = task.scheduler_message_responses.pop(message_id, None)
        return {"response": response}