Add wait and wake to blocking topic

[slaroche]

No description provided.

[slaroche]

@isra17 up for review

[slaroche]

Should we do the same wait when the topic failed?

    def run(self):
           try:
               ...
            except Exception:
                self.logger.exception("Topic failed")
                await asyncio.sleep(self.sleep_time.total_seconds()) # -> await self.wait()
                continue

[isra17]

Should we do the same wait when the topic failed?

    def run(self):
           try:
               ...
            except Exception:
                self.logger.exception("Topic failed")
                await asyncio.sleep(self.sleep_time.total_seconds()) # -> await self.wait()
                continue

I think it's fine for now. The proper thing to do would be proper backoff retries.