package userlist.repro;

import java.util.Collections;
import java.util.List;
import javax.cache.Cache;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;

/**
 *
 */
public class ScanQueryFilter {
    /** */
    public static void main(String[] args) {
        String ORG_CACHE = "org_cache_remote";

        final TcpDiscoveryVmIpFinder finder = new TcpDiscoveryVmIpFinder();
        finder.setAddresses(Collections.singletonList("127.0.0.1"));

        Ignite ignite = Ignition.start(
            new IgniteConfiguration()
                .setGridName("client")
                .setClientMode(true)
                .setPeerClassLoadingEnabled(true)
                .setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(finder)));

        CacheConfiguration<Long, Organization> orgCacheCfg = new CacheConfiguration<>(ORG_CACHE);
        orgCacheCfg.setIndexedTypes(Long.class, Organization.class);
        ignite.destroyCache(ORG_CACHE);
        IgniteCache<Long, Organization> cache = ignite.createCache(orgCacheCfg);
        cache.put(1L, new Organization(1L, "org1", true, "jurong east", "1111"));
        cache.put(2L, new Organization(2L, "org2", false, "orchard", "2222"));
        cache.put(3L, new Organization(3L, "org3", true, "jurong west", "3333"));
        cache.put(4L, new Organization(4L, "org4", false, "woodlands", "4444"));
        cache.put(5L, new Organization(5L, "org5", false, "changi", "5555"));
        // cache.put(6L, new Organization(6L, "org6", true, "jurong island", "6666"));

        IgniteCache<Long, BinaryObject> binaryCache = cache.withKeepBinary();

        List<Cache.Entry<Long, BinaryObject>> result;

        System.out.println("Scan by address");
        ScanQuery<Long, BinaryObject> scanAddress = new ScanQuery<>(
            new IgniteBiPredicate<Long, BinaryObject>() {
                @Override
                public boolean apply(Long aLong, BinaryObject binaryObject) {
                    // first time filter by jurong, got two entries, org1 and org3
                    // second time filter by changi, got two entries, org1 and org3
                    // third time filter by changi as well, uncomment org6, got three entries, org1, org3 and org6
                    return binaryObject.<String>field("address").startsWith("jurong");
                }
            }
        );
        result = binaryCache.query(scanAddress).getAll();
        System.out.println("result: " + result.size());
        for (Cache.Entry<Long, BinaryObject> entry : result) {
            System.out.println(entry.getValue().deserialize().toString());
        }

        ignite.close();
    }

    /**
     *
     */
    private static class MyObject {
        /** */
        String name;

        /** */
        public MyObject(String name) {
            this.name = name;
        }
    }

    private static class Organization {
        long id;

        @QuerySqlField(index = true)
        String address;

        public Organization(long l, String org1, boolean b, String s, String s1) {
            id = l;
            address = s;
        }

        @Override public String toString() {
            return "Organization{" +
                "id=" + id +
                ", address='" + address + '\'' +
                '}';
        }
    }
}