# ignite-648:
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/63e98328 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/63e98328 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/63e98328 Branch: refs/heads/ignite-648 Commit: 63e98328c42b58cf9c7264f21920cd6db9e220cc Parents: de9acc3 Author: ashutak <ashu...@gridgain.com> Authored: Fri Jun 19 21:16:09 2015 +0300 Committer: ashutak <ashu...@gridgain.com> Committed: Fri Jun 19 21:16:09 2015 +0300 ---------------------------------------------------------------------- .../cache/GridCacheAbstractFullApiSelfTest.java | 21 +- .../cache/GridCacheAbstractSelfTest.java | 43 +- .../framework/IgniteCacheProcessProxy.java | 38 +- .../framework/IgniteExProcessProxy.java | 465 ------------------ .../multijvm/framework/IgniteProcessProxy.java | 492 +++++++++++++++++++ 5 files changed, 561 insertions(+), 498 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63e98328/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java index bd8b502..9314b70 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java @@ -35,7 +35,6 @@ import org.apache.ignite.internal.util.typedef.*; import org.apache.ignite.internal.util.typedef.internal.*; import org.apache.ignite.lang.*; import org.apache.ignite.spi.discovery.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.spi.swapspace.inmemory.*; @@ -197,7 +196,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract Ignite ignite; if (isMultiJvm()) - ignite = IgniteExProcessProxy.grid(entry.getKey()); + ignite = IgniteProcessProxy.grid(entry.getKey()); else ignite = IgnitionEx.grid(entry.getKey()); @@ -237,21 +236,11 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract /** {@inheritDoc} */ @Override protected void afterTestsStopped() throws Exception { - IgniteExProcessProxy.killAll(); + IgniteProcessProxy.killAll(); super.afterTestsStopped(); } - - /** - * Gets flag whether nodes will run in one jvm or in separate jvms. - * - * @return <code>True</code> to run nodes in separate jvms. - */ - protected boolean isMultiJvm() { - return false; - } - /** {@inheritDoc} */ @Override protected Ignite startGrid(String gridName, GridSpringResourceContext ctx) throws Exception { if (!isMultiJvm() || gridName.endsWith("0")) { @@ -267,7 +256,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract return IgnitionEx.start(optimize(cfg), ctx); } - return new IgniteExProcessProxy(optimize(getConfiguration(gridName)), log, grid(0)); + return new IgniteProcessProxy(optimize(getConfiguration(gridName)), log, grid(0)); } /** {@inheritDoc} */ @@ -275,7 +264,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract if (!isMultiJvm() || idx == 0) return super.grid(idx); - return IgniteExProcessProxy.get(getTestGridName(idx)); + return IgniteProcessProxy.get(getTestGridName(idx)); } /** @@ -287,7 +276,7 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract if (!isMultiJvm() || idx == 0) return super.jcache(idx); - return IgniteExProcessProxy.get(getTestGridName(idx)).cache(null); + return IgniteProcessProxy.get(getTestGridName(idx)).cache(null); } /** {@inheritDoc} */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63e98328/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java index 3f14f49..f129dd0 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java @@ -23,6 +23,7 @@ import org.apache.ignite.cache.store.*; import org.apache.ignite.configuration.*; import org.apache.ignite.internal.*; import org.apache.ignite.internal.cluster.*; +import org.apache.ignite.internal.processors.cache.multijvm.framework.*; import org.apache.ignite.internal.util.lang.*; import org.apache.ignite.internal.util.typedef.*; import org.apache.ignite.internal.util.typedef.internal.*; @@ -357,11 +358,49 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest { } /** + * Gets flag whether nodes will run in one jvm or in separate jvms. + * + * @return <code>True</code> to run nodes in separate jvms. + */ + protected boolean isMultiJvm() { + return false; + } + + /** + * @param gridName Grid name. + * @return <code>True</code> if test was run in multy jvm mode and grid at another jvm. + */ + protected boolean isMultiJvmAndNodeIsRemote(String gridName) { + return isMultiJvm() && !gridName.endsWith("0"); + } + + /** + * @param idx Grid index. + * @return <code>True</code> if test was run in multy jvm mode and grid at another jvm. + */ + protected boolean isMultiJvmAndNodeIsRemote(int idx) { + return isMultiJvm() && idx != 0; + } + + /** * @param idx Index of grid. * @return Cache context. */ - protected GridCacheContext<String, Integer> context(int idx) { - return ((IgniteKernal)grid(idx)).<String, Integer>internalCache().context(); + protected GridCacheContext<String, Integer> context(final int idx) { + if (!isMultiJvmAndNodeIsRemote(idx)) + return ((IgniteKernal)grid(idx)).<String, Integer>internalCache().context(); + else { +// ((IgniteProcessProxy)grid(idx)).remoteInternalCache(); + + // TODO + final UUID id = ((IgniteProcessProxy)grid(idx)).getId(); + + return new GridCacheContext<String, Integer>() { + @Override public UUID localNodeId() { + return id; + } + }; + } } /** http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63e98328/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteCacheProcessProxy.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteCacheProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteCacheProcessProxy.java index 32c0bff..6e09286 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteCacheProcessProxy.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteCacheProcessProxy.java @@ -51,13 +51,13 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> { private final boolean isAsync; /** Ignite proxy. */ - private final transient IgniteExProcessProxy igniteProxy; + private final transient IgniteProcessProxy igniteProxy; /** * @param name Name. * @param proxy Ignite Process Proxy. */ - public IgniteCacheProcessProxy(String name, IgniteExProcessProxy proxy) { + public IgniteCacheProcessProxy(String name, IgniteProcessProxy proxy) { this(name, false, proxy); } @@ -66,7 +66,7 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> { * @param async * @param proxy Ignite Process Proxy. */ - public IgniteCacheProcessProxy(String name, boolean async, IgniteExProcessProxy proxy) { + public IgniteCacheProcessProxy(String name, boolean async, IgniteProcessProxy proxy) { cacheName = name; isAsync = async; gridId = proxy.getId(); @@ -93,8 +93,14 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> { } /** {@inheritDoc} */ - @Override public <C extends Configuration<K, V>> C getConfiguration(Class<C> clazz) { - throw new UnsupportedOperationException("Method should be supported."); + @Override public <C extends Configuration<K, V>> C getConfiguration(final Class<C> clazz) { + final Class cl = clazz; + + return (C)compute.call(new IgniteCallable<Object>() { + @Override public Object call() throws Exception { + return cache().getConfiguration(cl); + } + }); } /** {@inheritDoc} */ @@ -153,15 +159,17 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> { /** {@inheritDoc} */ @SuppressWarnings("unchecked") - @Override public Iterable<Entry<K, V>> localEntries(CachePeekMode... peekModes) throws CacheException { - // TODO: implement. -// return F.first(compute.broadcast(new IgniteClosureX<CachePeekMode[], Iterable>() { -// @Override public Iterable applyx(CachePeekMode... modes) { -// return Ignition.ignite(gridId).cache(cacheName).localEntries(modes); -// } -// }, peekModes)); + @Override public Iterable<Entry<K, V>> localEntries(final CachePeekMode... peekModes) throws CacheException { + return (Iterable<Entry<K, V>>)compute.call(new IgniteCallable<Object>() { + @Override public Object call() throws Exception { + Collection<Entry> res = new ArrayList<>(); + + for (Entry e : cache().localEntries(peekModes)) + res.add(e); - return Collections.emptyList(); + return res; + } + }); } /** {@inheritDoc} */ @@ -492,8 +500,8 @@ public class IgniteCacheProcessProxy<K, V> implements IgniteCache<K, V> { } /** {@inheritDoc} */ - @Override public <T> T unwrap(Class<T> clazz) { - throw new UnsupportedOperationException("Method should be supported."); + @Override public <T> T unwrap(final Class<T> clazz) { + throw new UnsupportedOperationException("Method cannot be supported because T can be unmarshalliable."); } /** {@inheritDoc} */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63e98328/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteExProcessProxy.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteExProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteExProcessProxy.java deleted file mode 100644 index b5b59c0..0000000 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteExProcessProxy.java +++ /dev/null @@ -1,465 +0,0 @@ -/* - * 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.processors.cache.multijvm.framework; - -import org.apache.ignite.*; -import org.apache.ignite.cache.affinity.*; -import org.apache.ignite.cluster.*; -import org.apache.ignite.configuration.*; -import org.apache.ignite.events.*; -import org.apache.ignite.internal.*; -import org.apache.ignite.internal.cluster.*; -import org.apache.ignite.internal.processors.cache.*; -import org.apache.ignite.internal.processors.hadoop.*; -import org.apache.ignite.internal.util.*; -import org.apache.ignite.internal.util.lang.*; -import org.apache.ignite.internal.util.typedef.*; -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.lang.*; -import org.apache.ignite.plugin.*; -import org.jetbrains.annotations.*; - -import java.util.*; -import java.util.concurrent.*; - -/** - * Ignite proxy for ignite instance at another JVM. - */ -public class IgniteExProcessProxy implements IgniteEx { - /** Grid proxies. */ - private transient static final Map<String, IgniteExProcessProxy> gridProxies = new HashMap<>(); - - /** Jvm process with ignite instance. */ - private transient final GridJavaProcess proc; - - /** Configuration. */ - private transient final IgniteConfiguration cfg; - - /** Local jvm grid. */ - private transient final Ignite locJvmGrid; - - /** Logger. */ - private transient final IgniteLogger log; - - /** Grid id. */ - private final UUID id = UUID.randomUUID(); - - /** Compute. */ - private transient final IgniteCompute compute; - - /** Remote ignite instance started latch. */ - private transient final CountDownLatch rmtNodeStartedLatch = new CountDownLatch(1); - - /** - * @param cfg Configuration. - * @param log Logger. - * @param locJvmGrid Local jvm grid. - */ - public IgniteExProcessProxy(final IgniteConfiguration cfg, final IgniteLogger log, final Ignite locJvmGrid) - throws Exception { - this.cfg = cfg; - this.locJvmGrid = locJvmGrid; - this.log = log.getLogger("jvm-" + id.toString().substring(0, id.toString().indexOf('-'))); - - String cfgFileName = IgniteNodeRunner.storeToFile(cfg.setNodeId(id)); - - List<String> jvmArgs = U.jvmArgs(); - - List<String> filteredJvmArgs = new ArrayList<>(); - - for (String arg : jvmArgs) { - if(!arg.toLowerCase().startsWith("-agentlib")) - filteredJvmArgs.add(arg); - } - - locJvmGrid.events().localListen(new IgnitePredicateX<Event>() { - @Override public boolean applyx(Event e) { - if (((DiscoveryEvent)e).eventNode().id().equals(id)) { - rmtNodeStartedLatch.countDown(); - - return false; - } - - return true; - } - }, EventType.EVT_NODE_JOINED); - - proc = GridJavaProcess.exec( - IgniteNodeRunner.class, - cfgFileName, // Params. - this.log, - // Optional closure to be called each time wrapped process prints line to system.out or system.err. - new IgniteInClosure<String>() { - @Override public void apply(String s) { - IgniteExProcessProxy.this.log.info(s); - } - }, - null, - filteredJvmArgs, // JVM Args. - System.getProperty("surefire.test.class.path") - ); - - assert rmtNodeStartedLatch.await(30, TimeUnit.SECONDS): "Remote node with id=" + id + " didn't join."; - - gridProxies.put(cfg.getGridName(), this); - - compute = locJvmGrid.compute(locJvmGrid.cluster().forNodeId(id)); - } - - /** - * @param gridName Grid name. - * @return Instance by name or <code>null</code>. - */ - public static IgniteExProcessProxy get(String gridName) { - return gridProxies.get(gridName); - } - - /** - * Kill all running processes. - * - * @throws Exception if failed. - */ - public static void killAll() throws Exception { - for (IgniteExProcessProxy ignite : gridProxies.values()) - try { - ignite.getProcess().kill(); - } - catch (Exception e) { - U.error(ignite.log, "Killing failed.", e); - } - - gridProxies.clear(); - } - - /** - * @return Local JVM grid instance. - */ - public Ignite localJvmGrid() { - return locJvmGrid; - } - - /** - * @return Grid id. - */ - public UUID getId() { - return id; - } - - /** {@inheritDoc} */ - @Override public String name() { - return cfg.getGridName(); - } - - /** {@inheritDoc} */ - @Override public IgniteLogger log() { - return log; - } - - /** {@inheritDoc} */ - @Override public IgniteConfiguration configuration() { - return cfg; - } - - /** {@inheritDoc} */ - @Override public <K extends GridCacheUtilityKey, V> IgniteInternalCache<K, V> utilityCache() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Nullable @Override public <K, V> IgniteInternalCache<K, V> cachex(@Nullable String name) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Nullable @Override public <K, V> IgniteInternalCache<K, V> cachex() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public Collection<IgniteInternalCache<?, ?>> cachesx( - @Nullable IgnitePredicate<? super IgniteInternalCache<?, ?>>... p) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public boolean eventUserRecordable(int type) { - return false; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public boolean allEventsUserRecordable(int[] types) { - return false; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public boolean isJmxRemoteEnabled() { - return false; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public boolean isRestartEnabled() { - return false; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Nullable @Override public IgniteFileSystem igfsx(@Nullable String name) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public Hadoop hadoop() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteClusterEx cluster() { - return (IgniteClusterEx)locJvmGrid.cluster(); - } - - /** {@inheritDoc} */ - @Nullable @Override public String latestVersion() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public ClusterNode localNode() { - return F.first(compute.broadcast(new IgniteClosureX<Object, ClusterNode>() { - @Override public ClusterNode applyx(Object o) { - return ((IgniteEx)Ignition.ignite(id)).localNode(); - } - }, null)); - } - - /** {@inheritDoc} */ - @Override public GridKernalContext context() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteCompute compute() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteCompute compute(ClusterGroup grp) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteMessaging message() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteMessaging message(ClusterGroup grp) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteEvents events() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteEvents events(ClusterGroup grp) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteServices services() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteServices services(ClusterGroup grp) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public ExecutorService executorService() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public ExecutorService executorService(ClusterGroup grp) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteProductVersion version() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteScheduler scheduler() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> createCache(CacheConfiguration<K, V> cacheCfg) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> createCache(String cacheName) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> getOrCreateCache(CacheConfiguration<K, V> cacheCfg) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> getOrCreateCache(String cacheName) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> void addCacheConfiguration(CacheConfiguration<K, V> cacheCfg) { - // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> createCache(CacheConfiguration<K, V> cacheCfg, - NearCacheConfiguration<K, V> nearCfg) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> getOrCreateCache(CacheConfiguration<K, V> cacheCfg, - NearCacheConfiguration<K, V> nearCfg) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> createNearCache(@Nullable String cacheName, NearCacheConfiguration<K, V> nearCfg) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> getOrCreateNearCache(@Nullable String cacheName, - NearCacheConfiguration<K, V> nearCfg) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public void destroyCache(String cacheName) { - // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteCache<K, V> cache(@Nullable final String name) { - return new IgniteCacheProcessProxy(name, this); - } - - /** {@inheritDoc} */ - @Override public IgniteTransactions transactions() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <K, V> IgniteDataStreamer<K, V> dataStreamer(@Nullable String cacheName) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteFileSystem fileSystem(String name) { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public Collection<IgniteFileSystem> fileSystems() { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteAtomicSequence atomicSequence(String name, long initVal, boolean create) throws IgniteException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteAtomicLong atomicLong(String name, long initVal, boolean create) throws IgniteException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <T> IgniteAtomicReference<T> atomicReference(String name, @Nullable T initVal, - boolean create) throws IgniteException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <T, S> IgniteAtomicStamped<T, S> atomicStamped(String name, @Nullable T initVal, @Nullable S initStamp, - boolean create) throws IgniteException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public IgniteCountDownLatch countDownLatch(String name, int cnt, boolean autoDel, - boolean create) throws IgniteException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <T> IgniteQueue<T> queue(String name, int cap, - @Nullable CollectionConfiguration cfg) throws IgniteException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <T> IgniteSet<T> set(String name, @Nullable CollectionConfiguration cfg) throws IgniteException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public <T extends IgnitePlugin> T plugin(String name) throws PluginNotFoundException { - return null; // TODO: CODE: implement. - } - - /** {@inheritDoc} */ - @Override public void close() throws IgniteException { - try { - getProcess().kill(); - } - catch (Exception e) { - e.printStackTrace(); - } - } - - /** {@inheritDoc} */ - @Override public <K> Affinity<K> affinity(String cacheName) { - return null; // TODO: CODE: implement. - } - - public GridJavaProcess getProcess() { - return proc; - } - - public static Ignite grid(@Nullable String name) { -// IgniteNamedInstance grid = name != null ? grids.get(name) : dfltGrid; - - Ignite res = gridProxies.get(name); - - if (res == null) - throw new IgniteIllegalStateException("Grid instance was not properly started " + - "or was already stopped: " + name); - - return res; - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63e98328/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteProcessProxy.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteProcessProxy.java new file mode 100644 index 0000000..538f316 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/multijvm/framework/IgniteProcessProxy.java @@ -0,0 +1,492 @@ +/* + * 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.processors.cache.multijvm.framework; + +import org.apache.ignite.*; +import org.apache.ignite.cache.affinity.*; +import org.apache.ignite.cluster.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.events.*; +import org.apache.ignite.internal.*; +import org.apache.ignite.internal.cluster.*; +import org.apache.ignite.internal.processors.cache.*; +import org.apache.ignite.internal.processors.hadoop.*; +import org.apache.ignite.internal.util.*; +import org.apache.ignite.internal.util.lang.*; +import org.apache.ignite.internal.util.typedef.*; +import org.apache.ignite.internal.util.typedef.internal.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.plugin.*; +import org.jetbrains.annotations.*; + +import java.util.*; +import java.util.concurrent.*; + +/** + * Ignite proxy for ignite instance at another JVM. + */ +public class IgniteProcessProxy implements IgniteEx { + /** Grid proxies. */ + private transient static final Map<String, IgniteProcessProxy> gridProxies = new HashMap<>(); + + /** Jvm process with ignite instance. */ + private transient final GridJavaProcess proc; + + /** Configuration. */ + private transient final IgniteConfiguration cfg; + + /** Local jvm grid. */ + private transient final Ignite locJvmGrid; + + /** Logger. */ + private transient final IgniteLogger log; + + /** Grid id. */ + private final UUID id = UUID.randomUUID(); + + /** Compute. */ + private transient final IgniteCompute compute; + + /** Remote ignite instance started latch. */ + private transient final CountDownLatch rmtNodeStartedLatch = new CountDownLatch(1); + + /** + * @param cfg Configuration. + * @param log Logger. + * @param locJvmGrid Local jvm grid. + */ + public IgniteProcessProxy(final IgniteConfiguration cfg, final IgniteLogger log, final Ignite locJvmGrid) + throws Exception { + this.cfg = cfg; + this.locJvmGrid = locJvmGrid; + this.log = log.getLogger("jvm-" + id.toString().substring(0, id.toString().indexOf('-'))); + + String cfgFileName = IgniteNodeRunner.storeToFile(cfg.setNodeId(id)); + + List<String> jvmArgs = U.jvmArgs(); + + List<String> filteredJvmArgs = new ArrayList<>(); + + for (String arg : jvmArgs) { + if(!arg.toLowerCase().startsWith("-agentlib")) + filteredJvmArgs.add(arg); + } + + locJvmGrid.events().localListen(new IgnitePredicateX<Event>() { + @Override public boolean applyx(Event e) { + if (((DiscoveryEvent)e).eventNode().id().equals(id)) { + rmtNodeStartedLatch.countDown(); + + return false; + } + + return true; + } + }, EventType.EVT_NODE_JOINED); + + proc = GridJavaProcess.exec( + IgniteNodeRunner.class, + cfgFileName, // Params. + this.log, + // Optional closure to be called each time wrapped process prints line to system.out or system.err. + new IgniteInClosure<String>() { + @Override public void apply(String s) { + IgniteProcessProxy.this.log.info(s); + } + }, + null, + filteredJvmArgs, // JVM Args. + System.getProperty("surefire.test.class.path") + ); + + assert rmtNodeStartedLatch.await(30, TimeUnit.SECONDS): "Remote node with id=" + id + " didn't join."; + + gridProxies.put(cfg.getGridName(), this); + + compute = locJvmGrid.compute(locJvmGrid.cluster().forNodeId(id)); + } + + /** + * @param gridName Grid name. + * @return Instance by name or <code>null</code>. + */ + public static IgniteProcessProxy get(String gridName) { + return gridProxies.get(gridName); + } + + /** + * Kill all running processes. + * + * @throws Exception if failed. + */ + public static void killAll() throws Exception { + for (IgniteProcessProxy ignite : gridProxies.values()) + try { + ignite.getProcess().kill(); + } + catch (Exception e) { + U.error(ignite.log, "Killing failed.", e); + } + + gridProxies.clear(); + } + + /** + * @return Local JVM grid instance. + */ + public Ignite localJvmGrid() { + return locJvmGrid; + } + + /** + * @return Grid id. + */ + public UUID getId() { + return id; + } + + /** {@inheritDoc} */ + @Override public String name() { + return cfg.getGridName(); + } + + /** {@inheritDoc} */ + @Override public IgniteLogger log() { + return log; + } + + /** {@inheritDoc} */ + @Override public IgniteConfiguration configuration() { + return cfg; + } + + /** {@inheritDoc} */ + @Override public <K extends GridCacheUtilityKey, V> IgniteInternalCache<K, V> utilityCache() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Nullable @Override public <K, V> IgniteInternalCache<K, V> cachex(@Nullable String name) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Nullable @Override public <K, V> IgniteInternalCache<K, V> cachex() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public Collection<IgniteInternalCache<?, ?>> cachesx( + @Nullable IgnitePredicate<? super IgniteInternalCache<?, ?>>... p) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public boolean eventUserRecordable(int type) { + return false; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public boolean allEventsUserRecordable(int[] types) { + return false; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public boolean isJmxRemoteEnabled() { + return false; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public boolean isRestartEnabled() { + return false; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Nullable @Override public IgniteFileSystem igfsx(@Nullable String name) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public Hadoop hadoop() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteClusterEx cluster() { + return (IgniteClusterEx)locJvmGrid.cluster(); + } + + /** {@inheritDoc} */ + @Nullable @Override public String latestVersion() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public ClusterNode localNode() { + return F.first(compute.broadcast(new IgniteClosureX<Object, ClusterNode>() { + @Override public ClusterNode applyx(Object o) { + return ((IgniteEx)Ignition.ignite(id)).localNode(); + } + }, null)); + } + + /** {@inheritDoc} */ + @Override public GridKernalContext context() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteCompute compute() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteCompute compute(ClusterGroup grp) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteMessaging message() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteMessaging message(ClusterGroup grp) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteEvents events() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteEvents events(ClusterGroup grp) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteServices services() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteServices services(ClusterGroup grp) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public ExecutorService executorService() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public ExecutorService executorService(ClusterGroup grp) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteProductVersion version() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteScheduler scheduler() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> createCache(CacheConfiguration<K, V> cacheCfg) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> createCache(String cacheName) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> getOrCreateCache(CacheConfiguration<K, V> cacheCfg) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> getOrCreateCache(String cacheName) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> void addCacheConfiguration(CacheConfiguration<K, V> cacheCfg) { + // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> createCache(CacheConfiguration<K, V> cacheCfg, + NearCacheConfiguration<K, V> nearCfg) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> getOrCreateCache(CacheConfiguration<K, V> cacheCfg, + NearCacheConfiguration<K, V> nearCfg) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> createNearCache(@Nullable String cacheName, NearCacheConfiguration<K, V> nearCfg) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> getOrCreateNearCache(@Nullable String cacheName, + NearCacheConfiguration<K, V> nearCfg) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public void destroyCache(String cacheName) { + // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteCache<K, V> cache(@Nullable final String name) { + return new IgniteCacheProcessProxy(name, this); + } + + /** {@inheritDoc} */ + @Override public IgniteTransactions transactions() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteDataStreamer<K, V> dataStreamer(@Nullable String cacheName) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteFileSystem fileSystem(String name) { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public Collection<IgniteFileSystem> fileSystems() { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteAtomicSequence atomicSequence(String name, long initVal, boolean create) throws IgniteException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteAtomicLong atomicLong(String name, long initVal, boolean create) throws IgniteException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <T> IgniteAtomicReference<T> atomicReference(String name, @Nullable T initVal, + boolean create) throws IgniteException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <T, S> IgniteAtomicStamped<T, S> atomicStamped(String name, @Nullable T initVal, @Nullable S initStamp, + boolean create) throws IgniteException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public IgniteCountDownLatch countDownLatch(String name, int cnt, boolean autoDel, + boolean create) throws IgniteException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <T> IgniteQueue<T> queue(String name, int cap, + @Nullable CollectionConfiguration cfg) throws IgniteException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <T> IgniteSet<T> set(String name, @Nullable CollectionConfiguration cfg) throws IgniteException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public <T extends IgnitePlugin> T plugin(String name) throws PluginNotFoundException { + return null; // TODO: CODE: implement. + } + + /** {@inheritDoc} */ + @Override public void close() throws IgniteException { + try { + getProcess().kill(); + } + catch (Exception e) { + e.printStackTrace(); + } + } + + /** {@inheritDoc} */ + @Override public <K> Affinity<K> affinity(String cacheName) { + return null; // TODO: CODE: implement. + } + + public GridJavaProcess getProcess() { + return proc; + } + + public static Ignite grid(@Nullable String name) { +// IgniteNamedInstance grid = name != null ? grids.get(name) : dfltGrid; + + Ignite res = gridProxies.get(name); + + if (res == null) + throw new IgniteIllegalStateException("Grid instance was not properly started " + + "or was already stopped: " + name); + + return res; + } + + // TODO delete or use. +// public <K, V> GridCacheAdapter<K, V> remoteInternalCache() { +// return (GridCacheAdapter<K, V>)compute.call(new MyCallable(id)); +// } +// +// /** +// */ +// private static class MyCallable implements IgniteCallable<Object> { +// private UUID id; +// +// public MyCallable() { +// // No-op. +// } +// +// public MyCallable(UUID id) { +// this.id = id; +// } +// +// public void setId(UUID id) { +// this.id = id; +// } +// +// @Override public Object call() throws Exception { +// return ((IgniteKernal)Ignition.ignite(id)).internalCache(); +// } +// } +}
