HazelcastMessageSearcher.java

package net.morimekta.providence.storage.hazelcast;

import com.hazelcast.core.IMap;
import com.hazelcast.query.Predicate;
import net.morimekta.providence.PMessage;
import net.morimekta.providence.PMessageBuilder;
import net.morimekta.providence.storage.MessageSearcher;

import javax.annotation.Nonnull;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class HazelcastMessageSearcher<Q, Key, M extends PMessage<M>, B extends PMessageBuilder<M>>
        implements MessageSearcher<Q, M> {
    private final IMap<Key, B> hazelcastMap;
    private final Function<Q, Predicate<Key, B>> makePredicate;

    public HazelcastMessageSearcher(Function<Q, Predicate<Key, B>> makePredicate,
                                    IMap<Key, B> hazelcastMap) {
        this.makePredicate = makePredicate;
        this.hazelcastMap = hazelcastMap;
    }

    @Nonnull
    @Override
    public List<M> search(@Nonnull Q query) {
        return stream(query).collect(Collectors.toList());
    }

    @Override
    public Stream<M> stream(@Nonnull Q query) {
        return hazelcastMap.values(makePredicate.apply(query))
                           .stream()
                           .map(PMessageBuilder::build);
    }
}