Repository: celery/celery
File: celery/worker/consumer/consumer.py (Python)

Code before:
                        and self._new_prefetch_count <= self.max_prefetch_count
                    ):
                        ack_log_error_promise.then(self._restore_prefetch_count_after_connection_restart,
                                                   on_error=self._restore_prefetch_count_after_connection_restart)
                        reject_log_error_promise.then(self._restore_prefetch_count_after_connection_restart,
                                                      on_error=self._restore_prefetch_count_after_connection_restart)

                    strategy(
                        message, payload,
                        ack_log_error_promise,
                        reject_log_error_promise,
                        callbacks,
                    )
                except (InvalidTaskError, ContentDisallowed) as exc:
                    return on_invalid_task(payload, message, exc)
                except DecodeError as exc:
                    return self.on_decode_error(message, exc)

        return on_task_received

    def _restore_prefetch_count_after_connection_restart(self, p, *args):
        with self.qos._mutex:
            if self._maximum_prefetch_restored:
                p.cancel()
                return

            new_prefetch_count = min(self.max_prefetch_count, self._new_prefetch_count)
            self.qos.value = self.initial_prefetch_count = new_prefetch_count
            self.qos.set(self.qos.value)

            self._maximum_prefetch_restored = new_prefetch_count == self.max_prefetch_count

    @property
    def max_prefetch_count(self):
        return self.pool.num_processes * self.prefetch_multiplier

    @property
    def _new_prefetch_count(self):
        return self.qos.value + self.prefetch_multiplier

    def __repr__(self):
        """``repr(self)``."""
        return '<Consumer: {self.hostname} ({state})>'.format(
            self=self, state=self.blueprint.human_state(),
        )


class Evloop(bootsteps.StartStopStep):
    """Event loop service.

    Note:

Review comment (suggestion):
@thedrow based on my test, I need to add *args as an argument
```
>>> import vine
>>> from vine.promises import promise
def raise_exception():
    raise KeyError('Foo')

>>> p = promise(raise_exception, on_error=print)
>>> p()
Foo
```
Foo is passed, as argument to on_error

Code after:
                    )

                    if (
                        not self._maximum_prefetch_restored
                        and self.restart_count > 0
                        and self._new_prefetch_count <= self.max_prefetch_count
                    ):
                        ack_log_error_promise.then(self._restore_prefetch_count_after_connection_restart,
                                                   on_error=self._restore_prefetch_count_after_connection_restart)
                        reject_log_error_promise.then(self._restore_prefetch_count_after_connection_restart,
                                                      on_error=self._restore_prefetch_count_after_connection_restart)

                    strategy(
                        message, payload,
                        ack_log_error_promise,
                        reject_log_error_promise,
                        callbacks,
                    )
                except (InvalidTaskError, ContentDisallowed) as exc:
                    return on_invalid_task(payload, message, exc)
                except DecodeError as exc:
                    return self.on_decode_error(message, exc)

        return on_task_received

    def _restore_prefetch_count_after_connection_restart(self, p, *args):
        with self.qos._mutex:
            if self._maximum_prefetch_restored:
                p.cancel()
                return

            new_prefetch_count = min(self.max_prefetch_count, self._new_prefetch_count)
            self.qos.value = self.initial_prefetch_count = new_prefetch_count
            self.qos.set(self.qos.value)

            self._maximum_prefetch_restored = new_prefetch_count == self.max_prefetch_count

    @property
    def max_prefetch_count(self):
        return self.pool.num_processes * self.prefetch_multiplier

    @property
    def _new_prefetch_count(self):
        return self.qos.value + self.prefetch_multiplier

    def __repr__(self):
        """``repr(self)``."""
        return '<Consumer: {self.hostname} ({state})>'.format(
            self=self, state=self.blueprint.human_state(),
        )