ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [03/56] [abbrv] incubator-ignite git commit: #ignite-128: Moved classes in other packages.
Date Tue, 10 Feb 2015 15:57:45 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/756034f3/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractConnectivitySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractConnectivitySelfTest.java
b/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractConnectivitySelfTest.java
new file mode 100644
index 0000000..a7b7d8b
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractConnectivitySelfTest.java
@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.client.integration;
+
+import org.apache.ignite.*;
+import org.apache.ignite.client.*;
+import org.apache.ignite.internal.client.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.testframework.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.jetbrains.annotations.*;
+
+import java.net.*;
+import java.util.*;
+
+/**
+ * Tests the REST client-server connectivity with various configurations.
+ */
+public abstract class ClientAbstractConnectivitySelfTest extends GridCommonAbstractTest {
+    /** */
+    private static final String WILDCARD_IP = "0.0.0.0";
+
+    /** */
+    private static final String LOOPBACK_IP = "127.0.0.1";
+
+    /**
+     * @return IP addresses.
+     * @throws Exception If failed.
+     */
+    private static IgniteBiTuple<Collection<String>, Collection<String>>
getAllIps() throws Exception {
+        return U.resolveLocalAddresses(InetAddress.getByName("0.0.0.0"));
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        GridClientFactory.stopAll();
+
+        G.stopAll(true);
+    }
+
+    /**
+     * Starts a REST-enabled node.
+     *
+     * @param name Node name.
+     * @param addr REST address (default if null).
+     * @param port REST port (default if null).
+     * @return Started node.
+     * @throws Exception If case of configuration or startup error.
+     */
+    protected abstract Ignite startRestNode(String name, @Nullable String addr, @Nullable
Integer port) throws Exception;
+
+    /**
+     * @return Default REST port.
+     */
+    protected abstract int defaultRestPort();
+
+    /**
+     * @return REST address attribute name.
+     */
+    protected abstract String restAddressAttributeName();
+
+    /**
+     * @return REST host name attribute name.
+     */
+    protected abstract String restHostNameAttributeName();
+
+    /**
+     * @return REST port attribute name.
+     */
+    protected abstract String restPortAttributeName();
+
+    /**
+     * @return REST protocol.
+     */
+    protected abstract GridClientProtocol protocol();
+
+    /**
+     * Starts a REST client.
+     *
+     * @param addr REST server address.
+     * @param port REST server port.
+     * @return A successfully started REST client.
+     * @throws GridClientException If failed to start REST client.
+     */
+    protected GridClient startClient(String addr, int port) throws GridClientException {
+        GridClientConfiguration cliCfg = new GridClientConfiguration();
+        cliCfg.setServers(Collections.singleton(addr + ":" + port));
+        cliCfg.setProtocol(protocol());
+
+        return GridClientFactory.start(cliCfg);
+    }
+
+    /**
+     * Tests correct behavior in case of 1 REST-enabled node
+     * with default settings.
+     *
+     * @throws Exception If failed.
+     */
+    public void testOneNodeDefaultHostAndPort() throws Exception {
+        startRestNode("grid1", null, null);
+
+        checkConnectivityByIp(LOOPBACK_IP, getAllIps());
+
+        String extIp = F.find(U.allLocalIps(), null, new IpV4AddressPredicate());
+
+        checkConnectivityByIp(extIp, getAllIps());
+    }
+
+    /**
+     * Tests correct behavior in case of 1 REST-enabled node
+     * with explicitly specified loopback address setting.
+     *
+     * @throws Exception If error occurs.
+     */
+    public void testOneNodeLoopbackHost() throws Exception {
+        startRestNode("grid1", LOOPBACK_IP, defaultRestPort());
+
+        checkConnectivityByIp(LOOPBACK_IP, F.t((Collection<String>)Collections.singleton(LOOPBACK_IP),
+            (Collection<String>)Collections.singleton("")));
+    }
+
+    /**
+     * Tests correct behavior in case of 1 REST-enabled node
+     * with explicitly specified 0.0.0.0 address.
+     *
+     * @throws Exception If error occurs.
+     */
+    public void testOneNodeZeroIpv4Address() throws Exception {
+        startRestNode("grid1", WILDCARD_IP, defaultRestPort());
+
+        Collection<String> addrs = new LinkedList<>();
+
+        addrs.add(LOOPBACK_IP);
+
+        Collection<String> nonLoopbackAddrs = U.allLocalIps();
+
+        assertNotNull(nonLoopbackAddrs);
+
+        addrs.addAll(F.view(nonLoopbackAddrs, new IpV4AddressPredicate()));
+
+        // The client should be able to connect through all IPv4 addresses.
+        for (String addr : addrs) {
+            log.info("Trying address: " + addr);
+
+            GridClient cli = startClient(addr, defaultRestPort());
+
+            List<GridClientNode> nodes = cli.compute().refreshTopology(true, false);
+
+            assertEquals(1, nodes.size());
+
+            GridClientNode node = F.first(nodes);
+
+            assertNotNull(node);
+
+            assertEquals(getAllIps().get1(), node.attribute(restAddressAttributeName()));
+            assertEquals(getAllIps().get2(), node.attribute(restHostNameAttributeName()));
+
+            List<String> nodeAddrs = node.tcpAddresses();
+
+            assertTrue(nodeAddrs.contains(LOOPBACK_IP));
+
+            assertTrue(F.containsAll(nodeAddrs, addrs));
+        }
+    }
+
+    /**
+     * Tests correct behavior in case of 2 REST-enabled nodes with default
+     * settings.
+     *
+     * @throws Exception If error occurs.
+     */
+    public void testTwoNodesDefaultHostAndPort() throws Exception {
+        startRestNode("grid1", null, null);
+        startRestNode("grid2", null, null);
+
+        GridClient cli = startClient(LOOPBACK_IP, defaultRestPort());
+
+        List<GridClientNode> nodes = cli.compute().refreshTopology(true, false);
+
+        assertEquals(2, nodes.size());
+
+        assertTrue(F.forAll(nodes, new P1<GridClientNode>() {
+            @Override public boolean apply(GridClientNode node) {
+                return node.tcpAddresses().contains(LOOPBACK_IP);
+            }
+        }));
+
+        GridTestUtils.assertOneToOne(
+            nodes,
+            new P1<GridClientNode>() {
+                @Override public boolean apply(GridClientNode node) {
+                    try {
+                        return eqAddresses(getAllIps(), node) &&
+                            Integer.valueOf(defaultRestPort()).equals(node.attribute(restPortAttributeName()));
+                    }
+                    catch (Exception ignored) {
+                        return false;
+                    }
+                }
+            },
+            new P1<GridClientNode>() {
+                @Override public boolean apply(GridClientNode node) {
+                    try {
+                        return eqAddresses(getAllIps(), node) &&
+                            Integer.valueOf(defaultRestPort() + 1).equals(node.attribute(restPortAttributeName()));
+                    }
+                    catch (Exception ignored) {
+                        return false;
+                    }
+                }
+            }
+        );
+    }
+
+    /**
+     * Tests correct behavior in case of shutdown node used to refresh topology state.
+     *
+     * @throws Exception If error occurs.
+     */
+    public void testRefreshTopologyOnNodeLeft() throws Exception {
+        startRestNode("grid1", null, null);
+        startRestNode("grid2", null, null);
+
+        GridClient cli = startClient(LOOPBACK_IP, defaultRestPort());
+
+        List<GridClientNode> nodes = cli.compute().refreshTopology(true, false);
+
+        assertEquals(2, nodes.size());
+
+        stopGrid("grid1");
+
+        nodes = cli.compute().refreshTopology(true, false);
+
+        assertEquals(1, nodes.size());
+
+        startRestNode("grid3", null, null);
+
+        nodes = cli.compute().refreshTopology(true, false);
+
+        assertEquals(2, nodes.size());
+
+        stopGrid("grid2");
+
+        nodes = cli.compute().refreshTopology(true, false);
+
+        assertEquals(1, nodes.size());
+    }
+
+    /**
+     * @param connectIp IP to test.
+     * @param nodeIp Expected IP reported to client.
+     * @throws GridClientException If failed.
+     */
+    private void checkConnectivityByIp(String connectIp, IgniteBiTuple<Collection<String>,
Collection<String>> nodeIp)
+        throws GridClientException {
+        GridClient cli = startClient(connectIp, defaultRestPort());
+
+        List<GridClientNode> nodes = cli.compute().refreshTopology(true, false);
+
+        assertEquals(1, nodes.size());
+
+        GridClientNode node = F.first(nodes);
+
+        assertNotNull(node);
+        assertTrue(eqAddresses(nodeIp, node));
+    }
+
+    /**
+     * @param nodeIp Node ip.
+     * @param node Node.
+     * @return {@code True} if addresses are equal, {@code false} otherwise.
+     */
+    private boolean eqAddresses(IgniteBiTuple<Collection<String>, Collection<String>>
nodeIp, GridClientNode node) {
+        return F.eqOrdered(nodeIp.get1(), (Collection<String>)(node.attribute(restAddressAttributeName())))
&&
+            F.eqOrdered(nodeIp.get2(), (Collection<String>)(node.attribute(restHostNameAttributeName())));
+    }
+
+    /**
+     * Predicate that returns IPv4 address strings.
+     */
+    private static class IpV4AddressPredicate implements P1<String> {
+        /** {@inheritDoc} */
+        @Override public boolean apply(String s) {
+            return s.matches("\\d+\\.\\d+\\.\\d+\\.\\d+");
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/756034f3/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
b/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
new file mode 100644
index 0000000..c6cedaa
--- /dev/null
+++ b/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
@@ -0,0 +1,858 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.client.integration;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.client.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.client.*;
+import org.apache.ignite.internal.client.balancer.*;
+import org.apache.ignite.internal.client.ssl.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.distributed.*;
+import org.apache.ignite.internal.processors.cache.version.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.*;
+import org.apache.ignite.spi.communication.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.internal.managers.communication.*;
+import org.apache.ignite.internal.processors.cache.transactions.*;
+import org.apache.ignite.internal.util.direct.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.testframework.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.jetbrains.annotations.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+import static java.util.concurrent.TimeUnit.*;
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+import static org.apache.ignite.cache.CacheDistributionMode.*;
+import static org.apache.ignite.cache.CacheMode.*;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
+
+/**
+ * Tests basic client behavior with multiple nodes.
+ */
+@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+public abstract class ClientAbstractMultiNodeSelfTest extends GridCommonAbstractTest {
+    /** */
+    private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+    /** Partitioned cache name. */
+    private static final String PARTITIONED_CACHE_NAME = "partitioned";
+
+    /** Replicated cache name. */
+    private static final String REPLICATED_CACHE_NAME = "replicated";
+
+    /** Replicated async cache name. */
+    private static final String REPLICATED_ASYNC_CACHE_NAME = "replicated_async";
+
+    /** Nodes count. */
+    public static final int NODES_CNT = 5;
+
+    /**
+     * Topology update frequency.
+     * Set it longer than router's, so we wouldn't receive failures
+     * caused only by obsolete topology on router.
+     */
+    static final int TOP_REFRESH_FREQ = 2500;
+
+    /** Path to jetty config. */
+    public static final String REST_JETTY_CFG = "modules/clients/src/test/resources/jetty/rest-jetty.xml";
+
+    /** Path to jetty config with SSl enabled. */
+    public static final String REST_JETTY_SSL_CFG = "modules/clients/src/test/resources/jetty/rest-jetty-ssl.xml";
+
+    /** Host. */
+    public static final String HOST = "127.0.0.1";
+
+    /** Base for tcp rest ports. */
+    public static final int REST_TCP_PORT_BASE = 12345;
+
+    /** Base for http rest ports, defined in {@link #REST_JETTY_CFG}. */
+    public static final int REST_HTTP_PORT_BASE = 11080;
+
+    /** Base for https rest ports, defined in {@link #REST_JETTY_SSL_CFG}. */
+    public static final int REST_HTTPS_PORT_BASE = 11443;
+
+    /** */
+    private static volatile boolean commSpiEnabled;
+
+    /** Flag to enable REST in node configuration. */
+    private boolean restEnabled = true;
+
+    /** Client instance for each test. */
+    private GridClient client;
+
+    /**
+     * @return Client protocol that should be used.
+     */
+    protected abstract GridClientProtocol protocol();
+
+    /**
+     * @return Server address to create first connection.
+     */
+    protected abstract String serverAddress();
+
+    /**
+     * @return SSL context factory to use if SSL or {@code null} to disable SSL usage.
+     */
+    @Nullable protected GridSslContextFactory sslContextFactory() {
+        return null;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
+        IgniteConfiguration c = super.getConfiguration(gridName);
+
+        c.setLocalHost(HOST);
+
+        assert c.getClientConnectionConfiguration() == null;
+
+        if (restEnabled) {
+            ClientConnectionConfiguration clientCfg = new ClientConnectionConfiguration();
+
+            clientCfg.setRestTcpPort(REST_TCP_PORT_BASE);
+
+            GridSslContextFactory sslCtxFactory = sslContextFactory();
+
+            if (sslCtxFactory != null) {
+                clientCfg.setRestTcpSslEnabled(true);
+                clientCfg.setRestTcpSslContextFactory(sslCtxFactory);
+            }
+
+            c.setClientConnectionConfiguration(clientCfg);
+        }
+
+        TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+        disco.setIpFinder(IP_FINDER);
+
+        c.setDiscoverySpi(disco);
+
+        TestCommunicationSpi spi = new TestCommunicationSpi();
+
+        spi.setLocalPort(GridTestUtils.getNextCommPort(getClass()));
+
+        c.setCommunicationSpi(spi);
+
+        c.setCacheConfiguration(cacheConfiguration(null), cacheConfiguration(PARTITIONED_CACHE_NAME),
+            cacheConfiguration(REPLICATED_CACHE_NAME), cacheConfiguration(REPLICATED_ASYNC_CACHE_NAME));
+
+        ThreadPoolExecutor exec = new ThreadPoolExecutor(
+            40,
+            40,
+            0,
+            MILLISECONDS,
+            new LinkedBlockingQueue<Runnable>());
+
+        exec.prestartAllCoreThreads();
+
+        c.setExecutorService(exec);
+
+        c.setExecutorServiceShutdown(true);
+
+        ThreadPoolExecutor sysExec = new ThreadPoolExecutor(
+            40,
+            40,
+            0,
+            MILLISECONDS,
+            new LinkedBlockingQueue<Runnable>());
+
+        sysExec.prestartAllCoreThreads();
+
+        c.setSystemExecutorService(sysExec);
+
+        c.setSystemExecutorServiceShutdown(true);
+
+        return c;
+    }
+
+    /**
+     * @param cacheName Cache name.
+     * @return Cache configuration.
+     * @throws Exception In case of error.
+     */
+    private CacheConfiguration cacheConfiguration(@Nullable String cacheName) throws Exception
{
+        CacheConfiguration cfg = defaultCacheConfiguration();
+
+        cfg.setAtomicityMode(TRANSACTIONAL);
+        cfg.setDistributionMode(NEAR_PARTITIONED);
+
+        if (cacheName == null)
+            cfg.setCacheMode(LOCAL);
+        else if (PARTITIONED_CACHE_NAME.equals(cacheName)) {
+            cfg.setCacheMode(PARTITIONED);
+
+            cfg.setBackups(0);
+        }
+        else
+            cfg.setCacheMode(REPLICATED);
+
+        cfg.setName(cacheName);
+
+        cfg.setWriteSynchronizationMode(REPLICATED_ASYNC_CACHE_NAME.equals(cacheName) ? FULL_ASYNC
: FULL_SYNC);
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        startGridsMultiThreaded(NODES_CNT);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTestsStopped() throws Exception {
+        info("Stopping grids.");
+
+        stopAllGrids();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        client = GridClientFactory.start(clientConfiguration());
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        if (client != null) {
+            GridClientFactory.stop(client.id(), false);
+
+            client = null;
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testSyncCommitRollbackFlags() throws Exception {
+        commSpiEnabled = true;
+
+        try {
+            GridClientData data = client.data(REPLICATED_ASYNC_CACHE_NAME);
+
+            info("Before put x1");
+
+            data.put("x1", "y1");
+
+            info("Before put x2");
+
+            data.flagsOn(GridClientCacheFlag.SYNC_COMMIT).put("x2", "y2");
+
+            info("Before put x3");
+
+            data.put("x3", "y3");
+
+            info("Before put x4");
+
+            data.flagsOn(GridClientCacheFlag.SYNC_COMMIT).put("x4", "y4");
+        }
+        finally {
+            commSpiEnabled = false;
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testEmptyProjections() throws Exception {
+        final GridClientCompute dflt = client.compute();
+
+        Collection<? extends GridClientNode> nodes = dflt.nodes();
+
+        assertEquals(NODES_CNT, nodes.size());
+
+        Iterator<? extends GridClientNode> iter = nodes.iterator();
+
+        final GridClientCompute singleNodePrj = dflt.projection(Collections.singletonList(iter.next()));
+
+        final GridClientNode second = iter.next();
+
+        final GridClientPredicate<GridClientNode> noneFilter = new GridClientPredicate<GridClientNode>()
{
+            @Override public boolean apply(GridClientNode node) {
+                return false;
+            }
+        };
+
+        final GridClientPredicate<GridClientNode> targetFilter = new GridClientPredicate<GridClientNode>()
{
+            @Override public boolean apply(GridClientNode node) {
+                return node.nodeId().equals(second.nodeId());
+            }
+        };
+
+        GridTestUtils.assertThrows(log(), new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                return dflt.projection(noneFilter).log(-1, -1);
+            }
+        }, GridServerUnreachableException.class, null);
+
+        GridTestUtils.assertThrows(log(), new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                return singleNodePrj.projection(second);
+            }
+        }, GridClientException.class, null);
+
+        GridTestUtils.assertThrows(log(), new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                return singleNodePrj.projection(targetFilter);
+            }
+        }, GridClientException.class, null);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testProjectionRun() throws Exception {
+        GridClientCompute dflt = client.compute();
+
+        Collection<? extends GridClientNode> nodes = dflt.nodes();
+
+        assertEquals(NODES_CNT, nodes.size());
+
+        for (int i = 0; i < NODES_CNT; i++) {
+            Ignite g = grid(i);
+
+            assert g != null;
+
+            GridClientNode clientNode = dflt.node(g.cluster().localNode().id());
+
+            assertNotNull("Client node for " + g.cluster().localNode().id() + " was not found",
clientNode);
+
+            GridClientCompute prj = dflt.projection(clientNode);
+
+            String res = prj.execute(TestTask.class.getName(), null);
+
+            assertNotNull(res);
+
+            assertEquals(g.cluster().localNode().id().toString(), res);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAffinityExecute() throws Exception {
+        GridClientCompute dflt = client.compute();
+
+        GridClientData data = client.data(PARTITIONED_CACHE_NAME);
+
+        Collection<? extends GridClientNode> nodes = dflt.nodes();
+
+        assertEquals(NODES_CNT, nodes.size());
+
+        for (int i = 0; i < NODES_CNT; i++) {
+            Ignite g = grid(i);
+
+            assert g != null;
+
+            int affinityKey = -1;
+
+            for (int key = 0; key < 10000; key++) {
+                if (g.cluster().localNode().id().equals(data.affinity(key))) {
+                    affinityKey = key;
+
+                    break;
+                }
+            }
+
+            if (affinityKey == -1)
+                throw new Exception("Unable to found key for which node is primary: " + g.cluster().localNode().id());
+
+            GridClientNode clientNode = dflt.node(g.cluster().localNode().id());
+
+            assertNotNull("Client node for " + g.cluster().localNode().id() + " was not found",
clientNode);
+
+            String res = dflt.affinityExecute(TestTask.class.getName(), PARTITIONED_CACHE_NAME,
affinityKey, null);
+
+            assertNotNull(res);
+
+            assertEquals(g.cluster().localNode().id().toString(), res);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testInvalidateFlag() throws Exception {
+        IgniteEx g0 = grid(0);
+
+        GridCache<String, String> cache = g0.cache(PARTITIONED_CACHE_NAME);
+
+        String key = null;
+
+        for (int i = 0; i < 10_000; i++) {
+            if (!cache.affinity().isPrimaryOrBackup(g0.localNode(), String.valueOf(i))) {
+                key = String.valueOf(i);
+
+                break;
+            }
+        }
+
+        assertNotNull(key);
+
+        cache.put(key, key); // Create entry in near cache, it is invalidated if INVALIDATE
flag is set.
+
+        assertNotNull(cache.peek(key));
+
+        GridClientData d = client.data(PARTITIONED_CACHE_NAME);
+
+        d.flagsOn(GridClientCacheFlag.INVALIDATE).put(key, "zzz");
+
+        for (Ignite g : G.allGrids()) {
+            cache = g.cache(PARTITIONED_CACHE_NAME);
+
+            if (cache.affinity().isPrimaryOrBackup(g.cluster().localNode(), key))
+                assertEquals("zzz", cache.peek(key));
+            else
+                assertNull(cache.peek(key));
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testClientAffinity() throws Exception {
+        GridClientData partitioned = client.data(PARTITIONED_CACHE_NAME);
+
+        Collection<Object> keys = new ArrayList<>();
+
+        keys.addAll(Arrays.asList(
+            Boolean.TRUE,
+            Boolean.FALSE,
+            1,
+            Integer.MAX_VALUE
+        ));
+
+        Random rnd = new Random();
+        StringBuilder sb = new StringBuilder();
+
+        // Generate some random strings.
+        for (int i = 0; i < 100; i++) {
+            sb.setLength(0);
+
+            for (int j = 0; j < 255; j++)
+                // Only printable ASCII symbols for test.
+                sb.append((char)(rnd.nextInt(0x7f - 0x20) + 0x20));
+
+            keys.add(sb.toString());
+        }
+
+        // Generate some more keys to achieve better coverage.
+        for (int i = 0; i < 100; i++)
+            keys.add(UUID.randomUUID());
+
+        for (Object key : keys) {
+            UUID nodeId = grid(0).mapKeyToNode(PARTITIONED_CACHE_NAME, key).id();
+
+            UUID clientNodeId = partitioned.affinity(key);
+
+            assertEquals("Invalid affinity mapping for REST response for key: " + key, nodeId,
clientNodeId);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testTopologyListener() throws Exception {
+        final Collection<UUID> added = new ArrayList<>(1);
+        final Collection<UUID> rmvd = new ArrayList<>(1);
+
+        final CountDownLatch addedLatch = new CountDownLatch(1);
+        final CountDownLatch rmvLatch = new CountDownLatch(1);
+
+        assertEquals(NODES_CNT, client.compute().refreshTopology(false, false).size());
+
+        GridClientTopologyListener lsnr = new GridClientTopologyListener() {
+            @Override public void onNodeAdded(GridClientNode node) {
+                added.add(node.nodeId());
+
+                addedLatch.countDown();
+            }
+
+            @Override public void onNodeRemoved(GridClientNode node) {
+                rmvd.add(node.nodeId());
+
+                rmvLatch.countDown();
+            }
+        };
+
+        client.addTopologyListener(lsnr);
+
+        try {
+            Ignite g = startGrid(NODES_CNT + 1);
+
+            UUID id = g.cluster().localNode().id();
+
+            assertTrue(addedLatch.await(2 * TOP_REFRESH_FREQ, MILLISECONDS));
+
+            assertEquals(1, added.size());
+            assertEquals(id, F.first(added));
+
+            stopGrid(NODES_CNT + 1);
+
+            assertTrue(rmvLatch.await(2 * TOP_REFRESH_FREQ, MILLISECONDS));
+
+            assertEquals(1, rmvd.size());
+            assertEquals(id, F.first(rmvd));
+        }
+        finally {
+            client.removeTopologyListener(lsnr);
+
+            stopGrid(NODES_CNT + 1);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testDisabledRest() throws Exception {
+        restEnabled = false;
+
+        final Ignite g = startGrid("disabled-rest");
+
+        try {
+            Thread.sleep(2 * TOP_REFRESH_FREQ);
+
+            // As long as we have round robin load balancer this will cause every node to
be queried.
+            for (int i = 0; i < NODES_CNT + 1; i++)
+                assertEquals(NODES_CNT + 1, client.compute().refreshTopology(false, false).size());
+
+            final GridClientData data = client.data(PARTITIONED_CACHE_NAME);
+
+            // Check rest-disabled node is unavailable.
+            try {
+                String affKey;
+
+                do {
+                    affKey = UUID.randomUUID().toString();
+                } while (!data.affinity(affKey).equals(g.cluster().localNode().id()));
+
+                data.put(affKey, "asdf");
+
+                assertEquals("asdf", cache(0, PARTITIONED_CACHE_NAME).get(affKey));
+            }
+            catch (GridServerUnreachableException e) {
+                // Thrown for direct client-node connections.
+                assertTrue("Unexpected exception message: " + e.getMessage(),
+                    e.getMessage().startsWith("No available endpoints to connect (is rest
enabled for this node?)"));
+            }
+            catch (GridClientException e) {
+                // Thrown for routed client-router-node connections.
+                String msg = e.getMessage();
+
+                assertTrue("Unexpected exception message: " + msg, protocol() == GridClientProtocol.TCP
?
+                    msg.contains("No available endpoints to connect (is rest enabled for
this node?)") : // TCP router.
+                    msg.startsWith("No available nodes on the router for destination node
ID"));         // HTTP router.
+            }
+
+            // Check rest-enabled nodes are available.
+            String affKey;
+
+            do {
+                affKey = UUID.randomUUID().toString();
+            } while (data.affinity(affKey).equals(g.cluster().localNode().id()));
+
+            data.put(affKey, "fdsa");
+
+            assertEquals("fdsa", cache(0, PARTITIONED_CACHE_NAME).get(affKey));
+        }
+        finally {
+            restEnabled = true;
+
+            G.stop(g.name(), true);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAffinityPut() throws Exception {
+        Thread.sleep(2 * TOP_REFRESH_FREQ);
+
+        assertEquals(NODES_CNT, client.compute().refreshTopology(false, false).size());
+
+        Map<UUID, Ignite> gridsByLocNode = new HashMap<>(NODES_CNT);
+
+        GridClientData partitioned = client.data(PARTITIONED_CACHE_NAME);
+
+        GridClientCompute compute = client.compute();
+
+        for (int i = 0; i < NODES_CNT; i++)
+            gridsByLocNode.put(grid(i).localNode().id(), grid(i));
+
+        for (int i = 0; i < 100; i++) {
+            String key = "key" + i;
+
+            UUID primaryNodeId = grid(0).mapKeyToNode(PARTITIONED_CACHE_NAME, key).id();
+
+            assertEquals("Affinity mismatch for key: " + key, primaryNodeId, partitioned.affinity(key));
+
+            assertEquals(primaryNodeId, partitioned.affinity(key));
+
+            // Must go to primary node only. Since backup count is 0, value must present
on
+            // primary node only.
+            partitioned.put(key, "val" + key);
+
+            for (Map.Entry<UUID, Ignite> entry : gridsByLocNode.entrySet()) {
+                Object val = entry.getValue().cache(PARTITIONED_CACHE_NAME).peek(key);
+
+                if (primaryNodeId.equals(entry.getKey()))
+                    assertEquals("val" + key, val);
+                else
+                    assertNull(val);
+            }
+        }
+
+        // Now check that we will see value in near cache in pinned mode.
+        for (int i = 100; i < 200; i++) {
+            String pinnedKey = "key" + i;
+
+            UUID primaryNodeId = grid(0).mapKeyToNode(PARTITIONED_CACHE_NAME, pinnedKey).id();
+
+            UUID pinnedNodeId = F.first(F.view(gridsByLocNode.keySet(), F.notEqualTo(primaryNodeId)));
+
+            GridClientNode node = compute.node(pinnedNodeId);
+
+            partitioned.pinNodes(node).put(pinnedKey, "val" + pinnedKey);
+
+            for (Map.Entry<UUID, Ignite> entry : gridsByLocNode.entrySet()) {
+                Object val = entry.getValue().cache(PARTITIONED_CACHE_NAME).peek(pinnedKey);
+
+                if (primaryNodeId.equals(entry.getKey()) || pinnedNodeId.equals(entry.getKey()))
+                    assertEquals("val" + pinnedKey, val);
+                else
+                    assertNull(val);
+            }
+        }
+    }
+
+    /**
+     * @return Client configuration for the test.
+     */
+    protected GridClientConfiguration clientConfiguration() throws GridClientException {
+        GridClientConfiguration cfg = new GridClientConfiguration();
+
+        cfg.setBalancer(getBalancer());
+
+        cfg.setTopologyRefreshFrequency(TOP_REFRESH_FREQ);
+
+        cfg.setProtocol(protocol());
+        cfg.setServers(Arrays.asList(serverAddress()));
+        cfg.setSslContextFactory(sslContextFactory());
+
+        GridClientDataConfiguration loc = new GridClientDataConfiguration();
+
+        GridClientDataConfiguration partitioned = new GridClientDataConfiguration();
+
+        partitioned.setName(PARTITIONED_CACHE_NAME);
+        partitioned.setAffinity(new GridClientPartitionAffinity());
+
+        GridClientDataConfiguration replicated = new GridClientDataConfiguration();
+        replicated.setName(REPLICATED_CACHE_NAME);
+
+        GridClientDataConfiguration replicatedAsync = new GridClientDataConfiguration();
+        replicatedAsync.setName(REPLICATED_ASYNC_CACHE_NAME);
+
+        cfg.setDataConfigurations(Arrays.asList(loc, partitioned, replicated, replicatedAsync));
+
+        return cfg;
+    }
+
+    /**
+     * Gets client load balancer.
+     *
+     * @return Load balancer.
+     */
+    protected GridClientLoadBalancer getBalancer() {
+        return new GridClientRoundRobinBalancer();
+    }
+
+    /**
+     * Test task. Returns a tuple in which first component is id of node that has split the
task,
+     * and second component is count of nodes that executed jobs.
+     */
+    private static class TestTask extends ComputeTaskSplitAdapter<Object, String> {
+        /** */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /** Count of tasks this job was split to. */
+        private int gridSize;
+
+        /** {@inheritDoc} */
+        @Override protected Collection<? extends ComputeJob> split(int gridSize, Object
arg)
+            throws IgniteCheckedException {
+            Collection<ComputeJobAdapter> jobs = new ArrayList<>(gridSize);
+
+            this.gridSize = gridSize;
+
+            final String locNodeId = ignite.cluster().localNode().id().toString();
+
+            for (int i = 0; i < gridSize; i++) {
+                jobs.add(new ComputeJobAdapter() {
+                    @SuppressWarnings("OverlyStrongTypeCast")
+                    @Override public Object execute() {
+                        try {
+                            Thread.sleep(1000);
+                        }
+                        catch (InterruptedException ignored) {
+                            Thread.currentThread().interrupt();
+                        }
+
+                        return new IgniteBiTuple<>(locNodeId, 1);
+                    }
+                });
+            }
+
+            return jobs;
+        }
+
+        /** {@inheritDoc} */
+        @Override public String reduce(List<ComputeJobResult> results) throws IgniteCheckedException
{
+            int sum = 0;
+
+            String locNodeId = null;
+
+            for (ComputeJobResult res : results) {
+                IgniteBiTuple<String, Integer> part = res.getData();
+
+                if (locNodeId == null)
+                    locNodeId = part.get1();
+
+                Integer i = part.get2();
+
+                if (i != null)
+                    sum += i;
+            }
+
+            assert gridSize == sum;
+
+            return locNodeId;
+        }
+    }
+
+    /**
+     * Communication SPI which checks cache flags.
+     */
+    @SuppressWarnings("unchecked")
+    private static class TestCommunicationSpi extends TcpCommunicationSpi {
+        /** {@inheritDoc} */
+        @Override public void sendMessage(ClusterNode node, GridTcpCommunicationMessageAdapter
msg)
+            throws IgniteSpiException {
+            checkSyncFlags((GridIoMessage)msg);
+
+            super.sendMessage(node, msg);
+        }
+
+        /**
+         * Check if flags in correct state.
+         *
+         * @param msg Message.
+         */
+        private void checkSyncFlags(GridIoMessage msg) {
+            if (!commSpiEnabled)
+                return;
+
+            Object o = msg.message();
+
+            if (!(o instanceof GridDistributedLockRequest))
+                return;
+
+            IgniteKernal g = (IgniteKernal)G.ignite(ignite.configuration().getNodeId());
+
+            GridCacheContext<Object, Object> cacheCtx = g.internalCache(REPLICATED_ASYNC_CACHE_NAME).context();
+
+            IgniteTxManager<Object, Object> tm = cacheCtx.tm();
+
+            GridCacheVersion v = ((GridCacheVersionable)o).version();
+
+            IgniteTxEx t = tm.tx(v);
+
+            if (t.hasWriteKey(cacheCtx.txKey("x1")))
+                assertFalse("Invalid tx flags: " + t, t.syncCommit());
+            else if (t.hasWriteKey(cacheCtx.txKey("x2")))
+                assertTrue("Invalid tx flags: " + t, t.syncCommit());
+            else if (t.hasWriteKey(cacheCtx.txKey("x3")))
+                assertFalse("Invalid tx flags: " + t, t.syncCommit());
+            else if (t.hasWriteKey(cacheCtx.txKey("x4")))
+                assertTrue("Invalid tx flags: " + t, t.syncCommit());
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testMultithreadedCommand() throws Exception {
+        final GridClientData data = client.data(PARTITIONED_CACHE_NAME);
+        final GridClientCompute compute = client.compute();
+        final AtomicInteger cnt = new AtomicInteger(0);
+
+        multithreaded(new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                for (int i = 0; i < 20; i++) {
+                    String key = UUID.randomUUID().toString();
+                    String val = UUID.randomUUID().toString();
+
+                    switch (cnt.incrementAndGet() % 4) {
+                        case 0: {
+                            assertTrue(data.put(key, val));
+                            assertEquals(val, data.get(key));
+                            assertTrue(data.remove(key));
+
+                            break;
+                        }
+
+                        case 1: {
+                            assertNotNull(data.metrics());
+
+                            break;
+                        }
+
+                        case 2: {
+                            String nodeId = compute.execute(TestTask.class.getName(), null);
+
+                            assertNotNull(nodeId);
+                            assertNotNull(compute.refreshNode(UUID.fromString(nodeId), true,
true));
+
+                            break;
+                        }
+
+                        case 3: {
+                            assertEquals(NODES_CNT, compute.refreshTopology(true, true).size());
+
+                            break;
+                        }
+                    }
+                }
+
+                return null;
+            }
+        }, 50, "multithreaded-client-access");
+    }
+}


Mime
View raw message