@Test public void testStoringLargeAmountOfStatuses() { List<TaskRelocationStatus> statusList = newRelocationStatuses(10_000); Map<String, Optional<Throwable>> result = store.createTaskRelocationStatuses(statusList).block(); assertThat(result).hasSize(statusList.size()); long failures = result.values().stream().filter(Optional::isPresent).count(); assertThat(failures).isZero(); }
@Override public Mono<Map<String, Optional<Throwable>>> createTaskRelocationStatuses(List<TaskRelocationStatus> taskRelocationStatuses) { return Mono.defer(() -> { CompletionStage<int[]> asyncAction = JooqUtils.executeAsync(() -> { loadToCache(findNotCached(taskRelocationStatuses), dslContext.configuration()); List<StoreQuery<JRelocationStatusRecord>> queries = taskRelocationStatuses.stream() .map(this::newCreateOrUpdateQuery) .collect(Collectors.toList()); return dslContext .batch(queries) .execute(); }, dslContext); MonoProcessor<Map<String, Optional<Throwable>>> callerProcessor = MonoProcessor.create(); asyncAction.handle((result, error) -> { Map<String, Optional<Throwable>> resultMap = new HashMap<>(); if (error == null) { taskRelocationStatuses.forEach(p -> { resultMap.put(p.getTaskId(), Optional.empty()); statusesByTaskId.put(p.getTaskId(), p); }); callerProcessor.onNext(resultMap); } else { callerProcessor.onError(error); } return null; }); return callerProcessor; }); }
JooqTaskRelocationResultStore implements TaskRelocationResultStore { @Override public Mono<Map<String, Optional<Throwable>>> createTaskRelocationStatuses(List<TaskRelocationStatus> taskRelocationStatuses) { return Mono.defer(() -> { CompletionStage<int[]> asyncAction = JooqUtils.executeAsync(() -> { loadToCache(findNotCached(taskRelocationStatuses), dslContext.configuration()); List<StoreQuery<JRelocationStatusRecord>> queries = taskRelocationStatuses.stream() .map(this::newCreateOrUpdateQuery) .collect(Collectors.toList()); return dslContext .batch(queries) .execute(); }, dslContext); MonoProcessor<Map<String, Optional<Throwable>>> callerProcessor = MonoProcessor.create(); asyncAction.handle((result, error) -> { Map<String, Optional<Throwable>> resultMap = new HashMap<>(); if (error == null) { taskRelocationStatuses.forEach(p -> { resultMap.put(p.getTaskId(), Optional.empty()); statusesByTaskId.put(p.getTaskId(), p); }); callerProcessor.onNext(resultMap); } else { callerProcessor.onError(error); } return null; }); return callerProcessor; }); } }
JooqTaskRelocationResultStore implements TaskRelocationResultStore { @Override public Mono<Map<String, Optional<Throwable>>> createTaskRelocationStatuses(List<TaskRelocationStatus> taskRelocationStatuses) { return Mono.defer(() -> { CompletionStage<int[]> asyncAction = JooqUtils.executeAsync(() -> { loadToCache(findNotCached(taskRelocationStatuses), dslContext.configuration()); List<StoreQuery<JRelocationStatusRecord>> queries = taskRelocationStatuses.stream() .map(this::newCreateOrUpdateQuery) .collect(Collectors.toList()); return dslContext .batch(queries) .execute(); }, dslContext); MonoProcessor<Map<String, Optional<Throwable>>> callerProcessor = MonoProcessor.create(); asyncAction.handle((result, error) -> { Map<String, Optional<Throwable>> resultMap = new HashMap<>(); if (error == null) { taskRelocationStatuses.forEach(p -> { resultMap.put(p.getTaskId(), Optional.empty()); statusesByTaskId.put(p.getTaskId(), p); }); callerProcessor.onNext(resultMap); } else { callerProcessor.onError(error); } return null; }); return callerProcessor; }); } @Inject  JooqTaskRelocationResultStore(DSLContext dslContext, TitusRuntime titusRuntime); }
JooqTaskRelocationResultStore implements TaskRelocationResultStore { @Override public Mono<Map<String, Optional<Throwable>>> createTaskRelocationStatuses(List<TaskRelocationStatus> taskRelocationStatuses) { return Mono.defer(() -> { CompletionStage<int[]> asyncAction = JooqUtils.executeAsync(() -> { loadToCache(findNotCached(taskRelocationStatuses), dslContext.configuration()); List<StoreQuery<JRelocationStatusRecord>> queries = taskRelocationStatuses.stream() .map(this::newCreateOrUpdateQuery) .collect(Collectors.toList()); return dslContext .batch(queries) .execute(); }, dslContext); MonoProcessor<Map<String, Optional<Throwable>>> callerProcessor = MonoProcessor.create(); asyncAction.handle((result, error) -> { Map<String, Optional<Throwable>> resultMap = new HashMap<>(); if (error == null) { taskRelocationStatuses.forEach(p -> { resultMap.put(p.getTaskId(), Optional.empty()); statusesByTaskId.put(p.getTaskId(), p); }); callerProcessor.onNext(resultMap); } else { callerProcessor.onError(error); } return null; }); return callerProcessor; }); } @Inject  JooqTaskRelocationResultStore(DSLContext dslContext, TitusRuntime titusRuntime); @Override Mono<Map<String, Optional<Throwable>>> createTaskRelocationStatuses(List<TaskRelocationStatus> taskRelocationStatuses); @Override Mono<List<TaskRelocationStatus>> getTaskRelocationStatusList(String taskId); }
JooqTaskRelocationResultStore implements TaskRelocationResultStore { @Override public Mono<Map<String, Optional<Throwable>>> createTaskRelocationStatuses(List<TaskRelocationStatus> taskRelocationStatuses) { return Mono.defer(() -> { CompletionStage<int[]> asyncAction = JooqUtils.executeAsync(() -> { loadToCache(findNotCached(taskRelocationStatuses), dslContext.configuration()); List<StoreQuery<JRelocationStatusRecord>> queries = taskRelocationStatuses.stream() .map(this::newCreateOrUpdateQuery) .collect(Collectors.toList()); return dslContext .batch(queries) .execute(); }, dslContext); MonoProcessor<Map<String, Optional<Throwable>>> callerProcessor = MonoProcessor.create(); asyncAction.handle((result, error) -> { Map<String, Optional<Throwable>> resultMap = new HashMap<>(); if (error == null) { taskRelocationStatuses.forEach(p -> { resultMap.put(p.getTaskId(), Optional.empty()); statusesByTaskId.put(p.getTaskId(), p); }); callerProcessor.onNext(resultMap); } else { callerProcessor.onError(error); } return null; }); return callerProcessor; }); } @Inject  JooqTaskRelocationResultStore(DSLContext dslContext, TitusRuntime titusRuntime); @Override Mono<Map<String, Optional<Throwable>>> createTaskRelocationStatuses(List<TaskRelocationStatus> taskRelocationStatuses); @Override Mono<List<TaskRelocationStatus>> getTaskRelocationStatusList(String taskId);  }