import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * Filter for distributed matrix keys.
 *
 * @param matrixUuid Matrix uuid.
 */
private static <K> IgnitePredicate<K> sparseKeyFilter(UUID matrixUuid) {
    return key -> {
        if (key instanceof DataStructureCacheKey)
            return ((DataStructureCacheKey)key).dataStructureId().equals(matrixUuid);
        else if (key instanceof IgniteBiTuple)
            return ((IgniteBiTuple<Integer, UUID>)key).get2().equals(matrixUuid);
        else if (key instanceof MatrixBlockKey)
            return ((MatrixBlockKey)key).dataStructureId().equals(matrixUuid);
        else if (key instanceof RowColMatrixKey)
            return ((RowColMatrixKey)key).dataStructureId().equals(matrixUuid);
        else if (key instanceof VectorBlockKey)
            return ((VectorBlockKey)key).dataStructureId().equals(matrixUuid);
        else
            throw new UnsupportedOperationException();
    };
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @param clause Query clause.
 * @param loc local query flag.
 * @param keepBinary keep binary flag.
 */
private void checkTextQuery(String clause, boolean loc, boolean keepBinary) throws Exception {
    final IgniteEx ignite = grid(0);

    if (F.isEmpty(clause))
        clause = "1*";

    // 1. Populate cache with data, calculating expected count in parallel.
    Set<Integer> exp = populateCache(ignite, loc, MAX_ITEM_COUNT, new IgnitePredicate<Integer>() {
        @Override
        public boolean apply(Integer x) {
            return String.valueOf(x).startsWith("1");
        }
    });

    // 2. Validate results.
    TextQuery qry = new TextQuery<>(Person.class, clause).setLocal(loc);

    validateQueryResults(ignite, qry, exp, keepBinary);

    clearCache(ignite);
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @param nodePred Node predicate.
 * @param numServices Number of configurations to generate.
 * @return Generated services configurations.
 */
private List<ServiceConfiguration> getConfigs(IgnitePredicate<ClusterNode> nodePred, int numServices) {
    List<ServiceConfiguration> cfgs = new ArrayList<>(numServices);

    for (int i = 0; i < numServices; i++) {
        String name = "testService-" + i;

        ServiceConfiguration cfg = new ServiceConfiguration();

        cfg.setName(name);
        cfg.setTotalCount(1);
        cfg.setMaxPerNodeCount(1);
        cfg.setService(new DummyService());
        cfg.setNodeFilter(nodePred);

        cfgs.add(cfg);
    }
    return cfgs;
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
@SuppressWarnings("serial")
public void testCacheConfigFilter() throws Exception {
    TestSuite dfltSuite = new ConfigVariationsTestSuiteBuilder("testSuite", NoopTest.class)
        .withBasicCacheParams()
        .build();

    final AtomicInteger cnt = new AtomicInteger();

    TestSuite filteredSuite = new ConfigVariationsTestSuiteBuilder("testSuite", NoopTest.class)
        .withBasicCacheParams()
        .withCacheConfigFilters(new IgnitePredicate<CacheConfiguration>() {
            @Override public boolean apply(CacheConfiguration configuration) {
                return cnt.getAndIncrement() % 2 == 0;
            }
        })
        .build();

    assertEquals(dfltSuite.countTestCases() / 2, filteredSuite.countTestCases());
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testResolveLocalAddresses() throws Exception {
    InetAddress inetAddress = InetAddress.getByName("0.0.0.0");

    IgniteBiTuple<Collection<String>, Collection<String>> addrs = U.resolveLocalAddresses(inetAddress);

    Collection<String> hostNames = addrs.get2();

    assertFalse(hostNames.contains(null));
    assertFalse(hostNames.contains(""));
    assertFalse(hostNames.contains("127.0.0.1"));

    assertFalse(F.exist(hostNames, new IgnitePredicate<String>() {
        @Override public boolean apply(String hostName) {
            return hostName.contains("localhost") || hostName.contains("0:0:0:0:0:0:0:1");
        }
    }));
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @param client Client.
 * @param disconnectLatch Disconnect event latch.
 * @param reconnectLatch Reconnect event latch.
 */
private void addListener(Ignite client, final CountDownLatch disconnectLatch, final CountDownLatch reconnectLatch) {
    client.events().localListen(new IgnitePredicate<Event>() {
        @Override public boolean apply(Event evt) {
            if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
                info("Disconnected: " + evt);

                disconnectLatch.countDown();
            }
            else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
                info("Reconnected: " + evt);

                reconnectLatch.countDown();
            }

            return true;
        }
    }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED);
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public <T extends Event> List<T> remoteQuery(IgnitePredicate<T> p, long timeout, @Nullable int... types) {
    A.notNull(p, "p");

    guard();

    try {
        return saveOrGet(ctx.event().remoteEventsAsync(compoundPredicate(p, types), prj.nodes(), timeout));
    }
    catch (IgniteCheckedException e) {
        throw U.convertException(e);
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的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.IgnitePredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public <T extends Event> Collection<T> localQuery(IgnitePredicate<T> p, @Nullable int... types) {
    A.notNull(p, "p");

    guard();

    try {
        return ctx.event().localEvents(compoundPredicate(p, types));
    }
    catch (IgniteCheckedException e) {
        throw U.convertException(e);
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @param p Predicate.
 * @param types Event types.
 * @return Compound predicate.
 */
private static <T extends Event> IgnitePredicate<T> compoundPredicate(final IgnitePredicate<T> p,
    @Nullable final int... types) {

    return F.isEmpty(types) ? p :
        new IgnitePredicate<T>() {
            @Override public boolean apply(T t) {
                for (int type : types) {
                    if (type == t.type())
                        return p.apply(t);
                }

                return false;
            }
        };
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public void start() throws IgniteCheckedException {
    Map<IgnitePredicate<? extends Event>, int[]> evtLsnrs = ctx.config().getLocalEventListeners();

    if (evtLsnrs != null) {
        for (IgnitePredicate<? extends Event> lsnr : evtLsnrs.keySet())
            addLocalEventListener(lsnr, evtLsnrs.get(lsnr));
    }

    startSpi();

    msgLsnr = new RequestListener();

    ctx.io().addMessageListener(TOPIC_EVENT, msgLsnr);

    if (log.isDebugEnabled())
        log.debug(startInfo());
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @return Cache API test suite.
 * @throws Exception If failed.
 */
public static TestSuite suite() throws Exception {
    return new ConfigVariationsTestSuiteBuilder(
        "Cache Read Through Variations Test",
        IgniteCacheReadThroughEvictionSelfTest.class)
        .withBasicCacheParams()
        .withIgniteConfigFilters(new IgnitePredicate<IgniteConfiguration>() {
            /** {@inheritDoc} */
            @SuppressWarnings("RedundantIfStatement")
            @Override public boolean apply(IgniteConfiguration cfg) {
                if (cfg.getMarshaller() != null && !(cfg.getMarshaller() instanceof BinaryMarshaller))
                    return false;

                if (cfg.isPeerClassLoadingEnabled())
                    return false;

                return true;
            }
        })
        .skipWaitPartitionMapExchange()
        .gridsCount(4).backups(1)
        .testedNodesCount(2).withClients()
        .build();
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @param cnt Number of grids.
 * @param startIdx Start node index.
 * @param list List of started grids.
 * @throws Exception If failed.
 */
private void startGrids(int cnt, int startIdx, Collection<Ignite> list) throws Exception {
    for (int i = 0; i < cnt; i++) {
        final Ignite g = startGrid(startIdx++);

        if (DEBUG)
            g.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    info("\n>>> Preload event [igniteInstanceName=" + g.name() + ", evt=" + evt + ']');

                    return true;
                }
            }, EVTS_CACHE_REBALANCE);

        list.add(g);
    }
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
protected void addDisconnectListener(
    final CountDownLatch disconnectedLatch,
    final CountDownLatch reconnectedLatch
) {
    grid(nodeClient).events().localListen(new IgnitePredicate<Event>() {
        @Override public boolean apply(Event event) {
            switch (event.type()) {
                case EventType.EVT_CLIENT_NODE_DISCONNECTED:
                    info("Client disconnected");

                    disconnectedLatch.countDown();

                    break;
                case EventType.EVT_CLIENT_NODE_RECONNECTED:
                    info("Client reconnected");

                    reconnectedLatch.countDown();
            }

            return true;
        }
    }, EventType.EVT_CLIENT_NODE_DISCONNECTED, EventType.EVT_CLIENT_NODE_RECONNECTED);
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * Applies all given predicates to all elements in given input collection and returns
 * {@code true} if all predicates evaluate to {@code true} for at least one element. Returns
 * {@code false} otherwise. Processing will short-circuit after first element evaluates to
 * {@code true} for all predicates.
 *
 * @param c Input collection.
 * @param p Optional set of checking predicates. If none provided - {@code true} is returned.
 * @param <V> Type of the collection element.
 * @return Returns {@code true} if all given predicates evaluate to {@code true} for
 *      at least one element. Returns {@code false} otherwise.
 */
@Deprecated
public static <V> boolean forAny(Iterable<? extends V> c, @Nullable IgnitePredicate<? super V>... p) {
    A.notNull(c, "c");

    if (!c.iterator().hasNext())
        return false;
    else if (isEmpty(p))
        return true;
    else if (isAlwaysFalse(p))
        return false;
    else if (isAlwaysTrue(p))
        return true;
    else {
        for (V v : c)
            if (isAll(v, p))
                return true;

        return false;
    }
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
    boolean b = in.readBoolean();

    if (b) {
        prjPredBytes = U.readByteArray(in);
        clsName = U.readString(in);
        depInfo = (GridDeploymentInfo)in.readObject();
    }
    else
        prjPred = (IgnitePredicate<ClusterNode>)in.readObject();

    hnd = (GridContinuousHandler)in.readObject();
    bufSize = in.readInt();
    interval = in.readLong();
    autoUnsubscribe = in.readBoolean();
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public ClusterGroup forPredicate(IgnitePredicate<ClusterNode> p) {
    A.notNull(p, "p");

    guard();

    try {
        if (p != null)
            ctx.resource().injectGeneric(p);

        return new ClusterGroupAdapter(ctx, subjId, this.p != null ? F.and(p, this.p) : p);
    }
    catch (IgniteCheckedException e) {
        throw U.convertException(e);
    }
    finally {
        unguard();
    }
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
    IgniteConfiguration cfg = super.getConfiguration(gridName);

    Map<IgnitePredicate<? extends Event>, int[]> map = new HashMap<>();

    // To make partitions exchanges longer.
    map.put(new P1<Event>() {
        @Override public boolean apply(Event evt) {
            try {
                Thread.sleep(10);
            }
            catch (InterruptedException e) {
                // No op.
            }

            return false;
        }
    }, new int[]{EVT_NODE_JOINED, EVT_NODE_FAILED, EVT_NODE_LEFT, EVT_DISCOVERY_CUSTOM_EVT});

    cfg.setLocalEventListeners(map);

    return cfg;
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * Starts statically configured caches received from remote nodes during exchange.
 *
 * @param nodeId Joining node ID.
 * @param exchTopVer Current exchange version.
 * @return Started caches descriptors.
 * @throws IgniteCheckedException If failed.
 */
public Collection<DynamicCacheDescriptor> startReceivedCaches(UUID nodeId, AffinityTopologyVersion exchTopVer)
    throws IgniteCheckedException {
    List<DynamicCacheDescriptor> started = cachesInfo.cachesReceivedFromJoin(nodeId);

    for (DynamicCacheDescriptor desc : started) {
        IgnitePredicate<ClusterNode> filter = desc.groupDescriptor().config().getNodeFilter();

        if (CU.affinityNode(ctx.discovery().localNode(), filter)) {
            prepareCacheStart(
                desc.cacheConfiguration(),
                desc,
                null,
                exchTopVer,
                false);
        }
    }

    return started;
}
 

import org.apache.ignite.lang.IgnitePredicate; //导入依赖的package包/类
/**
 * @param evts Events.
 * @param g Grid.
 * @param parts Parts.
 */
private void checkPartitionUnloadEvents(Collection<Event> evts, Ignite g,
    Collection<GridDhtLocalPartition> parts) {
    assertEquals(parts.size(), evts.size());

    for (Event evt : evts) {
        CacheRebalancingEvent unloadEvt = (CacheRebalancingEvent)evt;

        final int part = unloadEvt.partition();

        assertNotNull("Unexpected partition: " + part, F.find(parts, null,
            new IgnitePredicate<GridDhtLocalPartition>() {
                @Override
                public boolean apply(GridDhtLocalPartition e) {
                    return e.id() == part;
                }
            }));

        assertEquals(g.cache(DEFAULT_CACHE_NAME).getName(), unloadEvt.cacheName());
        assertEquals(g.cluster().localNode().id(), unloadEvt.node().id());
    }
}