slumberdb/slumberdb-service-model/src/main/java/org/boon/slumberdb/stores/BaseDataStore.java
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);
            }
        }
    }
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);
            }
        }
    }
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.
Called from worker thread . Processes the incoming queue for read and writes .