ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject incubator-ignite git commit: Ignite-107 code restores
Date Fri, 23 Jan 2015 15:10:04 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-107 79a47101e -> 3e9365289


Ignite-107 code restores


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/3e936528
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/3e936528
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/3e936528

Branch: refs/heads/ignite-107
Commit: 3e9365289e97dac312ec22d2fb00a63eb3c58fbd
Parents: 79a4710
Author: avinogradov <avinogradov@gridgain.com>
Authored: Fri Jan 23 18:09:54 2015 +0300
Committer: avinogradov <avinogradov@gridgain.com>
Committed: Fri Jan 23 18:09:54 2015 +0300

----------------------------------------------------------------------
 .../closure/GridClosureProcessor.java           | 570 ++++++++++++++-----
 .../processors/resource/GridResourceUtils.java  |   4 +-
 2 files changed, 421 insertions(+), 153 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3e936528/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
index a5324b8..2951104 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
@@ -25,6 +25,7 @@ import org.apache.ignite.compute.*;
 import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.processors.*;
 import org.apache.ignite.internal.util.*;
+import org.apache.ignite.internal.util.tostring.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.marshaller.*;
 import org.apache.ignite.resources.*;
@@ -36,6 +37,7 @@ import org.apache.ignite.internal.util.worker.*;
 import org.jdk8.backport.*;
 import org.jetbrains.annotations.*;
 
+import java.io.*;
 import java.util.*;
 import java.util.concurrent.*;
 
@@ -971,24 +973,9 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     private <T, R> ComputeJob job(final IgniteClosure<T, R> job, @Nullable final
T arg) {
         A.notNull(job, "job");
 
-        if (job instanceof ComputeJobMasterLeaveAware) {
-            return new GridMasterLeaveAwareComputeJobAdapter() {
-                @Nullable @Override public Object execute() {
-                    return job.apply(arg);
-                }
-
-                @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
-                    ((ComputeJobMasterLeaveAware)job).onMasterNodeLeft(ses);
-                }
-            };
-        }
-        else {
-            return new ComputeJobAdapter() {
-                @Nullable @Override public Object execute() {
-                    return job.apply(arg);
-                }
-            };
-        }
+        if (job instanceof ComputeJobMasterLeaveAware)
+            return new C1MLA<>(job, arg);
+        else return new C1<>(job, arg);
     }
 
     /**
@@ -998,37 +985,12 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @return Grid job made out of closure.
      */
     @SuppressWarnings("IfMayBeConditional")
-    private ComputeJob job(final Callable<?> c) {
+    private <R> ComputeJob job(final Callable<R> c) {
         A.notNull(c, "job");
 
-        if (c instanceof ComputeJobMasterLeaveAware) {
-            return new GridMasterLeaveAwareComputeJobAdapter() {
-                @Override public Object execute() {
-                    try {
-                        return c.call();
-                    }
-                    catch (Exception e) {
-                        throw new IgniteException(e);
-                    }
-                }
-
-                @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
-                    ((ComputeJobMasterLeaveAware)c).onMasterNodeLeft(ses);
-                }
-            };
-        }
-        else {
-            return new ComputeJobAdapter() {
-                @Override public Object execute() {
-                    try {
-                        return c.call();
-                    }
-                    catch (Exception e) {
-                        throw new IgniteException(e);
-                    }
-                }
-            };
-        }
+        if (c instanceof ComputeJobMasterLeaveAware)
+            return new C2MLA<>(c);
+        else return new C2<>(c);
     }
 
     /**
@@ -1040,53 +1002,12 @@ public class GridClosureProcessor extends GridProcessorAdapter {
      * @return Grid job made out of closure.
      */
     @SuppressWarnings(value = {"IfMayBeConditional", "UnusedDeclaration"})
-    private ComputeJob job(final Callable<?> c, @Nullable final String cacheName, final
Object affKey) {
+    private <R> ComputeJob job(final Callable<R> c, @Nullable final String cacheName,
final Object affKey) {
         A.notNull(c, "job");
 
-        if (c instanceof ComputeJobMasterLeaveAware) {
-            return new GridMasterLeaveAwareComputeJobAdapter() {
-                /** */
-                @CacheName
-                private final String cn = cacheName;
-
-                /** */
-                @CacheAffinityKeyMapped
-                private final Object ak = affKey;
-
-                @Override public Object execute() {
-                    try {
-                        return c.call();
-                    }
-                    catch (Exception e) {
-                        throw new IgniteException(e);
-                    }
-                }
-
-                @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
-                    ((ComputeJobMasterLeaveAware)c).onMasterNodeLeft(ses);
-                }
-            };
-        }
-        else {
-            return new ComputeJobAdapter() {
-                /** */
-                @CacheName
-                private final String cn = cacheName;
-
-                /** */
-                @CacheAffinityKeyMapped
-                private final Object ak = affKey;
-
-                @Override public Object execute() {
-                    try {
-                        return c.call();
-                    }
-                    catch (Exception e) {
-                        throw new IgniteException(e);
-                    }
-                }
-            };
-        }
+        if (c instanceof ComputeJobMasterLeaveAware)
+            return new C3MLA<>(c, cacheName, affKey);
+        else return new C3<>(c, cacheName, affKey);
     }
 
     /**
@@ -1099,28 +1020,9 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     private static ComputeJob job(final Runnable r) {
         A.notNull(r, "job");
 
-        if (r instanceof ComputeJobMasterLeaveAware) {
-            return new GridMasterLeaveAwareComputeJobAdapter() {
-                @Nullable @Override public Object execute() {
-                    r.run();
-
-                    return null;
-                }
-
-                @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
-                    ((ComputeJobMasterLeaveAware)r).onMasterNodeLeft(ses);
-                }
-            };
-        }
-        else {
-            return new ComputeJobAdapter() {
-                @Nullable @Override public Object execute() {
-                    r.run();
-
-                    return null;
-                }
-            };
-        }
+        if (r instanceof ComputeJobMasterLeaveAware)
+            return new C4MLA(r);
+        else  return new C4(r);
     }
 
     /**
@@ -1135,44 +1037,9 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     private ComputeJob job(final Runnable r, @Nullable final String cacheName, final Object
affKey) {
         A.notNull(r, "job");
 
-        if (r instanceof ComputeJobMasterLeaveAware) {
-            return new GridMasterLeaveAwareComputeJobAdapter() {
-                /** */
-                @CacheName
-                private final String cn = cacheName;
-
-                /** */
-                @CacheAffinityKeyMapped
-                private final Object ak = affKey;
-
-                @Nullable @Override public Object execute() {
-                    r.run();
-
-                    return null;
-                }
-
-                @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
-                    ((ComputeJobMasterLeaveAware)r).onMasterNodeLeft(ses);
-                }
-            };
-        }
-        else {
-            return new ComputeJobAdapter() {
-                /** */
-                @CacheName
-                private final String cn = cacheName;
-
-                /** */
-                @CacheAffinityKeyMapped
-                private final Object ak = affKey;
-
-                @Nullable @Override public Object execute() {
-                    r.run();
-
-                    return null;
-                }
-            };
-        }
+        if (r instanceof ComputeJobMasterLeaveAware)
+            return new C5MLA(r, cacheName, affKey);
+        else return new C5(r, cacheName, affKey);
     }
 
     /** */
@@ -1741,4 +1608,403 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             return F.jobResults(res);
         }
     }
+
+    /**
+     *
+     */
+    private static class C1<T, R> implements ComputeJob, Externalizable {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** {@inheritDoc} */
+        @Override public void cancel() {
+            // No-op.
+        }
+
+        /** */
+        protected IgniteClosure<T, R> job;
+
+        /** */
+        @GridToStringInclude
+        private T arg;
+
+        /**
+         *
+         */
+        public C1(){
+            // No-op.
+        }
+
+        /**
+         *
+         */
+        public C1(IgniteClosure<T, R> job, T arg) {
+            this.job = job;
+            this.arg = arg;
+        }
+
+        /** {@inheritDoc} */
+        @Nullable @Override public Object execute() {
+            return job.apply(arg);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeExternal(ObjectOutput out) throws IOException {
+            out.writeObject(job);
+            out.writeObject(arg);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+            job = (IgniteClosure<T, R>)in.readObject();
+            arg = (T)in.readObject();
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(C1.class, this);
+        }
+    }
+
+    /**
+     *
+     */
+    private static class C1MLA<T, R> extends C1<T, R> implements ComputeJobMasterLeaveAware{
+        /**
+         *
+         */
+        public C1MLA() {
+            super();
+        }
+
+        /**
+         *
+         */
+        public C1MLA(IgniteClosure<T, R> job, T arg) {
+            super(job, arg);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
+            ((ComputeJobMasterLeaveAware)job).onMasterNodeLeft(ses);
+        }
+    }
+
+    /**
+     *
+     */
+    private static class C2<R> implements ComputeJob, Externalizable {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** {@inheritDoc} */
+        @Override public void cancel() {
+            // No-op.
+        }
+
+        /** */
+        protected Callable<R> c;
+
+        /**
+         *
+         */
+        public C2(){
+            // No-op.
+        }
+
+        /**
+         *
+         */
+        public C2(Callable<R> c) {
+            this.c = c;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object execute() {
+            try {
+                return c.call();
+            }
+            catch (Exception e) {
+                throw new IgniteException(e);
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeExternal(ObjectOutput out) throws IOException {
+            out.writeObject(c);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+            c = (Callable<R>)in.readObject();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class C2MLA<R> extends C2<R> implements ComputeJobMasterLeaveAware{
+        /**
+         *
+         */
+        public C2MLA() {
+            super();
+        }
+
+        /**
+         *
+         */
+        public C2MLA(Callable<R> c) {
+            super(c);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
+            ((ComputeJobMasterLeaveAware)c).onMasterNodeLeft(ses);
+        }
+    }
+
+    /**
+     */
+    private static class C3<R> implements ComputeJob, Externalizable {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** {@inheritDoc} */
+        @Override public void cancel() {
+            // No-op.
+        }
+
+        /** */
+        @CacheName
+        private String cn;
+
+        /** */
+        @CacheAffinityKeyMapped
+        private Object ak;
+
+
+        /** */
+        protected Callable<R> c;
+
+        /**
+         *
+         */
+        public C3(){
+            // No-op.
+        }
+
+        /**
+         *
+         */
+        public C3(Callable<R> c, @Nullable String cacheName, Object affKey) {
+            this.cn = cacheName;
+            this.ak = affKey;
+            this.c = c;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object execute() {
+            try {
+                return c.call();
+            }
+            catch (Exception e) {
+                throw new IgniteException(e);
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeExternal(ObjectOutput out) throws IOException {
+            out.writeObject(cn);
+            out.writeObject(ak);
+            out.writeObject(c);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+            cn = (String)in.readObject();
+            ak = in.readObject();
+            c = (Callable<R>)in.readObject();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class C3MLA<R> extends C3<R> implements ComputeJobMasterLeaveAware{
+        /**
+         *
+         */
+        public C3MLA() {
+            super();
+        }
+
+        /**
+         *
+         */
+        public C3MLA(Callable<R> c, @Nullable String cacheName, Object affKey) {
+            super(c, cacheName, affKey);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
+            ((ComputeJobMasterLeaveAware)c).onMasterNodeLeft(ses);
+        }
+    }
+
+    /**
+     */
+    private static class C4 implements ComputeJob, Externalizable {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** {@inheritDoc} */
+        @Override public void cancel() {
+            // No-op.
+        }
+
+        /** */
+        protected Runnable r;
+
+        /**
+         *
+         */
+        public C4(){
+            // No-op.
+        }
+
+        /**
+         *
+         */
+        public C4(Runnable r) {
+            this.r = r;
+        }
+
+        /** {@inheritDoc} */
+        @Nullable @Override public Object execute() {
+            r.run();
+
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeExternal(ObjectOutput out) throws IOException {
+            out.writeObject(r);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+            r = (Runnable)in.readObject();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class C4MLA extends C4 implements ComputeJobMasterLeaveAware{
+        /**
+         *
+         */
+        public C4MLA() {
+            super();
+        }
+
+        /**
+         *
+         */
+        public C4MLA(Runnable r) {
+            super(r);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
+            ((ComputeJobMasterLeaveAware)r).onMasterNodeLeft(ses);
+        }
+
+    }
+
+    /**
+     */
+    private static class C5 implements ComputeJob, Externalizable {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** {@inheritDoc} */
+        @Override public void cancel() {
+            // No-op.
+        }
+
+        /** */
+        @CacheName
+        private String cn;
+
+        /** */
+        @CacheAffinityKeyMapped
+        private Object ak;
+
+        /** */
+        protected Runnable r;
+
+        /**
+         *
+         */
+        public C5(){
+            // No-op.
+        }
+
+        /**
+         *
+         */
+        public C5(Runnable r, @Nullable String cacheName, Object affKey) {
+            this.cn = cacheName;
+            this.ak = affKey;
+            this.r = r;
+        }
+
+        /** {@inheritDoc} */
+        @Nullable @Override public Object execute() {
+            r.run();
+
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeExternal(ObjectOutput out) throws IOException {
+            out.writeObject(cn);
+            out.writeObject(ak);
+            out.writeObject(r);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+            cn = (String)in.readObject();
+            ak = in.readObject();
+            r = (Runnable)in.readObject();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class C5MLA extends C5 implements ComputeJobMasterLeaveAware{
+        /**
+         *
+         */
+        public C5MLA() {
+            super();
+        }
+
+        /**
+         *
+         */
+        public C5MLA(Runnable r, @Nullable String cacheName, Object affKey) {
+            super(r, cacheName, affKey);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException
{
+            ((ComputeJobMasterLeaveAware)r).onMasterNodeLeft(ses);
+        }
+
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3e936528/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceUtils.java
index 572fd91..c80b167 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceUtils.java
@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.resource;
 
 import org.apache.ignite.*;
+import org.apache.ignite.lang.*;
 
 import java.lang.reflect.*;
 import java.util.concurrent.*;
@@ -99,6 +100,7 @@ final class GridResourceUtils {
 
         // Need to inspect anonymous classes, callable and runnable instances.
         return f.getName().startsWith("this$") || f.getName().startsWith("val$") ||
-            Callable.class.isAssignableFrom(f.getType()) || Runnable.class.isAssignableFrom(f.getType());
+                Callable.class.isAssignableFrom(f.getType()) || Runnable.class.isAssignableFrom(f.getType())
||
+                IgniteClosure.class.isAssignableFrom(f.getType());
     }
 }


Mime
View raw message