RrdpServiceImpl implements RrdpService { int storeSnapshotObjects(List<SnapshotObject> snapshotObjects, final RpkiRepositoryValidationRun validationRun) { final AtomicInteger counter = new AtomicInteger(); List<Either<ValidationResult, Pair<String, RpkiObject>>> converted = snapshotObjects.parallelStream().map((value) -> RpkiObjectUtils.createRpkiObject(value.getUri(), value.getContent()) ).collect(Collectors.toList()); storage.writeTx0(tx -> converted.forEach((maybeRpkiObject) -> storeSnapshotObject(tx, validationRun, maybeRpkiObject, counter) )); return counter.get(); } @Autowired  RrdpServiceImpl(
            final RrdpClient rrdpClient,
            final RpkiObjects rpkiObjects,
            final RpkiRepositories rpkiRepositories,
            final Storage storage,
            final RrdpMetricsService rrdpMetrics
    ); @Override boolean storeRepository(final RpkiRepository rpkiRepository, final RpkiRepositoryValidationRun validationRun);  }
@Test public void should_parse_and_save_snapshot() throws Exception { final TrustAnchor trustAnchor = TestObjects.newTrustAnchor(); wtx0(tx -> this.getTrustAnchors().add(tx, trustAnchor)); final Ref<TrustAnchor> trustAnchorRef = rtx(tx -> this.getTrustAnchors().makeRef(tx, trustAnchor.key())); RpkiRepository rpkiRepository = wtx(tx -> this.getRpkiRepositories().register(tx, trustAnchorRef, RRDP_RIPE_NET_NOTIFICATION_XML, RpkiRepository.Type.RRDP)); Ref<RpkiRepository> rpkiRepositoryRef = rtx(tx -> this.getRpkiRepositories().makeRef(tx, rpkiRepository.key())); final RrdpRepositoryValidationRun validationRun = wtx(tx -> this.getValidationRuns().add(tx, new RrdpRepositoryValidationRun(rpkiRepositoryRef))); new RrdpParser().parseSnapshot( Objects.fileIS("rrdp/snapshot2.xml"), (snapshotInfo) -> { }, (snapshotObject) -> { subject.storeSnapshotObjects(Collections.singletonList(snapshotObject), validationRun); } ); final List<RpkiObject> objects = rtx(tx -> this.getRpkiObjects().values(tx)); assertEquals(3, objects.size()); rtx0(tx -> { final String uri1 = "rsync: assertTrue(objects.stream().anyMatch(o -> uri1.equals(getLocations(tx, o).first()))); final String uri2 = "rsync: assertTrue(objects.stream().anyMatch(o -> uri2.equals(getLocations(tx, o).first()))); final String uri3 = "rsync: assertTrue(objects.stream().anyMatch(o -> uri3.equals(getLocations(tx, o).first()))); }); }