- private CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readAllData() {
- final Set<String> allShardNames = txContextFactory.getActorContext().getConfiguration().getAllShardNames();
- final Collection<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>> futures = new ArrayList<>(allShardNames.size());
-
- for (String shardName : allShardNames) {
- futures.add(singleShardRead(shardName, YangInstanceIdentifier.EMPTY));
- }
-
- final ListenableFuture<List<Optional<NormalizedNode<?, ?>>>> listFuture = Futures.allAsList(futures);
- final ListenableFuture<Optional<NormalizedNode<?, ?>>> aggregateFuture;
-
- aggregateFuture = Futures.transform(listFuture, new Function<List<Optional<NormalizedNode<?, ?>>>, Optional<NormalizedNode<?, ?>>>() {
- @Override
- public Optional<NormalizedNode<?, ?>> apply(final List<Optional<NormalizedNode<?, ?>>> input) {
- try {
- return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input,
- txContextFactory.getActorContext().getSchemaContext(),
- txContextFactory.getActorContext().getDatastoreContext().getLogicalStoreType());
- } catch (DataValidationFailedException e) {
- throw new IllegalArgumentException("Failed to aggregate", e);
- }
- }
- });
-
- return MappingCheckedFuture.create(aggregateFuture, ReadFailedException.MAPPER);
+ private FluentFuture<Optional<NormalizedNode>> readAllData() {
+ final var actorUtils = getActorUtils();
+ return RootScatterGather.gather(actorUtils, actorUtils.getConfiguration().getAllShardNames().stream()
+ .map(shardName -> singleShardRead(shardName, YangInstanceIdentifier.empty())));