@Test void testUpsertPackages() throws Exception { Package newPackage = mock(Package.class); when(newPackage.getId()).thenReturn("newPackageId"); Package unchangedPackage = mock(Package.class); when(unchangedPackage.getId()).thenReturn("unchangedPackageId"); Package updatedPackage = mock(Package.class); when(updatedPackage.getId()).thenReturn("updatedPackageId"); Package existingUpdatedPackage = mock(Package.class); when(existingUpdatedPackage.getId()).thenReturn("updatedPackageId"); @SuppressWarnings("unchecked") Query<Package> query = mock(Query.class); when(dataService.query(PACKAGE, Package.class)).thenReturn(query); when(query.in(eq(PackageMetadata.ID), any(Set.class))).thenReturn(query); when(query.findAll()).thenReturn(Stream.of(unchangedPackage, existingUpdatedPackage)); packagePersister.upsertPackages(Stream.of(newPackage, unchangedPackage, updatedPackage)); @SuppressWarnings("unchecked") ArgumentCaptor<Stream<Package>> addCaptor = ArgumentCaptor.forClass(Stream.class); verify(dataService).add(eq(PACKAGE), addCaptor.capture()); assertEquals(singletonList(newPackage), addCaptor.getValue().collect(toList())); @SuppressWarnings("unchecked") ArgumentCaptor<Stream<Package>> updateCaptor = ArgumentCaptor.forClass(Stream.class); verify(dataService).update(eq(PACKAGE), updateCaptor.capture()); assertEquals(singletonList(updatedPackage), updateCaptor.getValue().collect(toList())); }
@Transactional public void upsertPackages(Stream<Package> packages) { Iterator<List<Package>> partitions = Iterators.partition(packages.iterator(), BATCH_SIZE); partitions.forEachRemaining(this::upsertPackages); }
PackagePersister { @Transactional public void upsertPackages(Stream<Package> packages) { Iterator<List<Package>> partitions = Iterators.partition(packages.iterator(), BATCH_SIZE); partitions.forEachRemaining(this::upsertPackages); } }
PackagePersister { @Transactional public void upsertPackages(Stream<Package> packages) { Iterator<List<Package>> partitions = Iterators.partition(packages.iterator(), BATCH_SIZE); partitions.forEachRemaining(this::upsertPackages); } PackagePersister(DataService dataService); }
PackagePersister { @Transactional public void upsertPackages(Stream<Package> packages) { Iterator<List<Package>> partitions = Iterators.partition(packages.iterator(), BATCH_SIZE); partitions.forEachRemaining(this::upsertPackages); } PackagePersister(DataService dataService); @Transactional void upsertPackages(Stream<Package> packages); }
PackagePersister { @Transactional public void upsertPackages(Stream<Package> packages) { Iterator<List<Package>> partitions = Iterators.partition(packages.iterator(), BATCH_SIZE); partitions.forEachRemaining(this::upsertPackages); } PackagePersister(DataService dataService); @Transactional void upsertPackages(Stream<Package> packages);  }