pub fn new_listener(&mut self, activator: Activator) -> TraceAgentQueueReader<Tr>
    {
        // create a new queue for progress and batch information.
        let mut new_queue = VecDeque::new();

        // add the existing batches from the trace
        let mut upper = None;
        self.trace
            .borrow_mut()
            .trace
            .map_batches(|batch| {
                new_queue.push_back(TraceReplayInstruction::Batch(batch.clone(), Some(<Tr::Time as Timestamp>::minimum())));
                upper = Some(batch.upper().clone());
            });

        if let Some(upper) = upper {
            new_queue.push_back(TraceReplayInstruction::Frontier(upper));
        }

        let reference = Rc::new((activator, RefCell::new(new_queue)));

        // wraps the queue in a ref-counted ref cell and enqueue/return it.
        if let Some(queue) = self.queues.upgrade() {
            queue.borrow_mut().push(Rc::downgrade(&reference));
        }
        reference.0.activate();
        reference
    }
Attaches a new shared queue to the trace.

The queue is first populated with existing batches from the trace,
The queue will be immediately populated with existing historical batches from the trace, and until the reference
is dropped the queue will receive new batches as produced by the source `arrange` operator.
differential-dataflow/src/operators/arrange/agent.rs