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/09/07 11:17:03 UTC
[18/50] [abbrv] ignite git commit: Added node-aware infrastructure.
Added node-aware infrastructure.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/07b5a8b9
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/07b5a8b9
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/07b5a8b9
Branch: refs/heads/ignite-2649
Commit: 07b5a8b9f096068779e1440fdbc705c98755b649
Parents: 869e883
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Tue Sep 6 19:27:03 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Tue Sep 6 19:27:03 2016 +0300
----------------------------------------------------------------------
.../ignite/cache/store/jdbc/CacheJdbcBlobStore.java | 2 +-
.../java/org/apache/ignite/internal/IgnitionEx.java | 4 ++--
.../apache/ignite/internal/binary/BinaryContext.java | 5 ++++-
.../ignite/internal/processors/igfs/IgfsPaths.java | 4 +---
.../rest/protocols/tcp/GridTcpRestParser.java | 4 +---
.../marshaller/AbstractNodeNameAwareMarshaller.java | 5 +----
.../org/apache/ignite/marshaller/MarshallerUtils.java | 14 ++++++++++++++
.../checkpoint/sharedfs/SharedFsCheckpointSpi.java | 5 ++---
.../ignite/spi/discovery/tcp/TcpDiscoveryImpl.java | 2 +-
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 4 ++--
.../multicast/TcpDiscoveryMulticastIpFinder.java | 2 +-
.../communication/HadoopMarshallerFilter.java | 2 +-
.../apache/ignite/cache/websession/WebSessionV2.java | 2 +-
13 files changed, 32 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcBlobStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcBlobStore.java b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcBlobStore.java
index dd3da31..557ebf7 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcBlobStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheJdbcBlobStore.java
@@ -600,6 +600,6 @@ public class CacheJdbcBlobStore<K, V> extends CacheStoreAdapter<K, V> {
public void setIgnite(Ignite ignite) {
this.ignite = ignite;
- marsh.nodeName(ignite.name());
+ MarshallerUtils.withNodeName(marsh, ignite.name());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
index 07b1e2c..fae513e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
@@ -78,6 +78,7 @@ import org.apache.ignite.logger.LoggerNodeIdAware;
import org.apache.ignite.logger.java.JavaLogger;
import org.apache.ignite.marshaller.AbstractNodeNameAwareMarshaller;
import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.marshaller.MarshallerUtils;
import org.apache.ignite.marshaller.jdk.JdkMarshaller;
import org.apache.ignite.mxbean.IgnitionMXBean;
import org.apache.ignite.plugin.segmentation.SegmentationPolicy;
@@ -1935,8 +1936,7 @@ public class IgnitionEx {
marsh = new BinaryMarshaller();
}
- if (marsh instanceof AbstractNodeNameAwareMarshaller)
- ((AbstractNodeNameAwareMarshaller)marsh).nodeName(cfg.getGridName());
+ MarshallerUtils.withNodeName(marsh, cfg.getGridName());
myCfg.setMarshaller(marsh);
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
index aa9e8a5..7be3519 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryContext.java
@@ -81,6 +81,7 @@ import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.marshaller.MarshallerContext;
+import org.apache.ignite.marshaller.MarshallerUtils;
import org.apache.ignite.marshaller.optimized.OptimizedMarshaller;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
@@ -241,7 +242,9 @@ public class BinaryContext {
assert metaHnd != null;
assert igniteCfg != null;
- optmMarsh = new OptimizedMarshaller(false).nodeName(igniteCfg.getGridName());
+ optmMarsh = new OptimizedMarshaller(false);
+
+ MarshallerUtils.withNodeName(optmMarsh, igniteCfg.getGridName());
this.metaHnd = metaHnd;
this.igniteCfg = igniteCfg;
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
index 13f1daf..da0a347 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
@@ -78,9 +78,7 @@ public class IgfsPaths implements Externalizable {
else {
ByteArrayOutputStream out = new ByteArrayOutputStream();
- JdkMarshaller marsh = new JdkMarshaller().nodeName(igfsCtx.kernalContext().gridName());
-
- marsh.marshal(payload, out);
+ MarshallerUtils.withNodeName(new JdkMarshaller(), igfsCtx.kernalContext().gridName()).marshal(payload, out);
payloadBytes = out.toByteArray();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
index 77179b7..e32f0ec 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestParser.java
@@ -805,9 +805,7 @@ public class GridTcpRestParser implements GridNioParser {
flags |= BYTE_ARR_FLAG;
}
else {
- jdkMarshaller.nodeName(gridName);
-
- jdkMarshaller.marshal(obj, out);
+ MarshallerUtils.withNodeName(jdkMarshaller, gridName).marshal(obj, out);
flags |= SERIALIZED_FLAG;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/marshaller/AbstractNodeNameAwareMarshaller.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/marshaller/AbstractNodeNameAwareMarshaller.java b/modules/core/src/main/java/org/apache/ignite/marshaller/AbstractNodeNameAwareMarshaller.java
index 4425dfd..559ac73 100644
--- a/modules/core/src/main/java/org/apache/ignite/marshaller/AbstractNodeNameAwareMarshaller.java
+++ b/modules/core/src/main/java/org/apache/ignite/marshaller/AbstractNodeNameAwareMarshaller.java
@@ -40,17 +40,14 @@ public abstract class AbstractNodeNameAwareMarshaller extends AbstractMarshaller
* Set node name.
*
* @param nodeName Node name.
- * @return This instance for chaining.
*/
@SuppressWarnings("unchecked")
- public <T> T nodeName(@Nullable String nodeName) {
+ public void nodeName(@Nullable String nodeName) {
if (!nodeNameSet) {
this.nodeName = nodeName;
nodeNameSet = true;
}
-
- return (T)this;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/marshaller/MarshallerUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/marshaller/MarshallerUtils.java b/modules/core/src/main/java/org/apache/ignite/marshaller/MarshallerUtils.java
index 3c5a1df..ef0a93a 100644
--- a/modules/core/src/main/java/org/apache/ignite/marshaller/MarshallerUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/marshaller/MarshallerUtils.java
@@ -30,6 +30,20 @@ import java.io.OutputStream;
*/
public class MarshallerUtils {
/**
+ * Set node name to marshaller context if possible.
+ *
+ * @param marsh Marshaller instance.
+ * @param nodeName Node name.
+ * @return Marshaller instance.
+ */
+ public static Marshaller withNodeName(Marshaller marsh, @Nullable String nodeName) {
+ if (marsh instanceof AbstractNodeNameAwareMarshaller)
+ ((AbstractNodeNameAwareMarshaller)marsh).nodeName(nodeName);
+
+ return marsh;
+ }
+
+ /**
* Marshal object with node name taken from provided kernal context.
*
* @param ctx Kernal context.
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/spi/checkpoint/sharedfs/SharedFsCheckpointSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/checkpoint/sharedfs/SharedFsCheckpointSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/checkpoint/sharedfs/SharedFsCheckpointSpi.java
index 0dad446..c5a8c4c 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/checkpoint/sharedfs/SharedFsCheckpointSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/checkpoint/sharedfs/SharedFsCheckpointSpi.java
@@ -34,8 +34,8 @@ import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.marshaller.AbstractNodeNameAwareMarshaller;
import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.marshaller.MarshallerUtils;
import org.apache.ignite.marshaller.jdk.JdkMarshaller;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.resources.LoggerResource;
@@ -218,8 +218,7 @@ public class SharedFsCheckpointSpi extends IgniteSpiAdapter implements Checkpoin
marsh = ignite.configuration().getMarshaller() instanceof BinaryMarshaller ? new JdkMarshaller() :
ignite.configuration().getMarshaller();
- if (marsh instanceof AbstractNodeNameAwareMarshaller)
- ((AbstractNodeNameAwareMarshaller)marsh).nodeName(ignite.name());
+ MarshallerUtils.withNodeName(marsh, ignite.name());
folder = getNextSharedPath();
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
index d049314..9dd738e 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
@@ -339,7 +339,7 @@ abstract class TcpDiscoveryImpl {
* @throws IgniteCheckedException If failed.
*/
protected byte[] marshal(Object obj) throws IgniteCheckedException {
- return MarshallerUtils.marshal(spi.ignite().name(), spi.marsh, obj);
+ return MarshallerUtils.withNodeName(spi.marsh, spi.ignite().name()).marshal(obj);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index 1fb5482..abfa193 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -1378,7 +1378,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements DiscoverySpi, T
IgniteCheckedException err = null;
try {
- MarshallerUtils.marshal(ignite.name(), marsh, msg, out);
+ MarshallerUtils.withNodeName(marsh, ignite.name()).marshal(msg, out);
}
catch (IgniteCheckedException e) {
err = e;
@@ -1678,7 +1678,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements DiscoverySpi, T
for (Map.Entry<Integer, Serializable> entry : data.entrySet()) {
try {
- byte[] bytes = MarshallerUtils.marshal(ignite.name(), marsh, entry.getValue());
+ byte[] bytes = marsh.marshal(entry.getValue());
data0.put(entry.getKey(), bytes);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
index ba19ade..ff3d4fb 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
@@ -701,7 +701,7 @@ public class TcpDiscoveryMulticastIpFinder extends TcpDiscoveryVmIpFinder {
private AddressResponse(Collection<InetSocketAddress> addrs, final String gridName) throws IgniteCheckedException {
this.addrs = addrs;
- byte[] addrsData = MarshallerUtils.marshal(gridName, marsh, addrs);
+ byte[] addrsData = MarshallerUtils.withNodeName(marsh, gridName).marshal(addrs);
data = new byte[U.IGNITE_HEADER.length + addrsData.length];
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopMarshallerFilter.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopMarshallerFilter.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopMarshallerFilter.java
index 2c5c122..e71f24d 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopMarshallerFilter.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/taskexecutor/external/communication/HadoopMarshallerFilter.java
@@ -60,7 +60,7 @@ public class HadoopMarshallerFilter extends GridNioFilterAdapter {
@Override public GridNioFuture<?> onSessionWrite(GridNioSession ses, Object msg) throws IgniteCheckedException {
assert msg instanceof HadoopMessage : "Invalid message type: " + msg;
- return proceedSessionWrite(ses, MarshallerUtils.marshal(ses.gridName(), marshaller, msg));
+ return proceedSessionWrite(ses, MarshallerUtils.withNodeName(marshaller, ses.gridName()).marshal(msg));
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/07b5a8b9/modules/web/src/main/java/org/apache/ignite/cache/websession/WebSessionV2.java
----------------------------------------------------------------------
diff --git a/modules/web/src/main/java/org/apache/ignite/cache/websession/WebSessionV2.java b/modules/web/src/main/java/org/apache/ignite/cache/websession/WebSessionV2.java
index 2e2531d..dd191b6 100644
--- a/modules/web/src/main/java/org/apache/ignite/cache/websession/WebSessionV2.java
+++ b/modules/web/src/main/java/org/apache/ignite/cache/websession/WebSessionV2.java
@@ -358,7 +358,7 @@ class WebSessionV2 implements HttpSession {
@Nullable private byte[] marshal(final Object obj) throws IOException {
if (marshaller != null) {
try {
- return MarshallerUtils.marshal(gridName, marshaller, obj);
+ return MarshallerUtils.withNodeName(marshaller, gridName).marshal(obj);
}
catch (IgniteCheckedException e) {
throw new IOException(e);