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);
            }
        }
    }
Called from worker thread.
Processes the incoming queue for read and writes.