private void processWriteQueue ( ) throws InterruptedException { WriteStatus status = new WriteStatus ( ) ; while ( true ) { DataStoreRequest operation = writeOperationsQueue . poll ( dataStoreConfig . pollTimeoutMS ( ) , TimeUnit . MILLISECONDS ) ; while ( operation != null ) { status . tracker . addCall ( operation , outputDataQueue ) ; writeOperationsBatch . add ( operation ) ; if ( writeOperationsBatch . size ( ) > dataStoreConfig . processQueueMaxBatchSize ( ) ) { break ; } operation = writeOperationsQueue . poll ( ) ; } if ( writeOperationsBatch . size ( ) > 0 ) { try { status . writeBatchSize . add ( writeOperationsBatch . size ( ) ) ; recievedWriteBatch ( new ArrayList <> ( writeOperationsBatch ) ) ; } finally { writeOperationsBatch . clear ( ) ; } } else { flushWritesIfNeeded ( ) ; } if ( status . writeBatchSize . size ( ) > 1000 ) { status . sendBatchSize ( source , outputDataQueue ) ; } } }
Called from worker thread . Processes the incoming queue for read and writes .