void start() {
        Phaser synchronizer = phaser.get();
        if (synchronizer != null)
            synchronizer.register();

        new Thread(() -> {
            try {
                // Iterate through all documents and pass on to data handler
                outstanding.forEach((id, document) -> {
                    if (state.get() != State.RUNNING)
                        return;

                    try {
                        if (selector.accepts(new DocumentPut(document)) != Result.TRUE)
                            return;
                    }
                    catch (RuntimeException e) {
                        return;
                    }

                    Document copy = new Document(document.getDataType(), document.getId());
                    new FieldSetRepo().copyFields(document, copy, fieldSet);


                    if (synchronizer != null)
                        synchronizer.arriveAndAwaitAdvance();

                    if (data != null)
                        data.onMessage(new PutDocumentMessage(new DocumentPut(copy)),
                                       new AckToken(id));
                    else
                        outstanding.remove(id);

                    if (synchronizer != null)
                        synchronizer.arriveAndAwaitAdvance();
                });
                // Transition to a terminal state when done
                state.updateAndGet(current -> {
                    switch (current) {
                        case RUNNING:
                            control.onDone(VisitorControlHandler.CompletionCode.SUCCESS, "Success");
                            return State.SUCCESS;
                        case ABORTED:
                            control.onDone(VisitorControlHandler.CompletionCode.ABORTED, "Aborted by user");
                            return State.ABORTED;
                        default:
                            control.onDone(VisitorControlHandler.CompletionCode.FAILURE, "Unexpected state '" + current + "'");;
                            return State.FAILURE;
                    }
                });
            }
            // Transition to failure terminal state on error
            catch (Exception e) {
                state.set(State.FAILURE);
                outstanding.clear();
                control.onDone(VisitorControlHandler.CompletionCode.FAILURE, Exceptions.toMessageString(e));
            }
            finally {
                if (synchronizer != null)
                    synchronizer.arriveAndDeregister();
                if (data != null)
                    data.onDone();
            }
        }).start();
    }
Local visitor session that copies and iterates through all items in the local document access.
Each document must be ack'ed for the session to be done visiting, unless the destination is remote.
Only document puts are sent by this session, and this is done from a separate thread.

@author jonmv
documentapi/src/main/java/com/yahoo/documentapi/local/LocalVisitorSession.java