public boolean offer(long inUniqueId, TransactionInfoBaseMessage in)
    {
        ReplayEntry found = m_replayEntries.get(inUniqueId);

        if (in instanceof Iv2EndOfLogMessage) {
            m_mpiEOLReached = true;
            return true;
        }

        if (in instanceof MultiPartitionParticipantMessage) {
            //--------------------------------------------
            // DRv1 path, mark for future removal
            /*
             * DR sends multiple @LoadMultipartitionTable proc calls with the
             * same txnId, which is the snapshot txnId. For each partition,
             * there is a sentinel paired with the @LoadMultipartitionTable
             * call. Dedupe the sentinels the same way as we dedupe fragments,
             * so that there won't be sentinels end up in the sequencer where
             * matching fragments are deduped.
             */
            if (inUniqueId <= m_lastPolledFragmentUniqueId) {
                return true;
            }
            //--------------------------------------------
            if (found == null) {
                ReplayEntry newEntry = new ReplayEntry();
                newEntry.m_sentinelUniqueId = inUniqueId;
                m_replayEntries.put(inUniqueId, newEntry);
            }
            else {
                found.m_sentinelUniqueId = inUniqueId;
                assert(found.isReady());
            }
        }
        else if (in instanceof FragmentTaskMessage) {
            // already sequenced
            if (inUniqueId <= m_lastPolledFragmentUniqueId) {
                return false;
            }
            FragmentTaskMessage ftm  = (FragmentTaskMessage)in;
            if (found == null) {
                ReplayEntry newEntry = new ReplayEntry();
                newEntry.m_firstFragment = ftm;
                m_replayEntries.put(inUniqueId, newEntry);
            }
            else if (found.m_firstFragment == null) {
                found.m_firstFragment = ftm;
                assert(found.isReady());
            }
            else {
                found.addQueuedMessage(ftm);
            }
        }
        else if (in instanceof CompleteTransactionMessage) {
            // don't sequence CompleteTranscationMessage, throw them to scheduler directly
            return false;
        }
        else {
            //--------------------------------------------
            // DRv1 path, mark for future removal
            if (dedupe(inUniqueId, in) != null) {
                // Ignore an already seen txn
                return true;
            }
            //--------------------------------------------
            updateLastSeenUniqueId(inUniqueId, in);

            if (m_replayEntries.isEmpty() || !m_replayEntries.lastEntry().getValue().hasSentinel()) {
                // not-blocked work; rejected and not queued.
                return false;
            }
            else {
                // queued the message with the newest replayEntry
                m_replayEntries.lastEntry().getValue().addQueuedMessage(in);
            }
        }
        return true;
    }
public boolean offer ( long inUniqueId , TransactionInfoBaseMessage in ) { ReplayEntry found = m_replayEntries . get ( inUniqueId ) ; if ( in instanceof Iv2EndOfLogMessage ) { m_mpiEOLReached = true ; return true ; } if ( in instanceof MultiPartitionParticipantMessage ) { if ( inUniqueId <= m_lastPolledFragmentUniqueId ) { return true ; } if ( found == null ) { ReplayEntry newEntry = new ReplayEntry ( ) ; newEntry . m_sentinelUniqueId = inUniqueId ; m_replayEntries . put ( inUniqueId , newEntry ) ; } else { found . m_sentinelUniqueId = inUniqueId ; assert ( found . isReady ( ) ) ; } } else if ( in instanceof FragmentTaskMessage ) { if ( inUniqueId <= m_lastPolledFragmentUniqueId ) { return false ; } FragmentTaskMessage ftm = ( FragmentTaskMessage ) in ; if ( found == null ) { ReplayEntry newEntry = new ReplayEntry ( ) ; newEntry . m_firstFragment = ftm ; m_replayEntries . put ( inUniqueId , newEntry ) ; } else if ( found . m_firstFragment == null ) { found . m_firstFragment = ftm ; assert ( found . isReady ( ) ) ; } else { found . addQueuedMessage ( ftm ) ; } } else if ( in instanceof CompleteTransactionMessage ) { return false ; } else { if ( dedupe ( inUniqueId , in ) != null ) { return true ; } updateLastSeenUniqueId ( inUniqueId , in ) ; if ( m_replayEntries . isEmpty ( ) || ! m_replayEntries . lastEntry ( ) . getValue ( ) . hasSentinel ( ) ) { return false ; } else { m_replayEntries . lastEntry ( ) . getValue ( ) . addQueuedMessage ( in ) ; } } return true ; }
Offer a new message. Return false if the offered message can be run immediately.