RefDataOnHeapStore extends AbstractRefDataStore { @Override public Optional<RefDataValue> getValue(final MapDefinition mapDefinition, final String key) { Optional<RefDataValue> result; KeyValueMapKey keyValueMapKey = new KeyValueMapKey(mapDefinition, key); RefDataValue refDataValue = keyValueMap.get(keyValueMapKey); if (refDataValue != null) { result = Optional.of(refDataValue); } else { try { final long keyLong = Long.parseLong(key); NavigableMap<Range<Long>, RefDataValue> rangeSubMap = rangeValueNestedMap.get(mapDefinition); result = getValueByRange(rangeSubMap, keyLong); } catch (NumberFormatException e) { boolean doesStoreContainRanges = rangeValueNestedMap.containsKey(mapDefinition); if (doesStoreContainRanges) { throw new RuntimeException(LogUtil.message( "Key {} cannot be used with the range store as it cannot be converted to a long", key), e); } result = Optional.empty(); } } final Optional<RefDataValue> result2 = result; LAMBDA_LOGGER.trace(() -> LogUtil.message("getValue({}, {}) returning {}", mapDefinition, key, result2)); return result; } RefDataOnHeapStore(); @Override StorageType getStorageType(); @Override Optional<RefDataProcessingInfo> getAndTouchProcessingInfo(final RefStreamDefinition refStreamDefinition); @Override boolean isDataLoaded(final RefStreamDefinition refStreamDefinition); @Override boolean exists(final MapDefinition mapDefinition); @Override Optional<RefDataValue> getValue(final MapDefinition mapDefinition, final String key); @Override boolean consumeValueBytes(final MapDefinition mapDefinition,
                                     final String key,
                                     final Consumer<TypedByteBuffer> valueBytesConsumer); @Override List<RefStoreEntry> list(final int limit); @Override List<RefStoreEntry> list(final int limit,
                                    final Predicate<RefStoreEntry> filter); @Override long getKeyValueEntryCount(); @Override long getKeyRangeValueEntryCount(); @Override long getProcessingInfoEntryCount(); @Override void purgeOldData(); @Override void logAllContents(); @Override void logAllContents(final Consumer<String> logEntryConsumer);  }
@Test void testTreeMapReverseOrdering() { Comparator<Long> comparator = Comparator.reverseOrder(); NavigableMap<Long, String> treeMap = new TreeMap<>(comparator); treeMap.put(1L, "one"); treeMap.put(3L, "three"); treeMap.put(5L, "five"); treeMap.put(7L, "seven"); treeMap.forEach((key, value) -> LOGGER.info("{} => {}", key, value)); assertThat(treeMap.ceilingEntry(3L).getValue()).isEqualTo("three"); assertThat(treeMap.ceilingEntry(4L).getValue()).isEqualTo("three"); assertThat(treeMap.ceilingEntry(9L).getValue()).isEqualTo("seven"); Map.Entry<Long, String> entryThree = treeMap.ceilingEntry(4L); LOGGER.info("------------------"); SortedMap<Long, String> partMap = treeMap.tailMap(4L); partMap.forEach((key, value) -> LOGGER.info("{} => {}", key, value)); assertThat(treeMap.floorEntry(-1L).getValue()).isEqualTo("one"); assertThat(treeMap.floorEntry(3L).getValue()).isEqualTo("three"); assertThat(treeMap.floorEntry(4L).getValue()).isEqualTo("five"); }
@Disabled @Test void testLoaderConcurrency() { final RefStreamDefinition refStreamDefinition = buildUniqueRefStreamDefinition(); final long effectiveTimeMs = System.currentTimeMillis(); final MapDefinition mapDefinitionKey = new MapDefinition(refStreamDefinition, "MyKeyMap"); final MapDefinition mapDefinitionRange = new MapDefinition(refStreamDefinition, "MyRangeMap"); final int recCount = 1_000; Runnable loadTask = () -> { LOGGER.debug("Running loadTask on thread {}", Thread.currentThread().getName()); try { refDataStore.doWithLoaderUnlessComplete(refStreamDefinition, effectiveTimeMs, refDataLoader -> { refDataLoader.setCommitInterval(200); refDataLoader.initialise(false); long rangeStartInc = 0; long rangeEndExc; for (int i = 0; i < recCount; i++) { refDataLoader.put(mapDefinitionKey, "key" + i, StringValue.of("Value" + i)); rangeEndExc = rangeStartInc + 10; Range<Long> range = new Range<>(rangeStartInc, rangeEndExc); rangeStartInc = rangeEndExc; refDataLoader.put(mapDefinitionRange, range, StringValue.of("Value" + i)); } refDataLoader.completeProcessing(); LOGGER.debug("Finished loading data"); }); LOGGER.debug("Getting values"); LAMBDA_LOGGER.logDurationIfDebugEnabled(() -> { IntStream.range(0, recCount) .boxed() .sorted(Comparator.reverseOrder()) .forEach(i -> { Optional<RefDataValue> optValue = refDataStore.getValue(mapDefinitionKey, "key" + i); assertThat(optValue.isPresent()); optValue = refDataStore.getValue(mapDefinitionKey, Integer.toString((i * 10) + 5)); assertThat(optValue.isPresent()); }); }, () -> LogUtil.message("Getting {} entries, twice", recCount)); } catch (Exception e) { throw new RuntimeException(e); } LOGGER.debug("Finished running loadTask on thread {}", Thread.currentThread().getName()); }; ExecutorService executorService = Executors.newFixedThreadPool(6); List<CompletableFuture<Void>> futures = IntStream.rangeClosed(1, 10) .boxed() .map(i -> { LOGGER.debug("Running async task on thread {}", Thread.currentThread().getName()); final CompletableFuture<Void> future = CompletableFuture.runAsync(loadTask, executorService); LOGGER.debug("Got future"); return future; }) .collect(Collectors.toList()); futures.forEach(voidCompletableFuture -> { try { voidCompletableFuture.get(); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } }); LOGGER.debug("Finished all"); }
@Disabled @Test void testLoaderConcurrency_multipleStreamDefs() { final long effectiveTimeMs = System.currentTimeMillis(); final int recCount = 1_000; Consumer<RefStreamDefinition> loadTask = refStreamDefinition -> { final MapDefinition mapDefinitionKey = new MapDefinition(refStreamDefinition, "MyKeyMap"); final MapDefinition mapDefinitionRange = new MapDefinition(refStreamDefinition, "MyRangeMap"); LOGGER.debug("Running loadTask on thread {}", Thread.currentThread().getName()); try { refDataStore.doWithLoaderUnlessComplete(refStreamDefinition, effectiveTimeMs, refDataLoader -> { refDataLoader.setCommitInterval(200); refDataLoader.initialise(false); long rangeStartInc = 0; long rangeEndExc; for (int i = 0; i < recCount; i++) { refDataLoader.put(mapDefinitionKey, "key" + i, StringValue.of("Value" + i)); rangeEndExc = rangeStartInc + 10; Range<Long> range = new Range<>(rangeStartInc, rangeEndExc); rangeStartInc = rangeEndExc; refDataLoader.put(mapDefinitionRange, range, StringValue.of("Value" + i)); } refDataLoader.completeProcessing(); LOGGER.debug("Finished loading data"); }); LOGGER.debug("Getting values"); LAMBDA_LOGGER.logDurationIfDebugEnabled(() -> { IntStream.range(0, recCount) .boxed() .sorted(Comparator.reverseOrder()) .forEach(i -> { Optional<RefDataValue> optValue = refDataStore.getValue(mapDefinitionKey, "key" + i); assertThat(optValue.isPresent()); optValue = refDataStore.getValue(mapDefinitionKey, Integer.toString((i * 10) + 5)); assertThat(optValue.isPresent()); }); }, () -> LogUtil.message("Getting {} entries, twice", recCount)); } catch (Exception e) { throw new RuntimeException(e); } LOGGER.debug("Finished running loadTask on thread {}", Thread.currentThread().getName()); }; ExecutorService executorService = Executors.newFixedThreadPool(6); List<CompletableFuture<Void>> futures = IntStream.rangeClosed(1, 10) .boxed() .map(i -> { LOGGER.debug("Running async task on thread {}", Thread.currentThread().getName()); RefStreamDefinition refStreamDefinition = buildUniqueRefStreamDefinition(); final CompletableFuture<Void> future = CompletableFuture.runAsync( () -> loadTask.accept(refStreamDefinition), executorService); LOGGER.debug("Got future"); return future; }) .collect(Collectors.toList()); futures.forEach(voidCompletableFuture -> { try { voidCompletableFuture.get(); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } }); LOGGER.debug("Finished all"); }