public static MessageBusVisitorSession createForMessageBus(final MessageBus mbus,
                                                               final ScheduledExecutorService scheduledExecutorService,
                                                               final VisitorParameters params) throws ParseException {
        final AsyncTaskExecutor executor = new ThreadAsyncTaskExecutor(scheduledExecutorService);
        final MessageBusSenderFactory senderFactory = new MessageBusSenderFactory(mbus);
        final MessageBusReceiverFactory receiverFactory = new MessageBusReceiverFactory(mbus);
        final RoutingTable table = mbus.getRoutingTable(DocumentProtocol.NAME);

        return new MessageBusVisitorSession(params, executor, senderFactory, receiverFactory, table);
    }
We also have to explicitly keep track of how many messages we are processing
locally, as these may be asynchronous. We cannot invoke the control handler's
onDone() method until we have drained all such messages.
documentapi/src/main/java/com/yahoo/documentapi/messagebus/MessageBusVisitorSession.java