no way to catch the redis.exceptions.ConnectionError on pubsub.get_message

[alor]

when running the worker (from cli or with a custom class) the subscribe() method does not pass any exception_handler to self.pubsub.run_in_thread(sleep_time=0.2, daemon=True) so if a disconnection from Redis happens while waiting for messages in the pubsub channel, this code (inside redis python driver) is triggered:

            try:
                pubsub.get_message(ignore_subscribe_messages=True, timeout=sleep_time)
            except BaseException as e:
                if self.exception_handler is None:
                    raise
                self.exception_handler(e, pubsub, self)

and since there is no exception_handler the exception is raised.

the issue is that this code runs in a separate thread PubSubWorkerThread(threading.Thread) so there is no way to rescue from it in the worker class.

the program works correctly and reconnects after Redis connection is restored, but it's annoying to receive reports from bugsnag (setup as global catch all exceptions) for an unhandled exception that can occur sometimes but cannot be "silenced" since it's perfectly fine to get it.

[selwin]

I think the solution would be to configure bugsnag to ignore this exception. In addition, we can add logging statements around this area so users are aware of this. Mind opening a PR?

[alor]

are you suggesting to add a custom exception handler that only logs the issue and pass without re-raising the exception?
or you want the user to be able to specify a custom handler?

[selwin]

My suggestion was simply to add logging statements so we are aware that this is happening.

[alor]

something like this, will be ok? if this is what you mean, I'll create a PR with this changes

   def subscribe(self):
        """Subscribe to this worker's channel"""
        def handle_redis_exception(e, pubsub, thread):
            self.log.warn('Worker %s: Redis exception: %s', self.key, str(e))

        self.log.info('Subscribing to channel %s', self.pubsub_channel_name)
        self.pubsub = self.connection.pubsub()
        self.pubsub.subscribe(**{self.pubsub_channel_name: self.handle_payload})
        self.pubsub_thread = self.pubsub.run_in_thread(sleep_time=0.2, daemon=True, exception_handler=handle_redis_exception)