Wait for all the messages on the given queue to be
        processed.  This method is only meant to be used in tests
        to wait for all the messages in a queue to be processed.

        Raises:
          QueueJoinTimeout: When the timeout elapses.
          QueueNotFound: If the given queue was never declared.

        Parameters:
          queue_name(str): The queue to wait on.
          fail_fast(bool): When this is True and any message gets
            dead-lettered during the join, then an exception will be
            raised.  This will be True by default starting with
            version 2.0.
          timeout(Optional[int]): The max amount of time, in
            milliseconds, to wait on this queue.

def join(self, queue_name, *, fail_fast=False, timeout=None):
        """Wait for all the messages on the given queue to be
        processed.  This method is only meant to be used in tests
        to wait for all the messages in a queue to be processed.

        Raises:
          QueueJoinTimeout: When the timeout elapses.
          QueueNotFound: If the given queue was never declared.

        Parameters:
          queue_name(str): The queue to wait on.
          fail_fast(bool): When this is True and any message gets
            dead-lettered during the join, then an exception will be
            raised.  This will be True by default starting with
            version 2.0.
          timeout(Optional[int]): The max amount of time, in
            milliseconds, to wait on this queue.
        """
        try:
            queues = [
                self.queues[queue_name],
                self.queues[dq_name(queue_name)],
            ]
        except KeyError:
            raise QueueNotFound(queue_name)

        deadline = timeout and time.monotonic() + timeout / 1000
        while True:
            for queue in queues:
                timeout = deadline and deadline - time.monotonic()
                join_queue(queue, timeout=timeout)

            # We cycle through $queue then $queue.DQ then $queue
            # again in case the messages that were on the DQ got
            # moved back on $queue.
            for queue in queues:
                if queue.unfinished_tasks:
                    break
            else:
                if fail_fast:
                    for message in self.dead_letters_by_queue[queue_name]:
                        raise message._exception from None

                return