| /* |
| * 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 java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ThreadFactory; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import javax.cache.Cache; |
| import javax.cache.configuration.Factory; |
| import com.fasterxml.jackson.databind.JsonNode; |
| import com.fasterxml.jackson.databind.ObjectMapper; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.cache.store.CacheStore; |
| import org.apache.ignite.cache.store.CacheStoreAdapter; |
| import org.apache.ignite.compute.ComputeJob; |
| import org.apache.ignite.compute.ComputeJobAdapter; |
| import org.apache.ignite.compute.ComputeJobResult; |
| import org.apache.ignite.compute.ComputeTaskSplitAdapter; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.ConnectorConfiguration; |
| import org.apache.ignite.configuration.ConnectorMessageInterceptor; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.internal.client.GridClient; |
| import org.apache.ignite.internal.client.GridClientClosedException; |
| import org.apache.ignite.internal.client.GridClientCompute; |
| import org.apache.ignite.internal.client.GridClientConfiguration; |
| import org.apache.ignite.internal.client.GridClientData; |
| import org.apache.ignite.internal.client.GridClientDataConfiguration; |
| import org.apache.ignite.internal.client.GridClientException; |
| import org.apache.ignite.internal.client.GridClientFactory; |
| import org.apache.ignite.internal.client.GridClientFuture; |
| import org.apache.ignite.internal.client.GridClientNode; |
| import org.apache.ignite.internal.client.GridClientPredicate; |
| import org.apache.ignite.internal.client.GridClientProtocol; |
| import org.apache.ignite.internal.client.GridServerUnreachableException; |
| import org.apache.ignite.internal.client.ssl.GridSslContextFactory; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteBiInClosure; |
| import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; |
| import org.jetbrains.annotations.NotNull; |
| import org.jetbrains.annotations.Nullable; |
| import org.junit.Assert; |
| import org.junit.Test; |
| import static org.apache.ignite.IgniteSystemProperties.IGNITE_JETTY_PORT; |
| import static org.apache.ignite.cache.CacheMode.PARTITIONED; |
| import static org.apache.ignite.cache.CacheMode.REPLICATED; |
| |
| /** |
| * Tests for Java client. |
| */ |
| @SuppressWarnings("deprecation") |
| public abstract class ClientAbstractSelfTest extends GridCommonAbstractTest { |
| /** */ |
| private static final String REPLICATED_CACHE_NAME = "replicated"; |
| |
| /** */ |
| private static final String PARTITIONED_CACHE_NAME = "partitioned"; |
| |
| /** */ |
| public static final String HOST = "127.0.0.1"; |
| |
| /** */ |
| public static final int JETTY_PORT = 8080; |
| |
| /** */ |
| public static final int BINARY_PORT = 11212; |
| |
| /** Path to jetty config. */ |
| public static final String REST_JETTY_CFG = "modules/clients/src/test/resources/jetty/rest-jetty.xml"; |
| |
| /** Need to be static because configuration inits only once per class. */ |
| private static final ConcurrentMap<Object, Object> INTERCEPTED_OBJECTS = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private static final Map<String, HashMapStore> cacheStores = new HashMap<>(); |
| |
| /** */ |
| public static final String ROUTER_LOG_CFG = "modules/core/src/test/config/log4j2-test.xml"; |
| |
| /** */ |
| private static final String INTERCEPTED_SUF = "intercepted"; |
| |
| /** */ |
| private static final String[] TASK_ARGS = new String[] {"executing", "test", "task"}; |
| |
| /** Flag indicating whether intercepted objects should be overwritten. */ |
| private static volatile boolean overwriteIntercepted; |
| |
| /** */ |
| private ExecutorService exec; |
| |
| /** */ |
| protected GridClient client; |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTestsStarted() throws Exception { |
| System.setProperty(IGNITE_JETTY_PORT, Integer.toString(JETTY_PORT)); |
| |
| startGrid(); |
| |
| System.clearProperty(IGNITE_JETTY_PORT); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTest() throws Exception { |
| exec = Executors.newCachedThreadPool(); |
| |
| client = client(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void afterTest() throws Exception { |
| U.shutdownNow(ClientAbstractSelfTest.class, exec, log); |
| |
| exec = null; |
| |
| if (client != null) |
| GridClientFactory.stop(client.id(), true); |
| |
| client = null; |
| |
| synchronized (cacheStores) { |
| for (HashMapStore cacheStore : cacheStores.values()) |
| cacheStore.map.clear(); |
| } |
| |
| grid().cache(PARTITIONED_CACHE_NAME).clear(); |
| grid().cache(REPLICATED_CACHE_NAME).clear(); |
| |
| INTERCEPTED_OBJECTS.clear(); |
| } |
| |
| /** |
| * Gets protocol which should be used in client connection. |
| * |
| * @return Protocol. |
| */ |
| protected abstract GridClientProtocol protocol(); |
| |
| /** |
| * Gets server address to which client should connect. |
| * |
| * @return Server address in format "host:port". |
| */ |
| protected abstract String serverAddress(); |
| |
| /** |
| * @return Whether SSL should be used in test. |
| */ |
| protected abstract boolean useSsl(); |
| |
| /** |
| * @return SSL context factory used in test. |
| */ |
| protected abstract GridSslContextFactory sslContextFactory(); |
| |
| /** |
| * Get task name. |
| * |
| * @return Task name. |
| */ |
| protected String getTaskName() { |
| return TestTask.class.getName(); |
| } |
| |
| /** |
| * @return name of the sleep task for current test. |
| */ |
| protected String getSleepTaskName() { |
| return SleepTestTask.class.getName(); |
| } |
| |
| /** |
| * Get task argument. |
| * |
| * @return Task argument. |
| */ |
| protected Object getTaskArgument() { |
| return Arrays.asList(TASK_ARGS); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { |
| IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); |
| |
| cfg.setLocalHost(HOST); |
| |
| assert cfg.getConnectorConfiguration() == null; |
| |
| ConnectorConfiguration clientCfg = new ConnectorConfiguration(); |
| |
| clientCfg.setPort(BINARY_PORT); |
| |
| if (useSsl()) { |
| clientCfg.setSslEnabled(true); |
| |
| clientCfg.setSslContextFactory(sslContextFactory()); |
| } |
| |
| cfg.setConnectorConfiguration(clientCfg); |
| |
| cfg.setCacheConfiguration(cacheConfiguration(REPLICATED_CACHE_NAME), cacheConfiguration(PARTITIONED_CACHE_NAME)); |
| |
| clientCfg.setMessageInterceptor(new ConnectorMessageInterceptor() { |
| /** {@inheritDoc} */ |
| @Override public Object onReceive(@Nullable Object obj) { |
| if (obj != null) |
| INTERCEPTED_OBJECTS.put(obj, obj); |
| |
| return overwriteIntercepted && obj instanceof String ? |
| obj + INTERCEPTED_SUF : obj; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Object onSend(Object obj) { |
| if (obj != null) |
| INTERCEPTED_OBJECTS.put(obj, obj); |
| |
| return obj; |
| } |
| }); |
| |
| return cfg; |
| } |
| |
| /** |
| * @param cacheName Cache name. |
| * @return Cache configuration. |
| * @throws Exception In case of error. |
| */ |
| @SuppressWarnings("unchecked") |
| private static CacheConfiguration cacheConfiguration(@NotNull final String cacheName) throws Exception { |
| CacheConfiguration cfg = defaultCacheConfiguration(); |
| |
| cfg.setCacheMode(REPLICATED_CACHE_NAME.equals(cacheName) ? REPLICATED : PARTITIONED); |
| cfg.setName(cacheName); |
| |
| cfg.setCacheStoreFactory(new Factory<CacheStore>() { |
| @Override public CacheStore create() { |
| synchronized (cacheStores) { |
| HashMapStore cacheStore = cacheStores.get(cacheName); |
| |
| if (cacheStore == null) |
| cacheStores.put(cacheName, cacheStore = new HashMapStore()); |
| |
| return cacheStore; |
| } |
| } |
| }); |
| |
| cfg.setWriteThrough(true); |
| cfg.setReadThrough(true); |
| cfg.setLoadPreviousValue(true); |
| |
| if (cfg.getCacheMode() == PARTITIONED) |
| cfg.setBackups(1); |
| |
| return cfg; |
| } |
| |
| /** |
| * @return Client. |
| * @throws GridClientException In case of error. |
| */ |
| protected GridClient client() throws GridClientException { |
| return GridClientFactory.start(clientConfiguration()); |
| } |
| |
| /** |
| * @return Test client configuration. |
| */ |
| protected GridClientConfiguration clientConfiguration() throws GridClientException { |
| GridClientConfiguration cfg = new GridClientConfiguration(); |
| |
| GridClientDataConfiguration nullCache = new GridClientDataConfiguration(); |
| |
| GridClientDataConfiguration cache = new GridClientDataConfiguration(); |
| |
| cache.setName(PARTITIONED_CACHE_NAME); |
| |
| cfg.setDataConfigurations(Arrays.asList(nullCache, cache)); |
| |
| cfg.setProtocol(protocol()); |
| cfg.setServers(Collections.singleton(serverAddress())); |
| |
| // Setting custom executor, to avoid failures on client shutdown. |
| // And applying custom naming scheme to ease debugging. |
| cfg.setExecutorService(Executors.newCachedThreadPool(new ThreadFactory() { |
| private AtomicInteger cntr = new AtomicInteger(); |
| |
| /** {@inheritDoc} */ |
| @Override public Thread newThread(Runnable r) { |
| return new Thread(r, "client-worker-thread-" + cntr.getAndIncrement()); |
| } |
| })); |
| |
| if (useSsl()) |
| cfg.setSslContextFactory(sslContextFactory()); |
| |
| return cfg; |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testConnectable() throws Exception { |
| GridClient client = client(); |
| |
| List<GridClientNode> nodes = client.compute().refreshTopology(false, false); |
| |
| assertTrue(F.first(nodes).connectable()); |
| } |
| |
| /** |
| * Check async API methods don't generate exceptions. |
| * |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testNoAsyncExceptions() throws Exception { |
| GridClient client = client(); |
| |
| GridClientData data = client.data(PARTITIONED_CACHE_NAME); |
| GridClientCompute compute = client.compute().projection(new GridClientPredicate<GridClientNode>() { |
| @Override public boolean apply(GridClientNode e) { |
| return false; |
| } |
| }); |
| |
| Map<String, GridClientFuture<?>> futs = new LinkedHashMap<>(); |
| |
| futs.put("exec", compute.executeAsync("taskName", "taskArg")); |
| futs.put("affExec", compute.affinityExecuteAsync("taskName", "cacheName", "affKey", "taskArg")); |
| futs.put("refreshById", compute.refreshNodeAsync(UUID.randomUUID(), true, true)); |
| futs.put("refreshByIP", compute.refreshNodeAsync("nodeIP", true, true)); |
| futs.put("refreshTop", compute.refreshTopologyAsync(true, true)); |
| |
| GridClientFactory.stop(client.id(), false); |
| |
| futs.put("put", data.putAsync("key", "val")); |
| futs.put("putAll", data.putAllAsync(F.asMap("key", "val"))); |
| futs.put("get", data.getAsync("key")); |
| futs.put("getAll", data.getAllAsync(Collections.singletonList("key"))); |
| futs.put("remove", data.removeAsync("key")); |
| futs.put("removeAll", data.removeAllAsync(Collections.singletonList("key"))); |
| futs.put("replace", data.replaceAsync("key", "val")); |
| futs.put("cas", data.casAsync("key", "val", "val2")); |
| futs.put("metrics", data.metricsAsync()); |
| |
| for (Map.Entry<String, GridClientFuture<?>> e : futs.entrySet()) { |
| try { |
| e.getValue().get(); |
| |
| info("Expects '" + e.getKey() + "' fails with grid client exception."); |
| } |
| catch (GridServerUnreachableException | GridClientClosedException ignore) { |
| // No op: compute projection is empty. |
| } |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testGracefulShutdown() throws Exception { |
| GridClientCompute compute = client.compute(); |
| |
| Object taskArg = getTaskArgument(); |
| String taskName = getSleepTaskName(); |
| |
| GridClientFuture<Object> fut = compute.executeAsync(taskName, taskArg); |
| GridClientFuture<Object> fut2 = compute.executeAsync(taskName, taskArg); |
| |
| GridClientFactory.stop(client.id(), true); |
| |
| Assert.assertEquals(17, fut.get()); |
| Assert.assertEquals(17, fut2.get()); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testForceShutdown() throws Exception { |
| GridClientCompute compute = client.compute(); |
| |
| Object taskArg = getTaskArgument(); |
| String taskName = getSleepTaskName(); |
| |
| GridClientFuture<Object> fut = compute.executeAsync(taskName, taskArg); |
| |
| GridClientFactory.stop(client.id(), false); |
| |
| try { |
| fut.get(); |
| } |
| catch (GridClientClosedException ignored) { |
| return; |
| } |
| |
| Assert.fail("Expected GridClientClosedException."); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testShutdown() throws Exception { |
| GridClient c = client(); |
| |
| GridClientCompute compute = c.compute(); |
| |
| String taskName = getTaskName(); |
| Object taskArg = getTaskArgument(); |
| |
| Collection<GridClientFuture<Object>> futs = new ArrayList<>(); |
| |
| // Validate connection works. |
| compute.execute(taskName, taskArg); |
| |
| info(">>> First task executed successfully, running batch."); |
| |
| for (int i = 0; i < 10; i++) |
| futs.add(compute.executeAsync(taskName, taskArg)); |
| |
| // Stop client. |
| GridClientFactory.stop(c.id(), true); |
| |
| info(">>> Completed stop request."); |
| |
| int failed = 0; |
| |
| for (GridClientFuture<Object> fut : futs) { |
| try { |
| assertEquals(17, fut.get()); |
| } |
| catch (GridClientException e) { |
| failed++; |
| |
| log.warning("Task execution failed.", e); |
| } |
| } |
| |
| assertEquals(0, failed); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testExecute() throws Exception { |
| String taskName = getTaskName(); |
| Object taskArg = getTaskArgument(); |
| |
| GridClientCompute compute = client.compute(); |
| |
| assertEquals(Integer.valueOf(17), compute.execute(taskName, taskArg)); |
| assertEquals(Integer.valueOf(17), compute.executeAsync(taskName, taskArg).get()); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testTopology() throws Exception { |
| GridClientCompute compute = client.compute(); |
| |
| List<GridClientNode> top = compute.refreshTopology(true, true); |
| |
| assertNotNull(top); |
| assertEquals(1, top.size()); |
| |
| GridClientNode node = F.first(top); |
| |
| assertNotNull(node); |
| assertFalse(node.attributes().isEmpty()); |
| assertNotNull(node.tcpAddresses()); |
| assertEquals(grid().localNode().id(), node.nodeId()); |
| assertNotNull(node.metrics()); |
| |
| top = compute.refreshTopology(false, false); |
| |
| node = F.first(top); |
| |
| assertNotNull(top); |
| assertEquals(1, top.size()); |
| assertNull(node.metrics()); |
| assertTrue(node.attributes().isEmpty()); |
| |
| node = F.first(top); |
| |
| assertNotNull(node); |
| assertTrue(node.attributes().isEmpty()); |
| assertNull(node.metrics()); |
| assertNotNull(node.tcpAddresses()); |
| assertEquals(grid().localNode().id(), node.nodeId()); |
| |
| top = compute.refreshTopologyAsync(true, true).get(); |
| |
| assertNotNull(top); |
| assertEquals(1, top.size()); |
| |
| node = F.first(top); |
| |
| assertNotNull(node); |
| assertFalse(node.attributes().isEmpty()); |
| assertNotNull(node.metrics()); |
| assertNotNull(node.tcpAddresses()); |
| assertEquals(grid().localNode().id(), node.nodeId()); |
| |
| top = compute.refreshTopologyAsync(false, false).get(); |
| |
| assertNotNull(top); |
| assertEquals(1, top.size()); |
| |
| node = F.first(top); |
| |
| assertNotNull(node); |
| assertTrue(node.attributes().isEmpty()); |
| assertNull(node.metrics()); |
| assertNotNull(node.tcpAddresses()); |
| assertEquals(grid().localNode().id(), node.nodeId()); |
| } |
| |
| /** |
| * Test task. |
| */ |
| private static class TestTask extends ComputeTaskSplitAdapter<List<String>, Integer> { |
| /** {@inheritDoc} */ |
| @Override protected Collection<? extends ComputeJob> split(int gridSize, List<String> list) { |
| Collection<ComputeJobAdapter> jobs = new ArrayList<>(); |
| |
| if (list != null) |
| for (final String val : list) |
| jobs.add(new ComputeJobAdapter() { |
| @Override public Object execute() { |
| try { |
| Thread.sleep(1); |
| } |
| catch (InterruptedException ignored) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| return val == null ? 0 : val.length(); |
| } |
| }); |
| |
| return jobs; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Integer reduce(List<ComputeJobResult> results) { |
| int sum = 0; |
| |
| for (ComputeJobResult res : results) |
| sum += res.<Integer>getData(); |
| |
| return sum; |
| } |
| } |
| |
| /** |
| * Test task that sleeps 5 seconds. |
| */ |
| private static class SleepTestTask extends ComputeTaskSplitAdapter<List<String>, Integer> { |
| /** {@inheritDoc} */ |
| @Override protected Collection<? extends ComputeJob> split(int gridSize, List<String> list) { |
| Collection<ComputeJobAdapter> jobs = new ArrayList<>(); |
| |
| if (list != null) |
| for (final String val : list) |
| jobs.add(new ComputeJobAdapter() { |
| @Override public Object execute() { |
| try { |
| Thread.sleep(5000); |
| |
| return val == null ? 0 : val.length(); |
| } |
| catch (InterruptedException ignored) { |
| return -1; |
| } |
| } |
| }); |
| |
| return jobs; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Integer reduce(List<ComputeJobResult> results) { |
| int sum = 0; |
| |
| for (ComputeJobResult res : results) |
| sum += res.<Integer>getData(); |
| |
| return sum; |
| } |
| } |
| |
| /** JSON to java mapper. */ |
| private static final ObjectMapper JSON_MAPPER = new ObjectMapper(); |
| |
| /** |
| * Http test task with restriction to string arguments only. |
| */ |
| protected static class HttpTestTask extends ComputeTaskSplitAdapter<String, Integer> { |
| /** */ |
| private final TestTask delegate = new TestTask(); |
| |
| /** {@inheritDoc} */ |
| @Override protected Collection<? extends ComputeJob> split(int gridSize, String arg) { |
| if (arg.endsWith("intercepted")) |
| arg = arg.substring(0, arg.length() - 11); |
| |
| try { |
| JsonNode json = JSON_MAPPER.readTree(arg); |
| |
| List<String> list = null; |
| |
| if (json.isArray()) { |
| list = new ArrayList<>(); |
| |
| for (JsonNode child : json) |
| list.add(child.asText()); |
| } |
| |
| return delegate.split(gridSize, list); |
| } |
| catch (IOException e) { |
| throw new IgniteException(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Integer reduce(List<ComputeJobResult> results) { |
| return delegate.reduce(results); |
| } |
| } |
| |
| /** |
| * Http wrapper for sleep task. |
| */ |
| protected static class SleepHttpTestTask extends ComputeTaskSplitAdapter<String, Integer> { |
| /** */ |
| private final SleepTestTask delegate = new SleepTestTask(); |
| |
| /** {@inheritDoc} */ |
| @Override protected Collection<? extends ComputeJob> split(int gridSize, String arg) { |
| try { |
| JsonNode json = JSON_MAPPER.readTree(arg); |
| |
| List<String> list = null; |
| |
| if (json.isArray()) { |
| list = new ArrayList<>(); |
| |
| for (JsonNode child : json) |
| list.add(child.asText()); |
| } |
| |
| return delegate.split(gridSize, list); |
| } |
| catch (IOException e) { |
| throw new IgniteException(e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Integer reduce(List<ComputeJobResult> results) { |
| return delegate.reduce(results); |
| } |
| } |
| |
| /** |
| * Simple HashMap based cache store emulation. |
| */ |
| private static class HashMapStore extends CacheStoreAdapter<Object, Object> { |
| /** Map for cache store. */ |
| private final Map<Object, Object> map = new HashMap<>(); |
| |
| /** {@inheritDoc} */ |
| @Override public void loadCache(IgniteBiInClosure<Object, Object> clo, Object... args) { |
| for (Map.Entry e : map.entrySet()) |
| clo.apply(e.getKey(), e.getValue()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Object load(Object key) { |
| return map.get(key); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void write(Cache.Entry<?, ?> e) { |
| map.put(e.getKey(), e.getValue()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void delete(Object key) { |
| map.remove(key); |
| } |
| } |
| } |
