private void handleDocumentMessage(DocumentMessage msg) {
        Reply reply = msg.createReply();
        msg.swapState(reply);

        if (params.getLocalDataHandler() == null) {
            log.log(Level.SEVERE, sessionName + ": Got visitor data back to client with no local data destination.");
            reply.addError(new Error(ErrorCode.APP_FATAL_ERROR, "Visitor data with no local data destination"));
            receiver.reply(reply);
            return;
        }
        AckToken token = null;
        synchronized (stateMonitor) {
            // Prevent any further data callbacks if we have transitioned away from a working state.
            // Additionally, track that we have an in-flight message being processed so that we
            // do not trigger the visitor completion logic while there's work still being done.
            // Together these form a logical barrier that ensures no callbacks will be received _after_
            // onDone has been called.
            // The processing count tracking is reversed when ACKing the message on the session.
            // Ensuring 1-1 callback -> ACKing is the caller's responsibility.
            if (state.getState() == State.WORKING) {
                ++locallyProcessingCount;
                token = new AckToken(reply);
            }
        }
        try {
            if (token != null) {
                params.getLocalDataHandler().onMessage(msg, token);
            } else {
                // Local processing counter _not_ incremented
                reply.addError(new Error(ErrorCode.APP_FATAL_ERROR, "Visitor has been shut down"));
                receiver.reply(reply);
            }
        } catch (Exception e) {
            handleMessageProcessingException(reply, e, "DocumentMessage");
            synchronized (stateMonitor) {
                decrementLocallyProcessingCounter();
            }
            // Immediately reply since we cannot count on AckToken being registered
            receiver.reply(reply);
        }
    }
NOTE: not called from within lock, function must take lock itself
documentapi/src/main/java/com/yahoo/documentapi/messagebus/MessageBusVisitorSession.java