import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testQuery() {
    IgniteCache<String, String> cache = ignite().getOrCreateCache("testcache1");
    Set<String> keys = new HashSet<>();

    for (int i = 0; i < 100; i++) {
        cache.put("k" + i, "v" + i);
        keys.add("k" + i);
    }

    Query<Entry<String, String>> query = new ScanQuery<String, String>(new IgniteBiPredicate<String, String>() {
        private static final long serialVersionUID = 1L;

        @Override
        public boolean apply(String key, String value) {
            return Integer.parseInt(key.replace("k", "")) >= 50;
        }
    });

    List results = template.requestBodyAndHeader("ignite:cache:testcache1?operation=QUERY", keys, IgniteConstants.IGNITE_CACHE_QUERY, query, List.class);
    assert_().that(results.size()).isEqualTo(50);
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @param expect Expected events number.
 * @return Event receive latch.
 */
private CountDownLatch subscribeToPutEvents(int expect) {
    Ignite ignite = grid();

    // Listen to cache PUT events and expect as many as messages as test data items
    final CountDownLatch latch = new CountDownLatch(expect);

    @SuppressWarnings("serial") IgniteBiPredicate<UUID, CacheEvent> cb = new IgniteBiPredicate<UUID, CacheEvent>() {
        @Override public boolean apply(UUID uuid, CacheEvent evt) {
            latch.countDown();
            return true;
        }
    };

    ignite.events(ignite.cluster().forCacheNodes(DEFAULT_CACHE_NAME)).remoteListen(cb, null, EVT_CACHE_OBJECT_PUT);
    return latch;
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * Tests that Scan queries are executed in dedicated pool
 * @throws Exception If failed.
 */
public void testScanQueryUsesDedicatedThreadPool() throws Exception {
    try (Ignite client = startGrid("client")) {
        IgniteCache<Integer, Integer> cache = client.cache(CACHE_NAME);

        cache.put(0, 0);

        QueryCursor<Cache.Entry<Object, Object>> cursor = cache.query(
            new ScanQuery<>(new IgniteBiPredicate<Object, Object>() {
                @Override public boolean apply(Object o, Object o2) {
                    return F.eq(GridIoManager.currentPolicy(), GridIoPolicy.QUERY_POOL);
                }
            }));

        assertEquals(1, cursor.getAll().size());

        cursor.close();
    }
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * Example for scan query based on a predicate using binary objects.
 */
private static void scanQuery() {
    IgniteCache<BinaryObject, BinaryObject> cache = Ignition.ignite()
        .cache(PERSON_CACHE).withKeepBinary();

    ScanQuery<BinaryObject, BinaryObject> scan = new ScanQuery<>(
        new IgniteBiPredicate<BinaryObject, BinaryObject>() {
            @Override public boolean apply(BinaryObject key, BinaryObject person) {
                return person.<Double>field("salary") <= 1000;
            }
        }
    );

    // Execute queries for salary ranges.
    print("People with salaries between 0 and 1000 (queried with SCAN query): ", cache.query(scan).getAll());
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testResponseMessageOnRequestUnmarshallingFailed() throws Exception {
    readCnt.set(Integer.MAX_VALUE);

    try {
        jcache().query(new ScanQuery<>(new IgniteBiPredicate<TestKey, String>() {
            @Override public boolean apply(TestKey key, String val) {
                return false;
            }

            private void readObject(ObjectInputStream is) throws IOException {
                throw new IOException();
            }

            private void writeObject(ObjectOutputStream os) throws IOException {
                // No-op.
            }
        })).getAll();

        fail();
    }
    catch (Exception ignored) {
        // No-op.
    }
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * Subscribe to cache put events.
 */
private CountDownLatch subscribeToPutEvents(int expect) {
    Ignite ignite = grid();

    // Listen to cache PUT events and expect as many as messages as test data items
    final CountDownLatch latch = new CountDownLatch(expect);
    @SuppressWarnings("serial") IgniteBiPredicate<UUID, CacheEvent> callback =
        new IgniteBiPredicate<UUID, CacheEvent>() {
        @Override public boolean apply(UUID uuid, CacheEvent evt) {
            latch.countDown();

            return true;
        }
    };

    remoteLsnr = ignite.events(ignite.cluster().forCacheNodes(DEFAULT_CACHE_NAME))
        .remoteListen(callback, null, EVT_CACHE_OBJECT_PUT);

    return latch;
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @throws Exception In case of error.
 */
public void testDeployScanPredicate() throws Exception {
    startGrids(4);

    awaitPartitionMapExchange();

    try {
        IgniteCache<Object, Object> cache = grid(3).cache(DEFAULT_CACHE_NAME);

        // It is important that there are no too many keys.
        for (int i = 0; i < 1; i++)
            cache.put(i, i);

        Class predCls = grid(3).configuration().getClassLoader().loadClass(TEST_PREDICATE);

        IgniteBiPredicate<Object, Object> pred = (IgniteBiPredicate<Object, Object>)predCls.newInstance();

        List<Cache.Entry<Object, Object>> all = cache.query(new ScanQuery<>(pred)).getAll();

        assertEquals(1, all.size());
    }
    finally {
        stopAllGrids();
    }
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @param crd Exchange coordinator.
 * @param topVer Exchange topology version.
 */
private void blockExchangeFinish(Ignite crd, long topVer) {
    final AffinityTopologyVersion topVer0 = new AffinityTopologyVersion(topVer);

    TestRecordingCommunicationSpi.spi(crd).blockMessages(new IgniteBiPredicate<ClusterNode, Message>() {
        @Override public boolean apply(ClusterNode node, Message msg) {
            if (msg instanceof GridDhtPartitionsFullMessage) {
                GridDhtPartitionsFullMessage msg0 = (GridDhtPartitionsFullMessage)msg;

                return msg0.exchangeId() != null && msg0.exchangeId().topologyVersion().equals(topVer0);
            }

            return false;
        }
    });
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @param expect Expected count.
 * @return Latch to be counted down in listener.
 */
private CountDownLatch subscribeToPutEvents(int expect) {
    Ignite ignite = grid();

    // Listen to cache PUT events and expect as many as messages as test data items
    final CountDownLatch latch = new CountDownLatch(expect);

    IgniteBiPredicate<UUID, CacheEvent> cb = new IgniteBiPredicate<UUID, CacheEvent>() {
        @Override public boolean apply(UUID uuid, CacheEvent evt) {
            latch.countDown();

            return true;
        }
    };

    remoteLsnr = ignite.events(ignite.cluster().forCacheNodes(DEFAULT_CACHE_NAME))
        .remoteListen(cb, null, EVT_CACHE_OBJECT_PUT);

    return latch;
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @param cache Cache.
 */
private void cacheQuery(IgniteCache cache) {
    int keys = 100;

    Map<Integer, Integer> data = generateDataMap(keys);

    cache.putAll(data);

    ScanQuery<Integer, Integer> qry = new ScanQuery<>(new IgniteBiPredicate<Integer, Integer>() {
        @Override public boolean apply(Integer key, Integer val) {
            return key % 2 == 0;
        }
    });

    List<Cache.Entry<Integer, Integer>> all = cache.query(qry).getAll();

    assertEquals(all.size(), data.size() / 2);

    for (Cache.Entry<Integer, Integer> entry : all) {
        assertEquals(0, entry.getKey() % 2);
        assertEquals(entry.getValue(), data.get(entry.getKey()));
    }

    tearDown(cache);
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testNullTopicWithDeployment() throws Exception {
    Class<?> cls = getExternalClassLoader().loadClass(LSNR_CLS_NAME);

    grid(0).message().remoteListen(null, (IgniteBiPredicate<UUID, Object>)cls.newInstance());

    send();

    boolean s = GridTestUtils.waitForCondition(new PA() {
        @Override public boolean apply() {
            return checkDeployedListeners(GRID_CNT);
        }
    }, 2000);

    assertTrue(s);
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public <T extends Event> IgniteFuture<UUID> remoteListenAsync(int bufSize, long interval,
    boolean autoUnsubscribe, @Nullable IgniteBiPredicate<UUID, T> locLsnr, @Nullable IgnitePredicate<T> rmtFilter,
    @Nullable int... types) throws IgniteException {
    A.ensure(bufSize > 0, "bufSize > 0");
    A.ensure(interval >= 0, "interval >= 0");

    guard();

    try {
        GridEventConsumeHandler hnd = new GridEventConsumeHandler((IgniteBiPredicate<UUID, Event>)locLsnr,
            (IgnitePredicate<Event>)rmtFilter, types);

        return new IgniteFutureImpl<>(ctx.continuous().startRoutine(
            hnd,
            false,
            bufSize,
            interval,
            autoUnsubscribe,
            prj.predicate()));
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @param cctx Context.
 * @param type Query type.
 * @param filter Scan filter.
 * @param part Partition.
 * @param keepBinary Keep binary flag.
 */
public GridCacheQueryAdapter(GridCacheContext<?, ?> cctx,
    GridCacheQueryType type,
    @Nullable IgniteBiPredicate<Object, Object> filter,
    @Nullable IgniteClosure<Map.Entry, Object> transform,
    @Nullable Integer part,
    boolean keepBinary) {
    assert cctx != null;
    assert type != null;
    assert part == null || part >= 0;

    this.cctx = cctx;
    this.type = type;
    this.filter = filter;
    this.transform = transform;
    this.part = part;
    this.keepBinary = keepBinary;

    log = cctx.logger(getClass());

    metrics = new GridCacheQueryMetricsAdapter();

    this.incMeta = false;
    this.clsName = null;
    this.clause = null;
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testNonNullTopicWithDeployment() throws Exception {
    ClassLoader ldr = getExternalClassLoader();

    Class<?> topicCls = ldr.loadClass(TOPIC_CLS_NAME);
    Class<?> lsnrCls = ldr.loadClass(LSNR_CLS_NAME);

    Object topic = topicCls.newInstance();

    grid(0).message().remoteListen(topic, (IgniteBiPredicate<UUID, Object>)lsnrCls.newInstance());

    send(topic);

    boolean s = GridTestUtils.waitForCondition(new PA() {
        @Override public boolean apply() {
            return checkDeployedListeners(GRID_CNT);
        }
    }, 2000);

    assertTrue(s);
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public UUID remoteListen(@Nullable Object topic, IgniteBiPredicate<UUID, ?> p) {
    A.notNull(p, "p");

    guard();

    try {
        GridContinuousHandler hnd = new GridMessageListenHandler(topic, (IgniteBiPredicate<UUID, Object>)p);

        return saveOrGet(ctx.continuous().startRoutine(hnd,
            false,
            1,
            0,
            false,
            prj.predicate()));
    }
    catch (IgniteCheckedException e) {
        throw U.convertException(e);
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public IgniteFuture<UUID> remoteListenAsync(@Nullable Object topic,
    IgniteBiPredicate<UUID, ?> p) throws IgniteException {
    A.notNull(p, "p");

    guard();

    try {
        GridContinuousHandler hnd = new GridMessageListenHandler(topic, (IgniteBiPredicate<UUID, Object>)p);

        return new IgniteFutureImpl<>(ctx.continuous().startRoutine(hnd,
            false,
            1,
            0,
            false,
            prj.predicate()));
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * Combine this computation chain with other computation chain in the following way:
 * 1. perform this calculations chain and get result r.
 * 2. while 'cond(r)' is true, r = otherChain(r, context)
 * 3. return r.
 *
 * @param cond Condition checking if 'while' loop should continue.
 * @param otherChain Chain to be combined with this chain.
 * @return Combination of this chain and otherChain.
 */
default ComputationsChain<L, K, V, I, O> thenWhile(IgniteBiPredicate<O, L> cond,
                                                   ComputationsChain<L, K, V, O, O> otherChain) {
    ComputationsChain<L, K, V, I, O> me = this;
    return (input, context) -> {
        O res = me.process(input, context);

        while (cond.apply(res, context.localContext()))
            res = otherChain.process(res, context);

        return res;
    };
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * @param ignite     Node.
 * @param joinEvtCnt Expected events number.
 * @return Events latch.
 */
protected CountDownLatch expectJoinEvents(Ignite ignite, int joinEvtCnt) {
    final CountDownLatch latch = new CountDownLatch(joinEvtCnt);

    ignite.events().remoteListen(new IgniteBiPredicate<UUID, Event>() {
        @Override
        public boolean apply(UUID uuid, Event evt) {
            latch.countDown();
            return true;
        }
    }, null, EventType.EVT_NODE_JOINED);

    return latch;
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
@Override
protected JndiRegistry createRegistry() throws Exception {
    JndiRegistry answer = super.createRegistry();

    ScanQuery<Integer, Person> scanQuery1 = new ScanQuery<>(new IgniteBiPredicate<Integer, Person>() {
        private static final long serialVersionUID = 1L;

        @Override
        public boolean apply(Integer key, Person person) {
            return person.getId() > 50;
        }
    });

    CacheEntryEventSerializableFilter<Integer, Person> remoteFilter = new CacheEntryEventSerializableFilter<Integer, IgniteCacheContinuousQueryTest.Person>() {
        private static final long serialVersionUID = 5624973479995548199L;

        @Override
        public boolean evaluate(CacheEntryEvent<? extends Integer, ? extends Person> event) throws CacheEntryListenerException {
            return event.getValue().getId() > 150;
        }
    };

    answer.bind("query1", scanQuery1);
    answer.bind("remoteFilter1", remoteFilter);

    return answer;
}
 

import org.apache.ignite.lang.IgniteBiPredicate; //导入依赖的package包/类
/**
 * Combine this computation chain with other computation chain in the following way:
 * 1. perform this calculations chain and get result r.
 * 2. while 'cond(r)' is true, r = otherChain(r, context)
 * 3. return r.
 *
 * @param cond Condition checking if 'while' loop should continue.
 * @param otherChain Chain to be combined with this chain.
 * @return Combination of this chain and otherChain.
 */
default ComputationsChain<L, K, V, I, O> thenWhile(IgniteBiPredicate<O, L> cond,
    ComputationsChain<L, K, V, O, O> otherChain) {
    ComputationsChain<L, K, V, I, O> me = this;
    return (input, context) -> {
        O res = me.process(input, context);

        while (cond.apply(res, context.localContext()))
            res = otherChain.process(res, context);

        return res;
    };
}