import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Stream;

import com.vaadin.data.provider.AbstractBackEndDataProvider;
import com.vaadin.data.provider.Query;

public class StreamDataProvider<T> extends AbstractBackEndDataProvider<T, Predicate<T>> {

    private final Supplier<Stream<T>> streamSupplier;

    public StreamDataProvider(Supplier<Stream<T>> streamSupplier) {
        this.streamSupplier = streamSupplier;
    }

    @Override
    protected Stream<T> fetchFromBackEnd(Query<T, Predicate<T>> query) {
        Stream<T> stream = this.streamSupplier.get();
        if (query.getFilter().isPresent()) {
            stream = stream.filter(query.getFilter().get());
        }
        return stream.skip(query.getOffset()).limit(query.getLimit());
    }

    @Override
    protected int sizeInBackEnd(Query<T, Predicate<T>> query) {
        Stream<T> stream = this.streamSupplier.get();
        if (query.getFilter().isPresent()) {
            stream = stream.filter(query.getFilter().get());
        }
        long count = stream.skip(query.getOffset()).limit(query.getLimit()).count();
        return Math.toIntExact(count);
    }

}