Invoice Repository

CompletableFuture.java

//  Index
@Repository
public class InvoiceByLastNameRepository extends IndexRepository<String> {
    public InvoiceByLastNameRepository() {
        super("invoice_by_lastname", "lastname", Invoice::getLastName);
    }
}


public abstract class IndexRepository<T> {
    @Inject
    private Session session;

    private final String tableName;
    private final String valueName;
    private final Function<Invoice, T> valueGetter;
    private final Function<Criteria, T> criteriumGetter;

    private PreparedStatement insertStmt;
    private PreparedStatement findStmt;
    private PreparedStatement findWithOffsetStmt;

    @PostConstruct
    public void init() {
        insertStmt = session.prepare(
            QueryBuilder.insertInto(tableName)
                .value("user_id", bindMarker())
                .value("invoice_day", bindMarker())
                .value(valueName, bindMarker())
                .value("invoice_id", bindMarker())
        );

        findStmt = session.prepare(
            QueryBuilder.select()
                .column("invoice_id")
                .from(tableName)
                .where(eq("user_id", bindMarker()))
                .and(eq("invoice_day", bindMarker()))
                .and(eq(valueName, bindMarker()))
        );

        findWithOffsetStmt = session.prepare(
            QueryBuilder.select()
                .column("invoice_id")
                .from(tableName)
                .where(eq("user_id", bindMarker()))
                .and(eq("invoice_day", bindMarker()))
                .and(eq(valueName, bindMarker()))
                .and(lte("invoice_id", bindMarker()))
        );

    }

    @Override
    public void insert(Invoice invoice) {
        T value = valueGetter.apply(invoice);
        if (value != null) {
            session.execute(
                insertStmt.bind(
                    invoice.getId(),
                    Dates.toDate(invoice.getDayFirstEvent()),
                    value,
                    invoice.getId()));
        }
    }

    @Override
    public CompletableFuture<Iterator<UUID>> find(
                                                Criteria criteria,
                                                LocalDate day,
                                                UUID invoiceIdOffset) {
        T criterium = criteriumGetter.apply(criteria);
        if (criterium == null) {
            return CompletableFuture.completedFuture(null);
        }

        BoundStatement stmt;
        if (invoiceIdOffset == null) {
            stmt = findStmt.bind(criteria.getUserId(), Dates.toDate(day), criterium);
        } else {
            stmt = findWithOffsetStmt.bind(criteria.getUserId(), Dates.toDate(day), criterium, invoiceIdOffset);
        }
        return Jdk8.completableFuture(session.executeAsync(stmt))
            .thenApply(rs -> Iterators.transform(rs.iterator(), row -> row.getUUID(0)));
    }
}

// Guava to Java 8

public static <T> CompletableFuture<T> completableFuture(ListenableFuture<T> guavaFuture) {
    CompletableFuture<T> future = new CompletableFuture<>();

    Futures.addCallback(guavaFuture, new FutureCallback<T>() {
        @Override
        public void onSuccess(V result) {
            future.complete(result);
        }

        @Override
        public void onFailure(Throwable t) {
            future.completeExceptionally(t);
        }
    });

    return future;
}


// Dates
public static Date toDate(LocalDateTime date) {
    return date == null ? null : Date.from(date.atZone(ZoneOffset.systemDefault()).toInstant());
}

public static LocalDateTime toLocalDateTime(Date date) {
    return date == null ? null : LocalDateTime.ofInstant(date.toInstant(), ZoneOffset.systemDefault());
}

public static Date toDate(LocalDate date) {
    return date == null ? null : Date.from(date.atStartOfDay().atZone(ZoneOffset.systemDefault()).toInstant());
}

public static LocalDate toLocalDate(Date date) {
    return date == null ? null : LocalDateTime.ofInstant(date.toInstant(), ZoneOffset.systemDefault()).toLocalDate();
}


// Service
@Service
public class InvoiceSearchService {

    @Inject
    private InvoiceRepository invoiceRepository;
    @Inject
    private InvoiceByDayRepository byDayRepository;
    @Inject
    private InvoiceByLastNameRepository byLastNameRepository;
    @Inject
    private InvoiceByFirstNameRepository byLastNameRepository;
    @Inject
    private InvoiceByCityRepository byCityRepository;
    @Inject
    private InvoiceByZipCodeRepository byZipCodeRepository;


public ResultPage findByCriteria(Criteria criteria) {
    return byDateInteval(criteria, (crit, day, parcelIdOffset) -> {
        CompletableFuture<Iterator<UUID>> uuidIt;
        if (crit.hasIndexedCriteria()) {
            CompletableFuture<Iterator<UUID>>[] futures = Stream.<IndexRepository> of(
                            byLastNameRepository,
                            byFirstNameRepository,
                            byCityRepository,
                            byZipCodeRepository)
                    .map(repo -> repo.find(crit, day, offset))
                    .toArray(CompletableFuture[]::new);

            uuidIt = CompletableFuture.allOf(futures).thenApply(v ->
                    Iterators.intersection(TimeUUIDComparator.desc,
                        futures.stream()
                            .map(CompletableFuture::join)
                            .filter(Objects::nonNull)
                            .collect(toList()))
            );

        } else {
            uuidIt = byDayRepository.find(crit.getUserId(), day, parcelIdOffset);
        }

        return uuidIt;
    });
}


/**
 * Comparateur de TimeUUID équivalent à celui de Cassandra:
 * @see org.apache.cassandra.db.marshal.TimeUUIDType#compare()
 */
public enum TimeUUIDComparator implements Comparator<UUID> {
    desc {
        @Override
        public int compare(UUID o1, UUID o2) {
            long delta = o2.timestamp() - o1.timestamp();
            if (delta != 0) {
                return Ints.saturatedCast(delta);
            }
            return o2.compareTo(o1);
        }

    };
}


@FunctionalInterface
private static interface DayQuery {
    CompletableFuture<Iterator<UUID>> find(Criteria criteria, LocalDate day, UUID invoiceIdOffset);
}

private ResultPage byDateInteval(Criteria criteria, DayQuery dayQuery) {
    int limit = criteria.getLimit();

    LocalDate dayOffset = criteria.getDayOffset();
    UUID invoiceIdOffset = criteria.getInvoiceIdOffset();

    List<Invoice> resultList = new ArrayList<>(limit);
    LocalDate day = criteria.getLastDay();
    do {
        Iterator<UUID> uuidIt = dayQuery.find(criteria, day, invoiceIdOffset).join();
        limit -= loadInvoices(resultList, uuidIt, criteria, limit);

        if (uuidIt.hasNext()) {
            return new ResultPage(resultList, day, uuidIt.next());
        }
        day = day.minusDays(1);
        invoiceIdOffset = null;

    } while (!day.isBefore(criteria.getFirstDay()));

    return new ResultPage(resultList);
}



private int loadInvoices(List<Invoice> resultList,
                         Iterator<UUID> uuidIt,
                         Criteria criteria,
                         int limit) {
    int remains = limit;
    do {
        List<CompletableFuture<Invoice>> futureList = new ArrayList<>(limit);

        for (int i = 0; i < remains && uuidIt.hasNext(); ++i) {
            futureList.add(invoiceRepository.findOne(uuidIt.next()));
        }

        futureList.stream()
            .map(CompletableFuture::join)
            .filter(criteria)
            .forEach(resultList::add);

        remains = limit - resultList.size();

    } while (remains > 0 && uuidIt.hasNext());
    return resultList;
}