You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2015/11/02 15:02:56 UTC
ignite git commit: IGNITE-950-new - Fixing keepBinary for data
streamer.
Repository: ignite
Updated Branches:
refs/heads/ignite-950-new 6154fb52e -> f5e1f887e
IGNITE-950-new - Fixing keepBinary for data streamer.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f5e1f887
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f5e1f887
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f5e1f887
Branch: refs/heads/ignite-950-new
Commit: f5e1f887eefafc3933b037c895e124236e045169
Parents: 6154fb5
Author: Alexey Goncharuk <al...@gmail.com>
Authored: Mon Nov 2 17:02:35 2015 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Mon Nov 2 17:02:35 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/IgniteDataStreamer.java | 16 ++++++
.../datastreamer/DataStreamProcessor.java | 1 +
.../datastreamer/DataStreamerEntry.java | 6 +-
.../datastreamer/DataStreamerImpl.java | 16 +++++-
.../datastreamer/DataStreamerRequest.java | 58 ++++++++++++++------
.../datastreamer/DataStreamerUpdateJob.java | 7 ++-
.../platform/PlatformProcessorImpl.java | 2 +
7 files changed, 85 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5e1f887/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java b/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
index 79bd163..20d0057 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
@@ -146,6 +146,22 @@ public interface IgniteDataStreamer<K, V> extends AutoCloseable {
public void skipStore(boolean skipStore);
/**
+ * Gets flag indicating that objects should be kept in binary format when passed to the stream receiver.
+ * Default is {@code false}.
+ *
+ * @return Skip store flag.
+ */
+ public boolean keepBinary();
+
+ /**
+ * Sets flag indicating that objects should be kept in binary format when passes to the steam receiver.
+ * Default is {@code false}.
+ *
+ * @param keepBinary Keep binary flag.
+ */
+ public void keepBinary(boolean keepBinary);
+
+ /**
* Gets size of per node key-value pairs buffer.
*
* @return Per node buffer size.
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5e1f887/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
index 5150d83..a2aab77 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
@@ -289,6 +289,7 @@ public class DataStreamProcessor<K, V> extends GridProcessorAdapter {
col,
req.ignoreDeploymentOwnership(),
req.skipStore(),
+ req.keepBinary(),
updater);
Exception err = null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5e1f887/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerEntry.java
index 982b691..874d2e2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerEntry.java
@@ -82,10 +82,10 @@ public class DataStreamerEntry implements Map.Entry<KeyCacheObject, CacheObject>
* @param ctx Cache context.
* @return Map entry unwrapping internal key and value.
*/
- public <K, V> Map.Entry<K, V> toEntry(final GridCacheContext ctx) {
+ public <K, V> Map.Entry<K, V> toEntry(final GridCacheContext ctx, final boolean keepBinary) {
return new Map.Entry<K, V>() {
@Override public K getKey() {
- return key.value(ctx.cacheObjectContext(), false);
+ return (K)ctx.cacheObjectContext().unwrapPortableIfNeeded(key, keepBinary, false);
}
@Override public V setValue(V val) {
@@ -93,7 +93,7 @@ public class DataStreamerEntry implements Map.Entry<KeyCacheObject, CacheObject>
}
@Override public V getValue() {
- return val != null ? val.<V>value(ctx.cacheObjectContext(), false) : null;
+ return (V)ctx.cacheObjectContext().unwrapPortableIfNeeded(val, keepBinary, false);
}
};
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5e1f887/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
index 2190bf6..27eff0c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
@@ -211,6 +211,9 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
private boolean skipStore;
/** */
+ private boolean keepBinary;
+
+ /** */
private int maxRemapCnt = DFLT_MAX_REMAP_CNT;
/** Whether a warning at {@link DataStreamerImpl#allowOverwrite()} printed */
@@ -405,6 +408,16 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
}
/** {@inheritDoc} */
+ @Override public boolean keepBinary() {
+ return keepBinary;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void keepBinary(boolean keepBinary) {
+ this.keepBinary = keepBinary;
+ }
+
+ /** {@inheritDoc} */
@Override @Nullable public String cacheName() {
return cacheName;
}
@@ -1243,7 +1256,7 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
if (isLocNode) {
fut = ctx.closure().callLocalSafe(
- new DataStreamerUpdateJob(ctx, log, cacheName, entries, false, skipStore, rcvr), false);
+ new DataStreamerUpdateJob(ctx, log, cacheName, entries, false, skipStore, keepBinary, rcvr), false);
locFuts.add(fut);
@@ -1331,6 +1344,7 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
entries,
true,
skipStore,
+ keepBinary,
dep != null ? dep.deployMode() : null,
dep != null ? jobPda0.deployClass().getName() : null,
dep != null ? dep.userVersion() : null,
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5e1f887/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
index c1a1528..3d65304 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java
@@ -64,6 +64,9 @@ public class DataStreamerRequest implements Message {
/** */
private boolean skipStore;
+ /** Keep binary flag. */
+ private boolean keepBinary;
+
/** */
private DeploymentMode depMode;
@@ -117,6 +120,7 @@ public class DataStreamerRequest implements Message {
Collection<DataStreamerEntry> entries,
boolean ignoreDepOwnership,
boolean skipStore,
+ boolean keepBinary,
DeploymentMode depMode,
String sampleClsName,
String userVer,
@@ -133,6 +137,7 @@ public class DataStreamerRequest implements Message {
this.entries = entries;
this.ignoreDepOwnership = ignoreDepOwnership;
this.skipStore = skipStore;
+ this.keepBinary = keepBinary;
this.depMode = depMode;
this.sampleClsName = sampleClsName;
this.userVer = userVer;
@@ -192,6 +197,13 @@ public class DataStreamerRequest implements Message {
}
/**
+ * @return Keep binary flag.
+ */
+ public boolean keepBinary() {
+ return keepBinary;
+ }
+
+ /**
* @return Deployment mode.
*/
public DeploymentMode deploymentMode() {
@@ -294,48 +306,54 @@ public class DataStreamerRequest implements Message {
writer.incrementState();
case 6:
- if (!writer.writeMap("ldrParticipants", ldrParticipants, MessageCollectionItemType.UUID, MessageCollectionItemType.IGNITE_UUID))
+ if (!writer.writeBoolean("keepBinary", keepBinary))
return false;
writer.incrementState();
case 7:
- if (!writer.writeLong("reqId", reqId))
+ if (!writer.writeMap("ldrParticipants", ldrParticipants, MessageCollectionItemType.UUID, MessageCollectionItemType.IGNITE_UUID))
return false;
writer.incrementState();
case 8:
- if (!writer.writeByteArray("resTopicBytes", resTopicBytes))
+ if (!writer.writeLong("reqId", reqId))
return false;
writer.incrementState();
case 9:
- if (!writer.writeString("sampleClsName", sampleClsName))
+ if (!writer.writeByteArray("resTopicBytes", resTopicBytes))
return false;
writer.incrementState();
case 10:
- if (!writer.writeBoolean("skipStore", skipStore))
+ if (!writer.writeString("sampleClsName", sampleClsName))
return false;
writer.incrementState();
case 11:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeBoolean("skipStore", skipStore))
return false;
writer.incrementState();
case 12:
- if (!writer.writeByteArray("updaterBytes", updaterBytes))
+ if (!writer.writeMessage("topVer", topVer))
return false;
writer.incrementState();
case 13:
+ if (!writer.writeByteArray("updaterBytes", updaterBytes))
+ return false;
+
+ writer.incrementState();
+
+ case 14:
if (!writer.writeString("userVer", userVer))
return false;
@@ -407,7 +425,7 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 6:
- ldrParticipants = reader.readMap("ldrParticipants", MessageCollectionItemType.UUID, MessageCollectionItemType.IGNITE_UUID, false);
+ keepBinary = reader.readBoolean("keepBinary");
if (!reader.isLastRead())
return false;
@@ -415,7 +433,7 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 7:
- reqId = reader.readLong("reqId");
+ ldrParticipants = reader.readMap("ldrParticipants", MessageCollectionItemType.UUID, MessageCollectionItemType.IGNITE_UUID, false);
if (!reader.isLastRead())
return false;
@@ -423,7 +441,7 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 8:
- resTopicBytes = reader.readByteArray("resTopicBytes");
+ reqId = reader.readLong("reqId");
if (!reader.isLastRead())
return false;
@@ -431,7 +449,7 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 9:
- sampleClsName = reader.readString("sampleClsName");
+ resTopicBytes = reader.readByteArray("resTopicBytes");
if (!reader.isLastRead())
return false;
@@ -439,7 +457,7 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 10:
- skipStore = reader.readBoolean("skipStore");
+ sampleClsName = reader.readString("sampleClsName");
if (!reader.isLastRead())
return false;
@@ -447,7 +465,7 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 11:
- topVer = reader.readMessage("topVer");
+ skipStore = reader.readBoolean("skipStore");
if (!reader.isLastRead())
return false;
@@ -455,7 +473,7 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 12:
- updaterBytes = reader.readByteArray("updaterBytes");
+ topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
return false;
@@ -463,6 +481,14 @@ public class DataStreamerRequest implements Message {
reader.incrementState();
case 13:
+ updaterBytes = reader.readByteArray("updaterBytes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 14:
userVer = reader.readString("userVer");
if (!reader.isLastRead())
@@ -482,6 +508,6 @@ public class DataStreamerRequest implements Message {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 14;
+ return 15;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5e1f887/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
index 42084a3..34489c8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerUpdateJob.java
@@ -56,6 +56,9 @@ class DataStreamerUpdateJob implements GridPlainCallable<Object> {
/** */
private final StreamReceiver rcvr;
+ /** */
+ private boolean keepBinary;
+
/**
* @param ctx Context.
* @param log Log.
@@ -72,6 +75,7 @@ class DataStreamerUpdateJob implements GridPlainCallable<Object> {
Collection<DataStreamerEntry> col,
boolean ignoreDepOwnership,
boolean skipStore,
+ boolean keepBinary,
StreamReceiver<?, ?> rcvr) {
this.ctx = ctx;
this.log = log;
@@ -83,6 +87,7 @@ class DataStreamerUpdateJob implements GridPlainCallable<Object> {
this.col = col;
this.ignoreDepOwnership = ignoreDepOwnership;
this.skipStore = skipStore;
+ this.keepBinary = keepBinary;
this.rcvr = rcvr;
}
@@ -122,7 +127,7 @@ class DataStreamerUpdateJob implements GridPlainCallable<Object> {
if (unwrapEntries()) {
Collection<Map.Entry> col0 = F.viewReadOnly(col, new C1<DataStreamerEntry, Map.Entry>() {
@Override public Map.Entry apply(DataStreamerEntry e) {
- return e.toEntry(cctx);
+ return e.toEntry(cctx, keepBinary);
}
});
http://git-wip-us.apache.org/repos/asf/ignite/blob/f5e1f887/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
index efe6212..9828753 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/PlatformProcessorImpl.java
@@ -253,6 +253,8 @@ public class PlatformProcessorImpl extends GridProcessorAdapter implements Platf
throws IgniteCheckedException {
IgniteDataStreamer ldr = ctx.dataStream().dataStreamer(cacheName);
+ ldr.keepBinary(keepPortable);
+
return new PlatformDataStreamer(platformCtx, cacheName, (DataStreamerImpl)ldr, keepPortable);
}