pub fn new(trace: Tr, operator: OperatorInfo, logging: Option<crate::logging::Logger>) -> (Self, TraceWriter<Tr>)
    where
        Tr: Trace,
        Tr::Batch: Batch,
    {
        let trace = Rc::new(RefCell::new(TraceBox::new(trace)));
        let queues = Rc::new(RefCell::new(Vec::new()));

        if let Some(logging) = &logging {
            logging.log(
                crate::logging::TraceShare { operator: operator.global_id, diff: 1 }
            );
        }

        let reader = TraceAgent {
            trace: trace.clone(),
            queues: Rc::downgrade(&queues),
            logical_compaction: trace.borrow().logical_compaction.frontier().to_owned(),
            physical_compaction: trace.borrow().physical_compaction.frontier().to_owned(),
            temp_antichain: Antichain::new(),
            operator,
            logging,
        };

        let writer = TraceWriter::new(
            vec![<Tr::Time as Timestamp>::minimum()],
            Rc::downgrade(&trace),
            queues,
        );

        (reader, writer)
    }
Creates a new agent from a trace reader.
differential-dataflow/src/operators/arrange/agent.rs