You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2016/03/31 15:27:09 UTC
[5/6] ignite git commit: Refactorings.
Refactorings.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/1779445f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/1779445f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/1779445f
Branch: refs/heads/ignite-2925
Commit: 1779445fed6d011a63e73ccc12e4cf06314abc72
Parents: 30b2dde
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Thu Mar 31 16:20:19 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Thu Mar 31 16:20:19 2016 +0300
----------------------------------------------------------------------
.../dht/atomic/GridDhtAtomicCache.java | 99 ++++++++++++--------
1 file changed, 60 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/1779445f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index a681fcf..f2d299b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -994,28 +994,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
ctx.checkSecurity(SecurityPermission.CACHE_PUT);
- CacheOperationContext opCtx = ctx.operationContextPerCall();
-
- final GridNearAtomicUpdateFuture updateFut = new GridNearAtomicUpdateFuture(
- ctx,
- this,
- ctx.config().getWriteSynchronizationMode(),
- val != null ? UPDATE : TRANSFORM,
- Collections.singletonList(key),
- val != null ? Collections.singletonList(val) : Collections.singletonList(proc),
- invokeArgs,
- null,
- null,
- retval,
- false,
- opCtx != null ? opCtx.expiry() : null,
- filter,
- ctx.subjectIdPerCall(null, opCtx),
- ctx.kernalContext().job().currentTaskNameHash(),
- opCtx != null && opCtx.skipStore(),
- opCtx != null && opCtx.isKeepBinary(),
- opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES,
- waitTopFut);
+ final GridNearAtomicUpdateFuture updateFut =
+ createSingleUpdateFuture(key, val, proc, invokeArgs, retval, filter, waitTopFut);
return asyncOp(new CO<IgniteInternalFuture<Object>>() {
@Override public IgniteInternalFuture<Object> apply() {
@@ -1046,39 +1026,80 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
ctx.checkSecurity(SecurityPermission.CACHE_REMOVE);
+ final GridNearAtomicUpdateFuture updateFut =
+ createSingleUpdateFuture(key, null, null, null, retval, filter, true);
+
+ if (statsEnabled)
+ updateFut.listen(new UpdateRemoveTimeStatClosure<>(metrics0(), start));
+
+ return asyncOp(new CO<IgniteInternalFuture<Object>>() {
+ @Override public IgniteInternalFuture<Object> apply() {
+ updateFut.map();
+
+ return updateFut;
+ }
+ });
+ }
+
+ /**
+ * Craete future for single key-val pair update.
+ *
+ * @param key Key.
+ * @param val Value.
+ * @param proc Processor.
+ * @param invokeArgs Invoke arguments.
+ * @param retval Return value flag.
+ * @param filter Filter.
+ * @param waitTopFut Whether to wait for topology future.
+ * @return Future.
+ */
+ private GridNearAtomicUpdateFuture createSingleUpdateFuture(
+ K key,
+ @Nullable V val,
+ @Nullable EntryProcessor proc,
+ @Nullable Object[] invokeArgs,
+ boolean retval,
+ @Nullable final CacheEntryPredicate[] filter,
+ boolean waitTopFut
+ ) {
+ GridCacheOperation op;
+ Collection vals;
+
+ if (val != null) {
+ op = UPDATE;
+ vals = Collections.singletonList(val);
+ }
+ else if (proc != null) {
+ op = TRANSFORM;
+ vals = Collections.singletonList(proc);
+ }
+ else {
+ op = DELETE;
+ vals = null;
+ }
+
CacheOperationContext opCtx = ctx.operationContextPerCall();
- final GridNearAtomicUpdateFuture updateFut = new GridNearAtomicUpdateFuture(
+ return new GridNearAtomicUpdateFuture(
ctx,
this,
ctx.config().getWriteSynchronizationMode(),
- DELETE,
+ op,
Collections.singletonList(key),
- null,
- null,
+ vals,
+ invokeArgs,
null,
null,
retval,
false,
- (filter != null && opCtx != null) ? opCtx.expiry() : null,
+ opCtx != null ? opCtx.expiry() : null,
filter,
ctx.subjectIdPerCall(null, opCtx),
ctx.kernalContext().job().currentTaskNameHash(),
opCtx != null && opCtx.skipStore(),
opCtx != null && opCtx.isKeepBinary(),
opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES,
- true);
-
- if (statsEnabled)
- updateFut.listen(new UpdateRemoveTimeStatClosure<>(metrics0(), start));
-
- return asyncOp(new CO<IgniteInternalFuture<Object>>() {
- @Override public IgniteInternalFuture<Object> apply() {
- updateFut.map();
-
- return updateFut;
- }
- });
+ waitTopFut);
}
/**