Method run

tests/cardinality_one.rs:29–110

(mut self)

impl Run for Vec<Case<u64>> {
    fn run(mut self) {
        for mut case in self.drain(..) {
            timely::execute_directly(move |worker| {
                let mut server = Server::<u64, u64>::new(Default::default());
                let (send_results, results) = channel();

                dbg!(case.description);

                let mut deps = HashMap::new();
                for tx in case.transactions.iter() {
                    for datum in tx {
                        deps.entry(datum.2.clone()).or_insert_with(|| {
                            AttributeConfig::tx_time(InputSemantics::CardinalityOne)
                        });
                    }
                }

                let plan = case.plan;

                worker.dataflow::<u64, _, _>(|scope| {
                    for (dep, config) in deps.drain() {
                        server
                            .context
                            .internal
                            .create_transactable_attribute(&dep, config, scope)
                            .unwrap();
                    }

                    server
                        .test_single(
                            scope,
                            Rule {
                                name: "query".to_string(),
                                plan,
                            },
                        )
                        .inner
                        .sink(Pipeline, "Results", move |input| {
                            input.for_each(|_time, data| {
                                for datum in data.iter() {
                                    send_results.send(datum.clone()).unwrap()
                                }
                            });
                        });
                });

                let mut next_tx = 0;

                // Make sure we can pop off transactions one-by-one.
                case.transactions.reverse();

                for mut expected_tuples in case.expectations.drain(..) {
                    next_tx += 1;

                    if let Some(tx_data) = case.transactions.pop() {
                        server.transact(tx_data, 0, 0).unwrap();
                    }