protected synchronized Instance processBuffers(){
  if (m_firstBuffer != null && m_secondBuffer != null && m_firstBuffer.size() > 0 && m_secondBuffer.size() > 0) {
    if (m_stopRequested.get()) {
      return null;
    }
    InstanceHolder firstH=m_firstBuffer.peek();
    InstanceHolder secondH=m_secondBuffer.peek();
    Instance first=firstH.m_instance;
    Instance second=secondH.m_instance;
    int cmp=compare(first,second,firstH,secondH);
    if (cmp == 0) {
      Instance newInst=generateMergedInstance(m_firstBuffer.remove(),m_secondBuffer.remove());
      return newInst;
    }
 else     if (cmp < 0) {
      do {
        m_firstBuffer.remove();
        if (m_firstBuffer.size() > 0) {
          firstH=m_firstBuffer.peek();
          first=firstH.m_instance;
          cmp=compare(first,second,firstH,secondH);
        }
      }
 while (cmp < 0 && m_firstBuffer.size() > 0);
    }
 else {
      do {
        m_secondBuffer.remove();
        if (m_secondBuffer.size() > 0) {
          secondH=m_secondBuffer.peek();
          second=secondH.m_instance;
          cmp=compare(first,second,firstH,secondH);
        }
      }
 while (cmp > 0 && m_secondBuffer.size() > 0);
    }
  }
  return null;
}

protected synchronized Instance processBuffers ( ) { if ( m_firstBuffer != null && m_secondBuffer != null && m_firstBuffer . size ( ) > 0 && m_secondBuffer . size ( ) > 0 ) { if ( m_stopRequested . get ( ) ) { return null ; } InstanceHolder firstH = m_firstBuffer . peek ( ) ; InstanceHolder secondH = m_secondBuffer . peek ( ) ; Instance first = firstH . m_instance ; Instance second = secondH . m_instance ; int cmp = compare ( first , second , firstH , secondH ) ; if ( cmp == 0 ) { Instance newInst = generateMergedInstance ( m_firstBuffer . remove ( ) , m_secondBuffer . remove ( ) ) ; return newInst ; } else if ( cmp < 0 ) { do { m_firstBuffer . remove ( ) ; if ( m_firstBuffer . size ( ) > 0 ) { firstH = m_firstBuffer . peek ( ) ; first = firstH . m_instance ; cmp = compare ( first , second , firstH , secondH ) ; } } while ( cmp < 0 && m_firstBuffer . size ( ) > 0 ) ; } else { do { m_secondBuffer . remove ( ) ; if ( m_secondBuffer . size ( ) > 0 ) { secondH = m_secondBuffer . peek ( ) ; second = secondH . m_instance ; cmp = compare ( first , second , firstH , secondH ) ; } } while ( cmp > 0 && m_secondBuffer . size ( ) > 0 ) ; } } return null ; }
process the current state of the two buffers . compares the head of both buffers and will return a merged instance if they match on the key fields . will remove instances from one or the other buffer if it is behind according to the keys comparison
