pub fn import_core<G>(&mut self, scope: &G, name: &str) -> (Arranged<G, TraceAgent<Tr>>, ShutdownButton<CapabilitySet<Tr::Time>>)
    where
        G: Scope<Timestamp=Tr::Time>,
    {
        let trace = self.clone();

        let mut shutdown_button = None;

        let stream = {

            let shutdown_button_ref = &mut shutdown_button;
            source(scope, name, move |capability, info| {

                let capabilities = Rc::new(RefCell::new(Some(CapabilitySet::new())));

                let activator = scope.activator_for(Rc::clone(&info.address));
                let queue = self.new_listener(activator);

                let activator = scope.activator_for(info.address);
                *shutdown_button_ref = Some(ShutdownButton::new(capabilities.clone(), activator));

                capabilities.borrow_mut().as_mut().unwrap().insert(capability);

                move |output| {

                    let mut capabilities = capabilities.borrow_mut();
                    if let Some(ref mut capabilities) = *capabilities {

                        let mut borrow = queue.1.borrow_mut();
                        for instruction in borrow.drain(..) {
                            match instruction {
                                TraceReplayInstruction::Frontier(frontier) => {
                                    capabilities.downgrade(&frontier.borrow()[..]);
                                },
                                TraceReplayInstruction::Batch(batch, hint) => {
                                    if let Some(time) = hint {
                                        if !batch.is_empty() {
                                            let delayed = capabilities.delayed(&time);
                                            output.session(&delayed).give(batch);
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            })
        };

        (Arranged { stream, trace }, shutdown_button.unwrap())
    }
Imports an arrangement into the supplied scope.

# Examples

```
use timely::Config;
use timely::dataflow::ProbeHandle;
use timely::dataflow::operators::Probe;
use differential_dataflow::input::InputSession;
use differential_dataflow::operators::arrange::ArrangeBySelf;
use differential_dataflow::operators::reduce::Reduce;
use differential_dataflow::trace::Trace;

::timely::execute(Config::thread(), |worker| {

let mut input = InputSession::<_,(),isize>::new();
let mut probe = ProbeHandle::new();

// create a first dataflow
let mut trace = worker.dataflow::<u32,_,_>(|scope| {
// create input handle and collection.
input.to_collection(scope)
.arrange_by_self()
.trace
});

// do some work.
worker.step();
worker.step();

// create a second dataflow
let mut shutdown = worker.dataflow(|scope| {
let (arrange, button) = trace.import_core(scope, "Import");
arrange.stream.probe_with(&mut probe);
button
});

worker.step();
worker.step();
assert!(!probe.done());

shutdown.press();

worker.step();
worker.step();
assert!(probe.done());

}).unwrap();
```
differential-dataflow/src/operators/arrange/agent.rs