slumberdb/slumberdb-service-model/src/main/java/org/boon/slumberdb/stores/BaseDataStore.java
private void processReadQueue() throws InterruptedException {

        ReadStatus readStatus = new ReadStatus();


        while (true) {

            DataStoreRequest request = readOperationsQueue.poll(dataStoreConfig.pollTimeoutMS(), TimeUnit.MILLISECONDS);

            while (request != null) {

                readStatus.tracker.addCall(request, outputDataQueue);
                readOperationsBatch.add(request);
                if (readOperationsBatch.size() > dataStoreConfig.processQueueMaxBatchSize()) {
                    break;
                }
                request = readOperationsQueue.poll();

            }

            if (readOperationsBatch.size() > 0) {

                try {
                    recievedReadBatch(new ArrayList<>(readOperationsBatch));
                } finally {
                    readOperationsBatch.clear();
                }
            } else {
                flushReadsIfNeeded();
            }


            if (readStatus.readBatchSize.size() > 1_000) {


                StatCount count;
                final long now = Timer.timer().time();


                count = new StatCount(now, DataStoreSource.SERVER, Action.GET_STATS, "Thread TIME USER  BaseDataStore "
                        + Thread.currentThread().getName(), Sys.threadUserTime());
                this.outputDataQueue.put(count);

                count = new StatCount(now, DataStoreSource.SERVER, Action.GET_STATS, "Thread TIME CPU  BaseDataStore "
                        + Thread.currentThread().getName(), Sys.threadCPUTime());
                this.outputDataQueue.put(count);

                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.max", readStatus.readBatchSize.max());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.min", readStatus.readBatchSize.min());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.median", readStatus.readBatchSize.median());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.mean", readStatus.readBatchSize.mean());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.standardDeviation", readStatus.readBatchSize.standardDeviation());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.variance", readStatus.readBatchSize.variance());
                outputDataQueue.put(count);


                readStatus.readBatchSize.clear();

            }
        }

    }
private void processReadQueue() throws InterruptedException {

        ReadStatus readStatus = new ReadStatus();


        while (true) {

            DataStoreRequest request = readOperationsQueue.poll(dataStoreConfig.pollTimeoutMS(), TimeUnit.MILLISECONDS);

            while (request != null) {

                readStatus.tracker.addCall(request, outputDataQueue);
                readOperationsBatch.add(request);
                if (readOperationsBatch.size() > dataStoreConfig.processQueueMaxBatchSize()) {
                    break;
                }
                request = readOperationsQueue.poll();

            }

            if (readOperationsBatch.size() > 0) {

                try {
                    recievedReadBatch(new ArrayList<>(readOperationsBatch));
                } finally {
                    readOperationsBatch.clear();
                }
            } else {
                flushReadsIfNeeded();
            }


            if (readStatus.readBatchSize.size() > 1_000) {


                StatCount count;
                final long now = Timer.timer().time();


                count = new StatCount(now, DataStoreSource.SERVER, Action.GET_STATS, "Thread TIME USER  BaseDataStore "
                        + Thread.currentThread().getName(), Sys.threadUserTime());
                this.outputDataQueue.put(count);

                count = new StatCount(now, DataStoreSource.SERVER, Action.GET_STATS, "Thread TIME CPU  BaseDataStore "
                        + Thread.currentThread().getName(), Sys.threadCPUTime());
                this.outputDataQueue.put(count);

                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.max", readStatus.readBatchSize.max());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.min", readStatus.readBatchSize.min());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.median", readStatus.readBatchSize.median());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.mean", readStatus.readBatchSize.mean());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.standardDeviation", readStatus.readBatchSize.standardDeviation());
                outputDataQueue.put(count);


                count = new StatCount(now, source, Action.GET, "BaseDataStore readStatus.readBatchSize.variance", readStatus.readBatchSize.variance());
                outputDataQueue.put(count);


                readStatus.readBatchSize.clear();

            }
        }

    }
Called from worker thread.

@throws InterruptedException