You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/05/15 16:02:19 UTC
[17/47] incubator-ignite git commit: # IGNITE-831 Create
DiscoverySpiCustomMessage.
# IGNITE-831 Create DiscoverySpiCustomMessage.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/359680db
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/359680db
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/359680db
Branch: refs/heads/ignite-709_3
Commit: 359680dbd918f7459dc8dda7e6084a05a5ee7b42
Parents: d0dac7d
Author: sevdokimov <se...@gridgain.com>
Authored: Wed May 6 15:09:40 2015 +0300
Committer: sevdokimov <se...@gridgain.com>
Committed: Wed May 6 15:09:40 2015 +0300
----------------------------------------------------------------------
.../internal/events/DiscoveryCustomEvent.java | 14 ++++++------
.../discovery/CustomMessageWrapper.java | 5 ----
.../discovery/GridDiscoveryManager.java | 24 +++++++++++---------
.../GridCachePartitionExchangeManager.java | 4 ++--
.../discovery/DiscoverySpiCustomMessage.java | 7 ------
.../spi/discovery/DiscoverySpiListener.java | 5 ++--
.../discovery/tcp/TcpClientDiscoverySpi.java | 4 ++--
.../spi/discovery/tcp/TcpDiscoverySpi.java | 2 +-
.../discovery/AbstractDiscoverySelfTest.java | 8 ++++---
9 files changed, 32 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/main/java/org/apache/ignite/internal/events/DiscoveryCustomEvent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/events/DiscoveryCustomEvent.java b/modules/core/src/main/java/org/apache/ignite/internal/events/DiscoveryCustomEvent.java
index ee32692..ad33aae 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/events/DiscoveryCustomEvent.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/events/DiscoveryCustomEvent.java
@@ -34,7 +34,7 @@ public class DiscoveryCustomEvent extends DiscoveryEvent {
/**
* Built-in event type: custom event sent.
* <br>
- * Generated when someone invoke {@link GridDiscoveryManager#sendCustomEvent(Serializable)}.
+ * Generated when someone invoke {@link GridDiscoveryManager#sendCustomEvent(DiscoveryCustomMessage)}.
* <p>
*
* @see DiscoveryCustomEvent
@@ -42,7 +42,7 @@ public class DiscoveryCustomEvent extends DiscoveryEvent {
public static final int EVT_DISCOVERY_CUSTOM_EVT = 18;
/** */
- private Serializable data;
+ private DiscoveryCustomMessage customMsg;
/** Affinity topology version. */
private AffinityTopologyVersion affTopVer;
@@ -57,15 +57,15 @@ public class DiscoveryCustomEvent extends DiscoveryEvent {
/**
* @return Data.
*/
- public Serializable data() {
- return data;
+ public DiscoveryCustomMessage customMessage() {
+ return customMsg;
}
/**
- * @param data New data.
+ * @param customMsg New customMessage.
*/
- public void data(Serializable data) {
- this.data = data;
+ public void customMessage(DiscoveryCustomMessage customMsg) {
+ this.customMsg = customMsg;
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java
index f394fe1..9575cc7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java
@@ -35,11 +35,6 @@ class CustomMessageWrapper implements DiscoverySpiCustomMessage {
}
/** {@inheritDoc} */
- @Override public boolean forwardMinorVersion() {
- return delegate.forwardMinorVersion();
- }
-
- /** {@inheritDoc} */
@Nullable @Override public DiscoverySpiCustomMessage newMessageOnRingEnd() {
DiscoveryCustomMessage res = delegate.newMessageOnRingEnd();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 5533fa1..c5698a8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -344,8 +344,11 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
ClusterNode node,
Collection<ClusterNode> topSnapshot,
Map<Long, Collection<ClusterNode>> snapshots,
- @Nullable Serializable data
+ @Nullable DiscoverySpiCustomMessage spiCustomMsg
) {
+ DiscoveryCustomMessage customMsg = spiCustomMsg == null ? null
+ : ((CustomMessageWrapper)spiCustomMsg).delegate();
+
final ClusterNode locNode = localNode();
if (snapshots != null)
@@ -356,7 +359,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
if (type == EVT_NODE_METRICS_UPDATED)
verChanged = false;
else if (type == DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT) {
- if (data != null && ((DiscoverySpiCustomMessage)data).forwardMinorVersion()) {
+ if (customMsg != null && customMsg.forwardMinorVersion()) {
minorTopVer++;
verChanged = true;
@@ -380,9 +383,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
if (type == DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT) {
- if (data != null) {
- DiscoveryCustomMessage customMsg = ((CustomMessageWrapper)data).delegate();
-
+ if (customMsg != null) {
for (Class cls = customMsg.getClass(); cls != null; cls = cls.getSuperclass()) {
List<CustomEventListener<DiscoveryCustomMessage>> list = customEvtLsnrs.get(cls);
@@ -435,7 +436,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
return;
}
- discoWrk.addEvent(type, nextTopVer, node, topSnapshot, data);
+ discoWrk.addEvent(type, nextTopVer, node, topSnapshot, customMsg);
}
});
@@ -1567,8 +1568,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
/** Worker for discovery events. */
private class DiscoveryWorker extends GridWorker {
/** Event queue. */
- private final BlockingQueue<GridTuple5<Integer, AffinityTopologyVersion, ClusterNode, Collection<ClusterNode>, Serializable>> evts =
- new LinkedBlockingQueue<>();
+ private final BlockingQueue<GridTuple5<Integer, AffinityTopologyVersion, ClusterNode, Collection<ClusterNode>,
+ DiscoveryCustomMessage>> evts = new LinkedBlockingQueue<>();
/** Node segmented event fired flag. */
private boolean nodeSegFired;
@@ -1634,7 +1635,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
AffinityTopologyVersion topVer,
ClusterNode node,
Collection<ClusterNode> topSnapshot,
- @Nullable Serializable data
+ @Nullable DiscoveryCustomMessage data
) {
assert node != null : data;
@@ -1675,7 +1676,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
/** @throws InterruptedException If interrupted. */
@SuppressWarnings("DuplicateCondition")
private void body0() throws InterruptedException {
- GridTuple5<Integer, AffinityTopologyVersion, ClusterNode, Collection<ClusterNode>, Serializable> evt = evts.take();
+ GridTuple5<Integer, AffinityTopologyVersion, ClusterNode, Collection<ClusterNode>,
+ DiscoveryCustomMessage> evt = evts.take();
int type = evt.get1();
@@ -1793,7 +1795,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
customEvt.type(type);
customEvt.topologySnapshot(topVer.topologyVersion(), null);
customEvt.affinityTopologyVersion(topVer);
- customEvt.data(evt.get5());
+ customEvt.customMessage(evt.get5());
ctx.event().record(customEvt);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 5f82ae2..c7aa322 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -150,8 +150,8 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
else {
DiscoveryCustomEvent customEvt = (DiscoveryCustomEvent)e;
- if (customEvt.data() instanceof DynamicCacheChangeBatch) {
- DynamicCacheChangeBatch batch = (DynamicCacheChangeBatch)customEvt.data();
+ if (customEvt.customMessage() instanceof DynamicCacheChangeBatch) {
+ DynamicCacheChangeBatch batch = (DynamicCacheChangeBatch)customEvt.customMessage();
Collection<DynamicCacheChangeRequest> valid = new ArrayList<>(batch.requests().size());
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java
index 1550613..8d92deb 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java
@@ -26,13 +26,6 @@ import java.io.*;
*/
public interface DiscoverySpiCustomMessage extends Serializable {
/**
- * Whether or not minor version of topology should be increased on message receive.
- *
- * @return {@code true} if minor topology version should be increased.
- */
- public boolean forwardMinorVersion();
-
- /**
* Called when message passed the ring.
*/
@Nullable public DiscoverySpiCustomMessage newMessageOnRingEnd();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiListener.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiListener.java
index 7f17fe4..f46869d 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiListener.java
@@ -18,10 +18,9 @@
package org.apache.ignite.spi.discovery;
import org.apache.ignite.cluster.*;
-import org.apache.ignite.events.DiscoveryEvent;
+import org.apache.ignite.events.*;
import org.jetbrains.annotations.*;
-import java.io.*;
import java.util.*;
/**
@@ -47,5 +46,5 @@ public interface DiscoverySpiListener {
ClusterNode node,
Collection<ClusterNode> topSnapshot,
@Nullable Map<Long, Collection<ClusterNode>> topHist,
- @Nullable Serializable data);
+ @Nullable DiscoverySpiCustomMessage data);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
index 3e839f8..5ce7437 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
@@ -1332,7 +1332,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
if (node != null && node.visible()) {
try {
- Serializable msgObj = marsh.unmarshal(msg.messageBytes(), U.gridClassLoader());
+ DiscoverySpiCustomMessage msgObj = marsh.unmarshal(msg.messageBytes(), U.gridClassLoader());
notifyDiscovery(EVT_DISCOVERY_CUSTOM_EVT, topVer, node, allNodes(), msgObj);
}
@@ -1433,7 +1433,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
* @param top Topology snapshot.
*/
private void notifyDiscovery(int type, long topVer, ClusterNode node, Collection<ClusterNode> top,
- @Nullable Serializable data) {
+ @Nullable DiscoverySpiCustomMessage data) {
DiscoverySpiListener lsnr = TcpClientDiscoverySpi.this.lsnr;
if (lsnr != null) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/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 8051172..40dca05 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
@@ -4524,7 +4524,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
if (node != null) {
try {
- Serializable msgObj = marsh.unmarshal(msg.messageBytes(), U.gridClassLoader());
+ DiscoverySpiCustomMessage msgObj = marsh.unmarshal(msg.messageBytes(), U.gridClassLoader());
lsnr.onDiscovery(DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT,
msg.topologyVersion(),
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/359680db/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
index 3c61f00..9c6fbb4 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java
@@ -25,6 +25,7 @@ import org.apache.ignite.spi.*;
import org.apache.ignite.testframework.config.*;
import org.apache.ignite.testframework.junits.*;
import org.apache.ignite.testframework.junits.spi.*;
+import org.jetbrains.annotations.*;
import javax.management.*;
import java.io.*;
@@ -132,7 +133,7 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri
/** {@inheritDoc} */
@Override public void onDiscovery(int type, long topVer, ClusterNode node, Collection<ClusterNode> topSnapshot,
- Map<Long, Collection<ClusterNode>> topHist, Serializable data) {
+ Map<Long, Collection<ClusterNode>> topHist, @Nullable DiscoverySpiCustomMessage data) {
if (type == EVT_NODE_METRICS_UPDATED)
isMetricsUpdate = true;
}
@@ -205,7 +206,7 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri
DiscoverySpiListener locHeartbeatLsnr = new DiscoverySpiListener() {
@Override public void onDiscovery(int type, long topVer, ClusterNode node,
Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>> topHist,
- Serializable data) {
+ @Nullable DiscoverySpiCustomMessage data) {
// If METRICS_UPDATED came from local node
if (type == EVT_NODE_METRICS_UPDATED
&& node.id().equals(spi.getLocalNode().id()))
@@ -369,7 +370,8 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri
spi.setListener(new DiscoverySpiListener() {
@SuppressWarnings({"NakedNotify"})
@Override public void onDiscovery(int type, long topVer, ClusterNode node,
- Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>> topHist, Serializable data) {
+ Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>> topHist,
+ @Nullable DiscoverySpiCustomMessage data) {
info("Discovery event [type=" + type + ", node=" + node + ']');
synchronized (mux) {