http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServer.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServer.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServer.java index ac5b3f3..c8dc7ec 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServer.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServer.java @@ -220,7 +220,7 @@ public class GridGgfsServer { } /** {@inheritDoc} */ - @Override protected void body() throws InterruptedException, GridInterruptedException { + @Override protected void body() throws InterruptedException, InternalInterruptedException { try { GridGgfsDataInputStream dis = new GridGgfsDataInputStream(endpoint.inputStream()); @@ -249,7 +249,7 @@ public class GridGgfsServer { GridGgfsMessage msg = marsh.unmarshall(cmd, hdr, dis); - IgniteFuture<GridGgfsMessage> fut = hnd.handleAsync(ses, msg, dis); + InternalFuture<GridGgfsMessage> fut = hnd.handleAsync(ses, msg, dis); // If fut is null, no response is required. if (fut != null) { @@ -280,8 +280,8 @@ public class GridGgfsServer { } } else { - fut.listenAsync(new CIX1<IgniteFuture<GridGgfsMessage>>() { - @Override public void applyx(IgniteFuture<GridGgfsMessage> fut) { + fut.listenAsync(new CIX1<InternalFuture<GridGgfsMessage>>() { + @Override public void applyx(InternalFuture<GridGgfsMessage> fut) { GridGgfsMessage res; try { @@ -381,7 +381,7 @@ public class GridGgfsServer { } /** {@inheritDoc} */ - @Override protected void body() throws InterruptedException, GridInterruptedException { + @Override protected void body() throws InterruptedException, InternalInterruptedException { try { while (!Thread.currentThread().isInterrupted()) { GridIpcEndpoint client = srvEndpoint.accept();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServerHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServerHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServerHandler.java index 5927d01..433a611 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServerHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/GridGgfsServerHandler.java @@ -30,7 +30,7 @@ public interface GridGgfsServerHandler { * @return Future that will be completed when response is ready or {@code null} if no * response is required. */ - @Nullable public IgniteFuture<GridGgfsMessage> handleAsync(GridGgfsClientSession ses, + @Nullable public InternalFuture<GridGgfsMessage> handleAsync(GridGgfsClientSession ses, GridGgfsMessage msg, DataInput in); /** http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/IgniteFsOutputStreamImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/IgniteFsOutputStreamImpl.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/IgniteFsOutputStreamImpl.java index 04ab237..7d124c7 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/IgniteFsOutputStreamImpl.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/ggfs/IgniteFsOutputStreamImpl.java @@ -12,7 +12,6 @@ package org.gridgain.grid.kernal.processors.ggfs; import org.apache.ignite.*; import org.apache.ignite.fs.*; import org.apache.ignite.lang.*; -import org.gridgain.grid.*; import org.gridgain.grid.kernal.processors.task.*; import org.gridgain.grid.util.typedef.internal.*; import org.gridgain.grid.util.future.*; @@ -61,7 +60,7 @@ class IgniteFsOutputStreamImpl extends IgniteFsOutputStreamAdapter { private int remainderDataLen; /** Write completion future. */ - private final IgniteFuture<Boolean> writeCompletionFut; + private final InternalFuture<Boolean> writeCompletionFut; /** GGFS mode. */ private final IgniteFsMode mode; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopNoopProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopNoopProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopNoopProcessor.java index 6cc26ca..edc2f40 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopNoopProcessor.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopNoopProcessor.java @@ -42,7 +42,7 @@ public class GridHadoopNoopProcessor extends GridHadoopProcessorAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> submit(GridHadoopJobId jobId, GridHadoopJobInfo jobInfo) { + @Override public InternalFuture<?> submit(GridHadoopJobId jobId, GridHadoopJobInfo jobInfo) { return new GridFinishedFutureEx<>(new IgniteCheckedException("Hadoop is not available.")); } @@ -57,7 +57,7 @@ public class GridHadoopNoopProcessor extends GridHadoopProcessorAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> finishFuture(GridHadoopJobId jobId) throws IgniteCheckedException { + @Override public InternalFuture<?> finishFuture(GridHadoopJobId jobId) throws IgniteCheckedException { return null; } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopProcessorAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopProcessorAdapter.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopProcessorAdapter.java index a390081..370370a 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopProcessorAdapter.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/hadoop/GridHadoopProcessorAdapter.java @@ -48,7 +48,7 @@ public abstract class GridHadoopProcessorAdapter extends GridProcessorAdapter { * @param jobInfo Job info to submit. * @return Execution future. */ - public abstract IgniteFuture<?> submit(GridHadoopJobId jobId, GridHadoopJobInfo jobInfo); + public abstract InternalFuture<?> submit(GridHadoopJobId jobId, GridHadoopJobInfo jobInfo); /** * Gets Hadoop job execution status. @@ -75,7 +75,7 @@ public abstract class GridHadoopProcessorAdapter extends GridProcessorAdapter { * @return Job finish future or {@code null}. * @throws IgniteCheckedException If failed. */ - public abstract IgniteFuture<?> finishFuture(GridHadoopJobId jobId) throws IgniteCheckedException; + public abstract InternalFuture<?> finishFuture(GridHadoopJobId jobId) throws IgniteCheckedException; /** * Kills job. http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/job/GridJobWorker.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/job/GridJobWorker.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/job/GridJobWorker.java index 9d100e3..0ff772a 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/job/GridJobWorker.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/job/GridJobWorker.java @@ -482,7 +482,7 @@ public class GridJobWorker extends GridWorker implements GridTimeoutObject { } } catch (IgniteCheckedException e) { - if (sysStopping && e.hasCause(GridInterruptedException.class, InterruptedException.class)) { + if (sysStopping && e.hasCause(InternalInterruptedException.class, InterruptedException.class)) { ex = handleThrowable(e); assert ex != null; @@ -553,7 +553,7 @@ public class GridJobWorker extends GridWorker implements GridTimeoutObject { ex = new ComputeUserUndeclaredException(msg, e); } - else if (sysStopping && X.hasCause(e, InterruptedException.class, GridInterruptedException.class)) { + else if (sysStopping && X.hasCause(e, InterruptedException.class, InternalInterruptedException.class)) { msg = "Job got interrupted due to system stop (will attempt failover)."; ex = new ComputeExecutionRejectedException(e); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/query/GridQueryProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/query/GridQueryProcessor.java index 6bc0235..0b746bb 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/query/GridQueryProcessor.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/query/GridQueryProcessor.java @@ -160,7 +160,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { * @param valTypeName Value type name. * @return Future that will be completed when rebuilding of all indexes is finished. */ - public IgniteFuture<?> rebuildIndexes(@Nullable final String space, String valTypeName) { + public InternalFuture<?> rebuildIndexes(@Nullable final String space, String valTypeName) { if (!busyLock.enterBusy()) throw new IllegalStateException("Failed to rebuild indexes (grid is stopping)."); @@ -177,7 +177,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { * @param desc Type descriptor. * @return Future that will be completed when rebuilding of all indexes is finished. */ - private IgniteFuture<?> rebuildIndexes(@Nullable final String space, @Nullable final TypeDescriptor desc) { + private InternalFuture<?> rebuildIndexes(@Nullable final String space, @Nullable final TypeDescriptor desc) { if (idx == null) return new GridFinishedFuture<>(ctx, new IgniteCheckedException("Indexing is disabled.")); @@ -217,7 +217,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { * @return Future that will be completed when rebuilding of all indexes is finished. */ @SuppressWarnings("unchecked") - public IgniteFuture<?> rebuildAllIndexes() { + public InternalFuture<?> rebuildAllIndexes() { if (!busyLock.enterBusy()) throw new IllegalStateException("Failed to get space size (grid is stopping)."); @@ -225,7 +225,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { GridCompoundFuture<?, ?> fut = new GridCompoundFuture<Object, Object>(ctx); for (Map.Entry<TypeId, TypeDescriptor> e : types.entrySet()) - fut.add((IgniteFuture)rebuildIndexes(e.getKey().space, e.getValue())); + fut.add((InternalFuture)rebuildIndexes(e.getKey().space, e.getValue())); fut.markInitialized(); @@ -1296,9 +1296,9 @@ public class GridQueryProcessor extends GridProcessorAdapter { /** * @return Waits for initialization. - * @throws GridInterruptedException If thread is interrupted. + * @throws InternalInterruptedException If thread is interrupted. */ - boolean await() throws GridInterruptedException { + boolean await() throws InternalInterruptedException { return initializer.await(); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProcessor.java index ae20c07..d32f595 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProcessor.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProcessor.java @@ -71,7 +71,7 @@ public class GridRestProcessor extends GridProcessorAdapter { return handleAsync(req).get(); } - @Override public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req) { return handleAsync0(req); } }; @@ -80,7 +80,7 @@ public class GridRestProcessor extends GridProcessorAdapter { * @param req Request. * @return Future. */ - private IgniteFuture<GridRestResponse> handleAsync0(final GridRestRequest req) { + private InternalFuture<GridRestResponse> handleAsync0(final GridRestRequest req) { if (!busyLock.tryReadLock()) return new GridFinishedFuture<>(ctx, new IgniteCheckedException("Failed to handle request (received request while stopping grid).")); @@ -93,10 +93,10 @@ public class GridRestProcessor extends GridProcessorAdapter { GridWorker w = new GridWorker(ctx.gridName(), "rest-proc-worker", log) { @Override protected void body() { try { - IgniteFuture<GridRestResponse> res = handleRequest(req); + InternalFuture<GridRestResponse> res = handleRequest(req); - res.listenAsync(new IgniteInClosure<IgniteFuture<GridRestResponse>>() { - @Override public void apply(IgniteFuture<GridRestResponse> f) { + res.listenAsync(new IgniteInClosure<InternalFuture<GridRestResponse>>() { + @Override public void apply(InternalFuture<GridRestResponse> f) { try { fut.onDone(f.get()); } @@ -142,7 +142,7 @@ public class GridRestProcessor extends GridProcessorAdapter { * @param req Request. * @return Future. */ - private IgniteFuture<GridRestResponse> handleRequest(final GridRestRequest req) { + private InternalFuture<GridRestResponse> handleRequest(final GridRestRequest req) { if (startLatch.getCount() > 0) { try { startLatch.await(); @@ -187,7 +187,7 @@ public class GridRestProcessor extends GridProcessorAdapter { GridRestCommandHandler hnd = handlers.get(req.command()); - IgniteFuture<GridRestResponse> res = hnd == null ? null : hnd.handleAsync(req); + InternalFuture<GridRestResponse> res = hnd == null ? null : hnd.handleAsync(req); if (res == null) return new GridFinishedFuture<>(ctx, @@ -195,8 +195,8 @@ public class GridRestProcessor extends GridProcessorAdapter { final GridSecurityContext subjCtx0 = subjCtx; - return res.chain(new C1<IgniteFuture<GridRestResponse>, GridRestResponse>() { - @Override public GridRestResponse apply(IgniteFuture<GridRestResponse> f) { + return res.chain(new C1<InternalFuture<GridRestResponse>, GridRestResponse>() { + @Override public GridRestResponse apply(InternalFuture<GridRestResponse> f) { GridRestResponse res; try { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProtocolHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProtocolHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProtocolHandler.java index e05a6b8..504fa08 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProtocolHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/GridRestProtocolHandler.java @@ -28,5 +28,5 @@ public interface GridRestProtocolHandler { * @param req Request. * @return Future. */ - public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req); + public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/GridRestCommandHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/GridRestCommandHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/GridRestCommandHandler.java index 8614e7e..0a24495 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/GridRestCommandHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/GridRestCommandHandler.java @@ -28,5 +28,5 @@ public interface GridRestCommandHandler { * @param req Request. * @return Future. */ - public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req); + public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheCommandHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheCommandHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheCommandHandler.java index 1bab28a..312eef3 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheCommandHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheCommandHandler.java @@ -13,7 +13,6 @@ import org.apache.ignite.*; import org.apache.ignite.cluster.*; import org.apache.ignite.lang.*; import org.apache.ignite.resources.*; -import org.gridgain.grid.*; import org.gridgain.grid.cache.*; import org.gridgain.grid.cache.datastructures.*; import org.gridgain.grid.kernal.*; @@ -117,7 +116,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<GridRestResponse> handleAsync(final GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(final GridRestRequest req) { assert req instanceof GridRestCacheRequest : "Invalid command for topology handler: " + req; assert SUPPORTED_COMMANDS.contains(req.command()); @@ -143,7 +142,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { final Long ttl = req0.ttl(); - IgniteFuture<GridRestResponse> fut; + InternalFuture<GridRestResponse> fut; switch (cmd) { case CACHE_GET: { @@ -327,7 +326,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { * @return Operation result in future. * @throws IgniteCheckedException If failed */ - private IgniteFuture<GridRestResponse> executeCommand( + private InternalFuture<GridRestResponse> executeCommand( @Nullable UUID destId, UUID clientId, final String cacheName, @@ -372,7 +371,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { * @return Operation result in future. * @throws IgniteCheckedException If failed */ - private IgniteFuture<GridRestResponse> executeCommand( + private InternalFuture<GridRestResponse> executeCommand( @Nullable UUID destId, UUID clientId, final String cacheName, @@ -407,7 +406,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { * @return Future of operation result. * @throws IgniteCheckedException In case of error. */ - private static IgniteFuture<?> incrementOrDecrement(GridCacheProjection<Object, Object> cache, String key, + private static InternalFuture<?> incrementOrDecrement(GridCacheProjection<Object, Object> cache, String key, GridRestCacheRequest req, final boolean decr) throws IgniteCheckedException { assert cache != null; assert key != null; @@ -441,7 +440,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { * @return Future of operation result. * @throws IgniteCheckedException In case of any exception. */ - private static IgniteFuture<?> appendOrPrepend( + private static InternalFuture<?> appendOrPrepend( final GridKernalContext ctx, final GridCacheProjection<Object, Object> cache, final Object key, GridRestCacheRequest req, final boolean prepend) throws IgniteCheckedException { @@ -545,10 +544,10 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { * @param key Affinity key for previous operation. * @return Rest response. */ - private static IgniteClosure<IgniteFuture<?>, GridRestResponse> resultWrapper( + private static IgniteClosure<InternalFuture<?>, GridRestResponse> resultWrapper( final GridCacheProjection<Object, Object> c, @Nullable final Object key) { - return new CX1<IgniteFuture<?>, GridRestResponse>() { - @Override public GridRestResponse applyx(IgniteFuture<?> f) throws IgniteCheckedException { + return new CX1<InternalFuture<?>, GridRestResponse>() { + @Override public GridRestResponse applyx(InternalFuture<?> f) throws IgniteCheckedException { GridCacheRestResponse resp = new GridCacheRestResponse(); resp.setResponse(f.get()); @@ -607,7 +606,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { /** * Fixed result closure. */ - private static final class FixedResult extends CX1<IgniteFuture<?>, Object> { + private static final class FixedResult extends CX1<InternalFuture<?>, Object> { /** */ private static final long serialVersionUID = 0L; @@ -622,7 +621,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public Object applyx(IgniteFuture<?> f) throws IgniteCheckedException { + @Override public Object applyx(InternalFuture<?> f) throws IgniteCheckedException { f.get(); return res; @@ -633,7 +632,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { * Type alias. */ private abstract static class CacheCommand - extends IgniteClosure2X<GridCacheProjection<Object, Object>, GridKernalContext, IgniteFuture<?>> { + extends IgniteClosure2X<GridCacheProjection<Object, Object>, GridKernalContext, InternalFuture<?>> { /** */ private static final long serialVersionUID = 0L; @@ -644,7 +643,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { * Type alias. */ private abstract static class CacheProjectionCommand - extends IgniteClosure2X<GridCacheProjection<Object, Object>, GridKernalContext, IgniteFuture<?>> { + extends IgniteClosure2X<GridCacheProjection<Object, Object>, GridKernalContext, InternalFuture<?>> { /** */ private static final long serialVersionUID = 0L; @@ -778,7 +777,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { return c.getAsync(key); } } @@ -799,7 +798,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { return c.getAllAsync(keys); } } @@ -820,7 +819,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { return c.putAllAsync(map).chain(new FixedResult(true)); } } @@ -841,7 +840,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { return c.removexAsync(key); } } @@ -862,7 +861,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { return (F.isEmpty(keys) ? c.removeAllAsync() : c.removeAllAsync(keys)) .chain(new FixedResult(true)); } @@ -894,7 +893,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { return exp == null && val == null ? c.removexAsync(key) : exp == null ? c.putxIfAbsentAsync(key, val) : val == null ? c.removeAsync(key, exp) : @@ -928,7 +927,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { if (ttl != null) { GridCacheEntry<Object, Object> entry = c.entry(key); @@ -971,7 +970,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { GridCacheEntry<Object, Object> entry = c.entry(key); if (entry != null) { @@ -1011,7 +1010,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { GridCacheEntry<Object, Object> entry = c.entry(key); if (entry != null) { @@ -1046,7 +1045,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) throws IgniteCheckedException { return incrementOrDecrement(c, (String)key, req, false); } @@ -1073,7 +1072,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) throws IgniteCheckedException { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) throws IgniteCheckedException { return incrementOrDecrement(c, (String)key, req, true); } } @@ -1099,7 +1098,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) throws IgniteCheckedException { return appendOrPrepend(ctx, c, key, req, false); } @@ -1126,7 +1125,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) throws IgniteCheckedException { return appendOrPrepend(ctx, c, key, req, true); } @@ -1138,7 +1137,7 @@ public class GridCacheCommandHandler extends GridRestCommandHandlerAdapter { private static final long serialVersionUID = 0L; /** {@inheritDoc} */ - @Override public IgniteFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { + @Override public InternalFuture<?> applyx(GridCacheProjection<Object, Object> c, GridKernalContext ctx) { GridCacheMetrics metrics = c.cache().metrics(); assert metrics != null; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheQueryCommandHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheQueryCommandHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheQueryCommandHandler.java index 3303b41..f15691e 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheQueryCommandHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/cache/GridCacheQueryCommandHandler.java @@ -13,7 +13,6 @@ import org.apache.ignite.*; import org.apache.ignite.cluster.*; import org.apache.ignite.lang.*; import org.apache.ignite.resources.*; -import org.gridgain.grid.*; import org.gridgain.grid.cache.*; import org.gridgain.grid.cache.query.*; import org.gridgain.grid.kernal.*; @@ -62,7 +61,7 @@ public class GridCacheQueryCommandHandler extends GridRestCommandHandlerAdapter } /** {@inheritDoc} */ - @Override public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req) { assert req instanceof GridRestCacheQueryRequest; assert SUPPORTED_COMMANDS.contains(req.command()); @@ -107,7 +106,7 @@ public class GridCacheQueryCommandHandler extends GridRestCommandHandlerAdapter * @param c Closure to execute. * @return Execution future. */ - private IgniteFuture<GridRestResponse> execute(UUID destId, String cacheName, Callable<GridRestResponse> c) { + private InternalFuture<GridRestResponse> execute(UUID destId, String cacheName, Callable<GridRestResponse> c) { boolean locExec = destId == null || destId.equals(ctx.localNodeId()) || replicatedCacheAvailable(cacheName); if (locExec) @@ -136,16 +135,16 @@ public class GridCacheQueryCommandHandler extends GridRestCommandHandlerAdapter * @param c Closure to execute. * @return Execution future. */ - private IgniteFuture<GridRestResponse> broadcast(String cacheName, Callable<Object> c) { + private InternalFuture<GridRestResponse> broadcast(String cacheName, Callable<Object> c) { IgniteCompute comp = ctx.grid().compute(ctx.grid().forCache(cacheName)).withNoFailover().enableAsync(); try { comp.broadcast(c); - IgniteFuture<Collection<Object>> fut = comp.future(); + InternalFuture<Collection<Object>> fut = comp.future(); - return fut.chain(new C1<IgniteFuture<Collection<Object>>, GridRestResponse>() { - @Override public GridRestResponse apply(IgniteFuture<Collection<Object>> fut) { + return fut.chain(new C1<InternalFuture<Collection<Object>>, GridRestResponse>() { + @Override public GridRestResponse apply(InternalFuture<Collection<Object>> fut) { try { fut.get(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/log/GridLogCommandHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/log/GridLogCommandHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/log/GridLogCommandHandler.java index e8ecf93..470f66d 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/log/GridLogCommandHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/log/GridLogCommandHandler.java @@ -74,7 +74,7 @@ public class GridLogCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req) { assert req instanceof GridRestLogRequest : "Invalid command for topology handler: " + req; assert SUPPORTED_COMMANDS.contains(req.command()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/metadata/GridPortableMetadataHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/metadata/GridPortableMetadataHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/metadata/GridPortableMetadataHandler.java index 8e547b1..c763589 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/metadata/GridPortableMetadataHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/metadata/GridPortableMetadataHandler.java @@ -12,7 +12,6 @@ package org.gridgain.grid.kernal.processors.rest.handlers.metadata; import org.apache.ignite.*; import org.apache.ignite.lang.*; import org.apache.ignite.portables.*; -import org.gridgain.grid.*; import org.gridgain.grid.kernal.*; import org.gridgain.grid.kernal.processors.rest.*; import org.gridgain.grid.kernal.processors.rest.client.message.*; @@ -48,7 +47,7 @@ public class GridPortableMetadataHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req) { assert SUPPORTED_COMMANDS.contains(req.command()) : req.command(); try { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/task/GridTaskCommandHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/task/GridTaskCommandHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/task/GridTaskCommandHandler.java index 77cf114..02dd6c1 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/task/GridTaskCommandHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/task/GridTaskCommandHandler.java @@ -16,7 +16,6 @@ import org.apache.ignite.events.*; import org.apache.ignite.lang.*; import org.apache.ignite.portables.*; import org.apache.ignite.resources.*; -import org.gridgain.grid.*; import org.gridgain.grid.kernal.*; import org.gridgain.grid.kernal.managers.communication.*; import org.gridgain.grid.kernal.managers.eventstorage.*; @@ -121,7 +120,7 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req) { try { return handleAsyncUnsafe(req); } @@ -141,7 +140,7 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter { * @return Future. * @throws IgniteCheckedException On any handling exception. */ - private IgniteFuture<GridRestResponse> handleAsyncUnsafe(final GridRestRequest req) throws IgniteCheckedException { + private InternalFuture<GridRestResponse> handleAsyncUnsafe(final GridRestRequest req) throws IgniteCheckedException { assert req instanceof GridRestTaskRequest : "Invalid command for topology handler: " + req; assert SUPPORTED_COMMANDS.contains(req.command()); @@ -178,7 +177,7 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter { final UUID clientId = req.clientId(); - final ComputeTaskFuture<Object> taskFut; + final InternalComputeTaskFuture<Object> taskFut; if (locExec) { ClusterGroup prj = ctx.grid().forSubjectId(clientId); @@ -219,8 +218,8 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter { fut.onDone(res); } - taskFut.listenAsync(new IgniteInClosure<IgniteFuture<Object>>() { - @Override public void apply(IgniteFuture<Object> f) { + taskFut.listenAsync(new IgniteInClosure<InternalFuture<Object>>() { + @Override public void apply(InternalFuture<Object> f) { try { TaskDescriptor desc; @@ -240,9 +239,9 @@ public class GridTaskCommandHandler extends GridRestCommandHandlerAdapter { } if (async && locExec) { - assert taskFut instanceof ComputeTaskFuture; + assert taskFut instanceof InternalComputeTaskFuture; - IgniteUuid tid = ((ComputeTaskFuture)taskFut).getTaskSession().getId(); + IgniteUuid tid = ((InternalComputeTaskFuture)taskFut).getTaskSession().getId(); taskDescs.put(tid, desc); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/top/GridTopologyCommandHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/top/GridTopologyCommandHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/top/GridTopologyCommandHandler.java index 4116368..7efaa34 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/top/GridTopologyCommandHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/top/GridTopologyCommandHandler.java @@ -51,7 +51,7 @@ public class GridTopologyCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req) { assert req instanceof GridRestTopologyRequest : "Invalid command for topology handler: " + req; assert SUPPORTED_COMMANDS.contains(req.command()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/version/GridVersionCommandHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/version/GridVersionCommandHandler.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/version/GridVersionCommandHandler.java index 864f772..014379e 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/version/GridVersionCommandHandler.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/handlers/version/GridVersionCommandHandler.java @@ -42,7 +42,7 @@ public class GridVersionCommandHandler extends GridRestCommandHandlerAdapter { } /** {@inheritDoc} */ - @Override public IgniteFuture<GridRestResponse> handleAsync(GridRestRequest req) { + @Override public InternalFuture<GridRestResponse> handleAsync(GridRestRequest req) { assert req != null; assert SUPPORTED_COMMANDS.contains(req.command()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpMemcachedNioListener.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpMemcachedNioListener.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpMemcachedNioListener.java index 7b28af5..8e29ceb 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpMemcachedNioListener.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpMemcachedNioListener.java @@ -13,7 +13,6 @@ import org.apache.ignite.*; import org.apache.ignite.lang.*; import org.apache.ignite.marshaller.*; import org.apache.ignite.marshaller.jdk.*; -import org.gridgain.grid.*; import org.gridgain.grid.kernal.*; import org.gridgain.grid.kernal.processors.rest.*; import org.gridgain.grid.kernal.processors.rest.handlers.cache.*; @@ -113,20 +112,20 @@ public class GridTcpMemcachedNioListener extends GridNioServerListenerAdapter<Gr return; } - IgniteFuture<GridRestResponse> lastFut = ses.removeMeta(LAST_FUT.ordinal()); + InternalFuture<GridRestResponse> lastFut = ses.removeMeta(LAST_FUT.ordinal()); if (lastFut != null && lastFut.isDone()) lastFut = null; - IgniteFuture<GridRestResponse> f; + InternalFuture<GridRestResponse> f; if (lastFut == null) f = handleRequest0(ses, req, cmd); else { f = new GridEmbeddedFuture<>( lastFut, - new C2<GridRestResponse, Exception, IgniteFuture<GridRestResponse>>() { - @Override public IgniteFuture<GridRestResponse> apply(GridRestResponse res, Exception e) { + new C2<GridRestResponse, Exception, InternalFuture<GridRestResponse>>() { + @Override public InternalFuture<GridRestResponse> apply(GridRestResponse res, Exception e) { return handleRequest0(ses, req, cmd); } }, @@ -143,7 +142,7 @@ public class GridTcpMemcachedNioListener extends GridNioServerListenerAdapter<Gr * @param cmd Command. * @return Future or {@code null} if processed immediately. */ - @Nullable private IgniteFuture<GridRestResponse> handleRequest0( + @Nullable private InternalFuture<GridRestResponse> handleRequest0( final GridNioSession ses, final GridMemcachedMessage req, final GridTuple3<GridRestCommand, Boolean, Boolean> cmd @@ -158,10 +157,10 @@ public class GridTcpMemcachedNioListener extends GridNioServerListenerAdapter<Gr return null; } - IgniteFuture<GridRestResponse> f = hnd.handleAsync(createRestRequest(req, cmd.get1())); + InternalFuture<GridRestResponse> f = hnd.handleAsync(createRestRequest(req, cmd.get1())); - f.listenAsync(new CIX1<IgniteFuture<GridRestResponse>>() { - @Override public void applyx(IgniteFuture<GridRestResponse> f) throws IgniteCheckedException { + f.listenAsync(new CIX1<InternalFuture<GridRestResponse>>() { + @Override public void applyx(InternalFuture<GridRestResponse> f) throws IgniteCheckedException { GridRestResponse restRes = f.get(); // Handle 'Stat' command (special case because several packets are included in response). http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpRestNioListener.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpRestNioListener.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpRestNioListener.java index f80ab41..a4c85d2 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpRestNioListener.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/rest/protocols/tcp/GridTcpRestNioListener.java @@ -12,7 +12,6 @@ package org.gridgain.grid.kernal.processors.rest.protocols.tcp; import org.apache.ignite.*; import org.apache.ignite.lang.*; import org.gridgain.client.marshaller.*; -import org.gridgain.grid.*; import org.gridgain.grid.kernal.*; import org.gridgain.grid.kernal.processors.rest.*; import org.gridgain.grid.kernal.processors.rest.client.message.*; @@ -170,8 +169,8 @@ public class GridTcpRestNioListener extends GridNioServerListenerAdapter<GridCli final GridRestRequest req = createRestRequest(ses, msg); if (req != null) - hnd.handleAsync(req).listenAsync(new CI1<IgniteFuture<GridRestResponse>>() { - @Override public void apply(IgniteFuture<GridRestResponse> fut) { + hnd.handleAsync(req).listenAsync(new CI1<InternalFuture<GridRestResponse>>() { + @Override public void apply(InternalFuture<GridRestResponse> fut) { GridClientResponse res = new GridClientResponse(); res.requestId(msg.requestId()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java index 46ecf48..6bceeb3 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/service/GridServiceProcessor.java @@ -139,13 +139,13 @@ public class GridServiceProcessor extends GridProcessorAdapter { ManagedServiceConfiguration[] cfgs = ctx.config().getServiceConfiguration(); if (cfgs != null) { - Collection<IgniteFuture<?>> futs = new ArrayList<>(); + Collection<InternalFuture<?>> futs = new ArrayList<>(); for (ManagedServiceConfiguration c : ctx.config().getServiceConfiguration()) futs.add(deploy(c)); // Await for services to deploy. - for (IgniteFuture<?> f : futs) + for (InternalFuture<?> f : futs) f.get(); } @@ -254,7 +254,7 @@ public class GridServiceProcessor extends GridProcessorAdapter { * @param svc Service. * @return Future. */ - public IgniteFuture<?> deployNodeSingleton(ClusterGroup prj, String name, ManagedService svc) { + public InternalFuture<?> deployNodeSingleton(ClusterGroup prj, String name, ManagedService svc) { return deployMultiple(prj, name, svc, 0, 1); } @@ -263,7 +263,7 @@ public class GridServiceProcessor extends GridProcessorAdapter { * @param svc Service. * @return Future. */ - public IgniteFuture<?> deployClusterSingleton(ClusterGroup prj, String name, ManagedService svc) { + public InternalFuture<?> deployClusterSingleton(ClusterGroup prj, String name, ManagedService svc) { return deployMultiple(prj, name, svc, 1, 1); } @@ -274,7 +274,7 @@ public class GridServiceProcessor extends GridProcessorAdapter { * @param maxPerNodeCnt Max per-node count. * @return Future. */ - public IgniteFuture<?> deployMultiple(ClusterGroup prj, String name, ManagedService svc, int totalCnt, + public InternalFuture<?> deployMultiple(ClusterGroup prj, String name, ManagedService svc, int totalCnt, int maxPerNodeCnt) { ManagedServiceConfiguration cfg = new ManagedServiceConfiguration(); @@ -294,7 +294,7 @@ public class GridServiceProcessor extends GridProcessorAdapter { * @param affKey Affinity key. * @return Future. */ - public IgniteFuture<?> deployKeyAffinitySingleton(String name, ManagedService svc, String cacheName, Object affKey) { + public InternalFuture<?> deployKeyAffinitySingleton(String name, ManagedService svc, String cacheName, Object affKey) { A.notNull(affKey, "affKey"); ManagedServiceConfiguration cfg = new ManagedServiceConfiguration(); @@ -313,7 +313,7 @@ public class GridServiceProcessor extends GridProcessorAdapter { * @param cfg Service configuration. * @return Future for deployment. */ - public IgniteFuture<?> deploy(ManagedServiceConfiguration cfg) { + public InternalFuture<?> deploy(ManagedServiceConfiguration cfg) { A.notNull(cfg, "cfg"); validate(cfg); @@ -404,7 +404,7 @@ public class GridServiceProcessor extends GridProcessorAdapter { * @param name Service name. * @return Future. */ - public IgniteFuture<?> cancel(String name) { + public InternalFuture<?> cancel(String name) { while (true) { try { GridFutureAdapter<?> fut = new GridFutureAdapter<>(ctx); @@ -442,8 +442,8 @@ public class GridServiceProcessor extends GridProcessorAdapter { * @return Future. */ @SuppressWarnings("unchecked") - public IgniteFuture<?> cancelAll() { - Collection<IgniteFuture<?>> futs = new ArrayList<>(); + public InternalFuture<?> cancelAll() { + Collection<InternalFuture<?>> futs = new ArrayList<>(); for (GridCacheEntry<Object, Object> e : cache.entrySetx()) { if (!(e.getKey() instanceof GridServiceDeploymentKey)) @@ -807,14 +807,14 @@ public class GridServiceProcessor extends GridProcessorAdapter { try { cp.execute(svcCtx); } - catch (InterruptedException | GridInterruptedException ignore) { + catch (InterruptedException | InternalInterruptedException ignore) { if (log.isDebugEnabled()) log.debug("Service thread was interrupted [name=" + svcCtx.name() + ", execId=" + svcCtx.executionId() + ']'); } catch (IgniteException e) { if (e.hasCause(InterruptedException.class) || - e.hasCause(GridInterruptedException.class)) { + e.hasCause(InternalInterruptedException.class)) { if (log.isDebugEnabled()) log.debug("Service thread was interrupted [name=" + svcCtx.name() + ", execId=" + svcCtx.executionId() + ']'); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/streamer/IgniteStreamerImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/streamer/IgniteStreamerImpl.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/streamer/IgniteStreamerImpl.java index c18a154..5baa608 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/streamer/IgniteStreamerImpl.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/streamer/IgniteStreamerImpl.java @@ -352,7 +352,7 @@ public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { execFut.get(); } catch (IgniteCheckedException e) { - if (!e.hasCause(GridInterruptedException.class)) + if (!e.hasCause(InternalInterruptedException.class)) U.warn(log, "Failed to wait for batch execution future completion (will ignore) " + "[execFut=" + execFut + ", e=" + e + ']'); } @@ -562,7 +562,7 @@ public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { * @param stageName Stage name. * @param evts Events. * @return Future. - * @throws GridInterruptedException If failed. + * @throws InternalInterruptedException If failed. */ private GridStreamerStageExecutionFuture addEvents0( @Nullable IgniteUuid execId, @@ -572,7 +572,7 @@ public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { @Nullable Collection<UUID> execNodeIds, String stageName, Collection<?> evts - ) throws GridInterruptedException { + ) throws InternalInterruptedException { assert !F.isEmpty(evts); assert !F.isEmpty(stageName); @@ -603,7 +603,7 @@ public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -744,8 +744,8 @@ public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { execSvc.submit(worker); - batchFut.listenAsync(new CI1<IgniteFuture<Object>>() { - @Override public void apply(IgniteFuture<Object> t) { + batchFut.listenAsync(new CI1<InternalFuture<Object>>() { + @Override public void apply(InternalFuture<Object> t) { BatchExecutionFuture fut = (BatchExecutionFuture)t; if (log.isDebugEnabled()) @@ -888,7 +888,7 @@ public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { addEvents0(null, fut.failoverAttemptCount() + 1, 0, null, Collections.singleton(ctx.localNodeId()), fut.stageName(), fut.events()); } - catch (GridInterruptedException e) { + catch (InternalInterruptedException e) { e.printStackTrace(); assert false : "Failover submit should never attempt to acquire semaphore: " + fut + ']'; @@ -1270,7 +1270,7 @@ public class IgniteStreamerImpl implements IgniteStreamerEx, Externalizable { } /** {@inheritDoc} */ - @Override protected void body() throws InterruptedException, GridInterruptedException { + @Override protected void body() throws InterruptedException, InternalInterruptedException { try { long start = U.currentTimeMillis(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/processors/task/GridTaskProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/task/GridTaskProcessor.java b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/task/GridTaskProcessor.java index d05d5ba..c338947 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/processors/task/GridTaskProcessor.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/processors/task/GridTaskProcessor.java @@ -271,7 +271,7 @@ public class GridTaskProcessor extends GridProcessorAdapter { * @param <T> Task argument type. * @param <R> Task return value type. */ - public <T, R> ComputeTaskFuture<R> execute(Class<? extends ComputeTask<T, R>> taskCls, @Nullable T arg) { + public <T, R> InternalComputeTaskFuture<R> execute(Class<? extends ComputeTask<T, R>> taskCls, @Nullable T arg) { assert taskCls != null; lock.readLock(); @@ -294,7 +294,7 @@ public class GridTaskProcessor extends GridProcessorAdapter { * @param <T> Task argument type. * @param <R> Task return value type. */ - public <T, R> ComputeTaskFuture<R> execute(ComputeTask<T, R> task, @Nullable T arg) { + public <T, R> InternalComputeTaskFuture<R> execute(ComputeTask<T, R> task, @Nullable T arg) { return execute(task, arg, false); } @@ -306,7 +306,7 @@ public class GridTaskProcessor extends GridProcessorAdapter { * @param <T> Task argument type. * @param <R> Task return value type. */ - public <T, R> ComputeTaskFuture<R> execute(ComputeTask<T, R> task, @Nullable T arg, boolean sys) { + public <T, R> InternalComputeTaskFuture<R> execute(ComputeTask<T, R> task, @Nullable T arg, boolean sys) { lock.readLock(); try { @@ -342,7 +342,7 @@ public class GridTaskProcessor extends GridProcessorAdapter { * @param <T> Task argument type. * @param <R> Task return value type. */ - public <T, R> ComputeTaskFuture<R> execute(String taskName, @Nullable T arg) { + public <T, R> InternalComputeTaskFuture<R> execute(String taskName, @Nullable T arg) { assert taskName != null; lock.readLock(); @@ -368,7 +368,7 @@ public class GridTaskProcessor extends GridProcessorAdapter { * @return Task future. */ @SuppressWarnings("unchecked") - private <T, R> ComputeTaskFuture<R> startTask( + private <T, R> InternalComputeTaskFuture<R> startTask( @Nullable String taskName, @Nullable Class<?> taskCls, @Nullable ComputeTask<T, R> task, @@ -608,23 +608,23 @@ public class GridTaskProcessor extends GridProcessorAdapter { /** * @param sesId Task's session id. - * @return A {@link org.apache.ignite.compute.ComputeTaskFuture} instance or {@code null} if no such task found. + * @return A {@link InternalComputeTaskFuture} instance or {@code null} if no such task found. */ - @Nullable public <R> ComputeTaskFuture<R> taskFuture(IgniteUuid sesId) { + @Nullable public <R> InternalComputeTaskFuture<R> taskFuture(IgniteUuid sesId) { GridTaskWorker<?, ?> taskWorker = tasks.get(sesId); - return taskWorker != null ? (ComputeTaskFuture<R>)taskWorker.getTaskFuture() : null; + return taskWorker != null ? (InternalComputeTaskFuture<R>)taskWorker.getTaskFuture() : null; } /** * @return Active task futures. */ @SuppressWarnings("unchecked") - public <R> Map<IgniteUuid, ComputeTaskFuture<R>> taskFutures() { - Map<IgniteUuid, ComputeTaskFuture<R>> res = U.newHashMap(tasks.size()); + public <R> Map<IgniteUuid, InternalComputeTaskFuture<R>> taskFutures() { + Map<IgniteUuid, InternalComputeTaskFuture<R>> res = U.newHashMap(tasks.size()); for (GridTaskWorker taskWorker : tasks.values()) { - ComputeTaskFuture<R> fut = taskWorker.getTaskFuture(); + InternalComputeTaskFuture<R> fut = taskWorker.getTaskFuture(); res.put(fut.getTaskSession().getId(), fut); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/visor/cache/VisorCachePreloadTask.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/visor/cache/VisorCachePreloadTask.java b/modules/core/src/main/java/org/gridgain/grid/kernal/visor/cache/VisorCachePreloadTask.java index d9bcd72..7dab4aa 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/visor/cache/VisorCachePreloadTask.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/visor/cache/VisorCachePreloadTask.java @@ -48,14 +48,14 @@ public class VisorCachePreloadTask extends VisorOneNodeTask<Set<String>, Void> { /** {@inheritDoc} */ @Override protected Void run(Set<String> cacheNames) throws IgniteCheckedException { - Collection<IgniteFuture<?>> futs = new ArrayList<>(); + Collection<InternalFuture<?>> futs = new ArrayList<>(); for(GridCache c : g.cachesx()) { if (cacheNames.contains(c.name())) futs.add(c.forceRepartition()); } - for (IgniteFuture f: futs) + for (InternalFuture f: futs) f.get(); return null; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/kernal/visor/compute/VisorComputeCancelSessionsTask.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/visor/compute/VisorComputeCancelSessionsTask.java b/modules/core/src/main/java/org/gridgain/grid/kernal/visor/compute/VisorComputeCancelSessionsTask.java index faeddeb..49f93ce 100644 --- a/modules/core/src/main/java/org/gridgain/grid/kernal/visor/compute/VisorComputeCancelSessionsTask.java +++ b/modules/core/src/main/java/org/gridgain/grid/kernal/visor/compute/VisorComputeCancelSessionsTask.java @@ -12,7 +12,6 @@ package org.gridgain.grid.kernal.visor.compute; import org.apache.ignite.*; import org.apache.ignite.compute.*; import org.apache.ignite.lang.*; -import org.gridgain.grid.*; import org.gridgain.grid.kernal.processors.task.*; import org.gridgain.grid.kernal.visor.*; import org.gridgain.grid.util.typedef.internal.*; @@ -61,10 +60,10 @@ public class VisorComputeCancelSessionsTask extends VisorMultiNodeTask<Map<UUID, if (sesIds != null && !sesIds.isEmpty()) { IgniteCompute compute = g.compute(g.forLocal()); - Map<IgniteUuid, ComputeTaskFuture<Object>> futs = compute.activeTaskFutures(); + Map<IgniteUuid, InternalComputeTaskFuture<Object>> futs = compute.activeTaskFutures(); for (IgniteUuid sesId : sesIds) { - ComputeTaskFuture<Object> fut = futs.get(sesId); + InternalComputeTaskFuture<Object> fut = futs.get(sesId); if (fut != null) fut.cancel(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/util/GridAtomicInitializer.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/util/GridAtomicInitializer.java b/modules/core/src/main/java/org/gridgain/grid/util/GridAtomicInitializer.java index 6814ce2..0d60148 100644 --- a/modules/core/src/main/java/org/gridgain/grid/util/GridAtomicInitializer.java +++ b/modules/core/src/main/java/org/gridgain/grid/util/GridAtomicInitializer.java @@ -85,9 +85,9 @@ public class GridAtomicInitializer<T> { * Await for completion. * * @return {@code true} If initialization was completed successfully. - * @throws GridInterruptedException If thread was interrupted. + * @throws InternalInterruptedException If thread was interrupted. */ - public boolean await() throws GridInterruptedException { + public boolean await() throws InternalInterruptedException { if (!finished) { synchronized (mux) { while (!finished) http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/util/GridUtils.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/util/GridUtils.java b/modules/core/src/main/java/org/gridgain/grid/util/GridUtils.java index 803badd..d5648af 100644 --- a/modules/core/src/main/java/org/gridgain/grid/util/GridUtils.java +++ b/modules/core/src/main/java/org/gridgain/grid/util/GridUtils.java @@ -5602,17 +5602,17 @@ public abstract class GridUtils { * Converts {@link InterruptedException} to {@link IgniteCheckedException}. * * @param mux Mux to wait on. - * @throws GridInterruptedException If interrupted. + * @throws InternalInterruptedException If interrupted. */ @SuppressWarnings({"WaitNotInLoop", "WaitWhileNotSynced"}) - public static void wait(Object mux) throws GridInterruptedException { + public static void wait(Object mux) throws InternalInterruptedException { try { mux.wait(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -6822,16 +6822,16 @@ public abstract class GridUtils { * Awaits for condition. * * @param cond Condition to await for. - * @throws GridInterruptedException Wrapped {@link InterruptedException} + * @throws InternalInterruptedException Wrapped {@link InterruptedException} */ - public static void await(Condition cond) throws GridInterruptedException { + public static void await(Condition cond) throws InternalInterruptedException { try { cond.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -6842,16 +6842,16 @@ public abstract class GridUtils { * @param time The maximum time to wait, * @param unit The unit of the {@code time} argument. * @return {@code false} if the waiting time detectably elapsed before return from the method, else {@code true} - * @throws GridInterruptedException Wrapped {@link InterruptedException} + * @throws InternalInterruptedException Wrapped {@link InterruptedException} */ - public static boolean await(Condition cond, long time, TimeUnit unit) throws GridInterruptedException { + public static boolean await(Condition cond, long time, TimeUnit unit) throws InternalInterruptedException { try { return cond.await(time, unit); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -6859,9 +6859,9 @@ public abstract class GridUtils { * Awaits for the latch. * * @param latch Latch to wait for. - * @throws GridInterruptedException Wrapped {@link InterruptedException}. + * @throws InternalInterruptedException Wrapped {@link InterruptedException}. */ - public static void await(CountDownLatch latch) throws GridInterruptedException { + public static void await(CountDownLatch latch) throws InternalInterruptedException { try { if (latch.getCount() > 0) latch.await(); @@ -6869,7 +6869,7 @@ public abstract class GridUtils { catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -6881,17 +6881,17 @@ public abstract class GridUtils { * @param unit Time unit for timeout. * @return {@code True} if the count reached zero and {@code false} * if the waiting time elapsed before the count reached zero. - * @throws GridInterruptedException Wrapped {@link InterruptedException}. + * @throws InternalInterruptedException Wrapped {@link InterruptedException}. */ public static boolean await(CountDownLatch latch, long timeout, TimeUnit unit) - throws GridInterruptedException { + throws InternalInterruptedException { try { return latch.await(timeout, unit); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -6955,16 +6955,16 @@ public abstract class GridUtils { * Sleeps for given number of milliseconds. * * @param ms Time to sleep. - * @throws GridInterruptedException Wrapped {@link InterruptedException}. + * @throws InternalInterruptedException Wrapped {@link InterruptedException}. */ - public static void sleep(long ms) throws GridInterruptedException { + public static void sleep(long ms) throws InternalInterruptedException { try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -6972,9 +6972,9 @@ public abstract class GridUtils { * Joins worker. * * @param w Worker. - * @throws GridInterruptedException Wrapped {@link InterruptedException}. + * @throws InternalInterruptedException Wrapped {@link InterruptedException}. */ - public static void join(GridWorker w) throws GridInterruptedException { + public static void join(GridWorker w) throws InternalInterruptedException { try { if (w != null) w.join(); @@ -6982,7 +6982,7 @@ public abstract class GridUtils { catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -7003,7 +7003,7 @@ public abstract class GridUtils { catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } catch (CancellationException e) { throw new IgniteCheckedException(e); @@ -7014,16 +7014,16 @@ public abstract class GridUtils { * Joins thread. * * @param t Thread. - * @throws GridInterruptedException Wrapped {@link InterruptedException}. + * @throws InternalInterruptedException Wrapped {@link InterruptedException}. */ - public static void join(Thread t) throws GridInterruptedException { + public static void join(Thread t) throws InternalInterruptedException { try { t.join(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -7031,16 +7031,16 @@ public abstract class GridUtils { * Acquires a permit from provided semaphore. * * @param sem Semaphore. - * @throws GridInterruptedException Wrapped {@link InterruptedException}. + * @throws InternalInterruptedException Wrapped {@link InterruptedException}. */ - public static void acquire(Semaphore sem) throws GridInterruptedException { + public static void acquire(Semaphore sem) throws InternalInterruptedException { try { sem.acquire(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new GridInterruptedException(e); + throw new InternalInterruptedException(e); } } @@ -7219,10 +7219,10 @@ public abstract class GridUtils { * @param f Future to listen to. * @param log Logger. */ - public static void asyncLogError(IgniteFuture<?> f, final IgniteLogger log) { + public static void asyncLogError(InternalFuture<?> f, final IgniteLogger log) { if (f != null) - f.listenAsync(new CI1<IgniteFuture<?>>() { - @Override public void apply(IgniteFuture<?> f) { + f.listenAsync(new CI1<InternalFuture<?>>() { + @Override public void apply(InternalFuture<?> f) { try { f.get(); } @@ -9027,4 +9027,45 @@ public abstract class GridUtils { return list; } + + /** + * Propagates {@code throwable} as-is if it is an instance of {@link RuntimeException} or {@link Error}, or else as + * a last resort, wraps it in a {@code RuntimeException} then propagates. + * <p> + * This method always throws an exception. The {@code RuntimeException} return type is only for client code to make + * Java type system happy in case a return value is required by the enclosing method. + * @param throwable + * @return nothing will ever be returned; this return type is only for your convenience. + */ + public static IgniteException wrap(Throwable throwable) { + if (throwable instanceof Error) + throw (Error)throwable; + + if (throwable instanceof RuntimeException) + throw (RuntimeException)throwable; + + + throw new IgniteException(throwable); + } + + /** + * Propagates {@code throwable} as-is if it is an instance of {@link RuntimeException} or {@link Error}, or else as + * a last resort, wraps it in a {@code RuntimeException} then propagates. + * <p> + * This method always throws an exception. The {@code RuntimeException} return type is only for client code to make + * Java type system happy in case a return value is required by the enclosing method. + * @param throwable + * @param msg Message for wrapping exception. + * @return nothing will ever be returned; this return type is only for your convenience. + */ + public static IgniteException wrap(Throwable throwable, String msg) { + if (throwable instanceof Error) + throw (Error)throwable; + + if (throwable instanceof RuntimeException) + throw (RuntimeException)throwable; + + + throw new IgniteException(msg, throwable); + } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundFuture.java b/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundFuture.java index bf89359..325f872 100644 --- a/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundFuture.java +++ b/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundFuture.java @@ -32,10 +32,10 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { private static final long serialVersionUID = 0L; /** Futures. */ - private final ConcurrentLinkedDeque8<IgniteFuture<T>> futs = new ConcurrentLinkedDeque8<>(); + private final ConcurrentLinkedDeque8<InternalFuture<T>> futs = new ConcurrentLinkedDeque8<>(); /** Pending futures. */ - private final Collection<IgniteFuture<T>> pending = new ConcurrentLinkedDeque8<>(); + private final Collection<InternalFuture<T>> pending = new ConcurrentLinkedDeque8<>(); /** Listener call count. */ private final AtomicInteger lsnrCalls = new AtomicInteger(); @@ -89,7 +89,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { * @param futs Futures to add. */ public GridCompoundFuture(GridKernalContext ctx, @Nullable IgniteReducer<T, R> rdc, - @Nullable Iterable<IgniteFuture<T>> futs) { + @Nullable Iterable<InternalFuture<T>> futs) { super(ctx); this.rdc = rdc; @@ -100,7 +100,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { /** {@inheritDoc} */ @Override public boolean cancel() throws IgniteCheckedException { if (onCancelled()) { - for (IgniteFuture<T> fut : futs) + for (InternalFuture<T> fut : futs) fut.cancel(); return true; @@ -114,7 +114,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { * * @return Collection of futures. */ - public Collection<IgniteFuture<T>> futures() { + public Collection<InternalFuture<T>> futures() { return futs; } @@ -123,7 +123,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { * * @return Pending futures. */ - public Collection<IgniteFuture<T>> pending() { + public Collection<InternalFuture<T>> pending() { return pending; } @@ -158,7 +158,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { * * @param fut Future to add. */ - public void add(IgniteFuture<T> fut) { + public void add(InternalFuture<T> fut) { assert fut != null; pending.add(fut); @@ -180,7 +180,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { * * @param futs Futures to add. */ - public void addAll(@Nullable IgniteFuture<T>... futs) { + public void addAll(@Nullable InternalFuture<T>... futs) { addAll(F.asList(futs)); } @@ -189,9 +189,9 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { * * @param futs Futures to add. */ - public void addAll(@Nullable Iterable<IgniteFuture<T>> futs) { + public void addAll(@Nullable Iterable<InternalFuture<T>> futs) { if (futs != null) - for (IgniteFuture<T> fut : futs) + for (InternalFuture<T> fut : futs) add(fut); } @@ -283,8 +283,8 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { "cancelled", isCancelled(), "err", error(), "futs", - F.viewReadOnly(futs, new C1<IgniteFuture<T>, String>() { - @Override public String apply(IgniteFuture<T> f) { + F.viewReadOnly(futs, new C1<InternalFuture<T>, String>() { + @Override public String apply(InternalFuture<T> f) { return Boolean.toString(f.isDone()); } }) @@ -294,12 +294,12 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { /** * Listener for futures. */ - private class Listener implements IgniteInClosure<IgniteFuture<T>> { + private class Listener implements IgniteInClosure<InternalFuture<T>> { /** */ private static final long serialVersionUID = 0L; /** {@inheritDoc} */ - @Override public void apply(IgniteFuture<T> fut) { + @Override public void apply(InternalFuture<T> fut) { pending.remove(fut); try { @@ -336,7 +336,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> { err.compareAndSet(null, e); } - catch (IgniteFutureCancelledException e) { + catch (InternalFutureCancelledException e) { if (log.isDebugEnabled()) log.debug("Failed to execute compound future reducer [lsnr=" + this + ", e=" + e + ']'); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/90948e67/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundIdentityFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundIdentityFuture.java b/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundIdentityFuture.java index ce28737..51af9b2 100644 --- a/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundIdentityFuture.java +++ b/modules/core/src/main/java/org/gridgain/grid/util/future/GridCompoundIdentityFuture.java @@ -49,7 +49,7 @@ public class GridCompoundIdentityFuture<T> extends GridCompoundFuture<T, T> { * @param futs Futures to add. */ public GridCompoundIdentityFuture(GridKernalContext ctx, @Nullable IgniteReducer<T, T> rdc, - @Nullable Iterable<IgniteFuture<T>> futs) { + @Nullable Iterable<InternalFuture<T>> futs) { super(ctx, rdc, futs); }
