ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [27/50] [abbrv] incubator-ignite git commit: ignite-312 review
Date Fri, 27 Feb 2015 01:11:34 GMT
ignite-312 review


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

Branch: refs/heads/ignite-264
Commit: 2bfb89322473b6579a59d114b45eb6e9532df4d6
Parents: 7423b7c
Author: Yakov Zhdanov <yzhdanov@gridgain.com>
Authored: Wed Feb 25 22:07:59 2015 +0300
Committer: Yakov Zhdanov <yzhdanov@gridgain.com>
Committed: Wed Feb 25 22:07:59 2015 +0300

----------------------------------------------------------------------
 .../ignite/events/DiscoveryCustomEvent.java     |  56 +++
 .../org/apache/ignite/events/EventType.java     |  12 +
 .../discovery/GridDiscoveryManager.java         |  78 ++--
 .../ignite/internal/util/IgniteUtils.java       | 364 ++-----------------
 .../ignite/spi/discovery/DiscoverySpi.java      |   7 +
 .../spi/discovery/DiscoverySpiListener.java     |  11 +-
 .../discovery/tcp/TcpClientDiscoverySpi.java    |   7 +-
 .../spi/discovery/tcp/TcpDiscoverySpi.java      |  39 +-
 .../discovery/tcp/TcpDiscoverySpiAdapter.java   |   3 +
 .../TcpDiscoveryCustomEventMessage.java         |  66 ++++
 .../internal/GridDiscoveryEventSelfTest.java    |  43 +++
 .../discovery/AbstractDiscoverySelfTest.java    |   7 +-
 12 files changed, 321 insertions(+), 372 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/modules/core/src/main/java/org/apache/ignite/events/DiscoveryCustomEvent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/events/DiscoveryCustomEvent.java
b/modules/core/src/main/java/org/apache/ignite/events/DiscoveryCustomEvent.java
new file mode 100644
index 0000000..af15055
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/events/DiscoveryCustomEvent.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.events;
+
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.io.*;
+
+/**
+ *
+ */
+public class DiscoveryCustomEvent extends DiscoveryEvent {
+    /** */
+    private Serializable data;
+
+    /**
+     * Default constructor.
+     */
+    public DiscoveryCustomEvent() {
+        type(EventType.EVT_DISCOVERY_CUSTOM_EVT);
+    }
+
+    /**
+     * @return Data.
+     */
+    public Serializable data() {
+        return data;
+    }
+
+    /**
+     * @param data New data.
+     */
+    public void data(Serializable data) {
+        this.data = data;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(DiscoveryCustomEvent.class, this, super.toString());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/modules/core/src/main/java/org/apache/ignite/events/EventType.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/events/EventType.java b/modules/core/src/main/java/org/apache/ignite/events/EventType.java
index 9f7e2c4..d9ffc2a 100644
--- a/modules/core/src/main/java/org/apache/ignite/events/EventType.java
+++ b/modules/core/src/main/java/org/apache/ignite/events/EventType.java
@@ -17,8 +17,10 @@
 
 package org.apache.ignite.events;
 
+import org.apache.ignite.internal.managers.discovery.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 
+import java.io.*;
 import java.util.*;
 
 /**
@@ -158,6 +160,16 @@ public interface EventType {
     public static final int EVT_CLIENT_NODE_RECONNECTED = 17;
 
     /**
+     * Built-in event type: custom event sent.
+     * <br>
+     * Generated when someone invoke {@link GridDiscoveryManager#sendCustomEvent(Serializable)}.
+     * <p>
+     *
+     * @see DiscoveryCustomEvent
+     */
+    public static final int EVT_DISCOVERY_CUSTOM_EVT = 18;
+
+    /**
      * Built-in event type: task started.
      * <p>
      * NOTE: all types in range <b>from 1 to 1000 are reserved</b> for

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/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 711229a..b65da50 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
@@ -255,8 +255,14 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
         }
 
         getSpi().setListener(new DiscoverySpiListener() {
-            @Override public void onDiscovery(int type, long topVer, ClusterNode node, Collection<ClusterNode>
topSnapshot,
-                Map<Long, Collection<ClusterNode>> snapshots) {
+            @Override public void onDiscovery(
+                int type,
+                long topVer,
+                ClusterNode node,
+                Collection<ClusterNode> topSnapshot,
+                Map<Long, Collection<ClusterNode>> snapshots,
+                @Nullable Serializable data
+            ) {
                 final ClusterNode locNode = localNode();
 
                 if (snapshots != null)
@@ -270,7 +276,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
                 // Put topology snapshot into discovery history.
                 // There is no race possible between history maintenance and concurrent discovery
                 // event notifications, since SPI notifies manager about all events from
this listener.
-                if (type != EVT_NODE_METRICS_UPDATED) {
+                if (type != EVT_NODE_METRICS_UPDATED && type != EVT_DISCOVERY_CUSTOM_EVT)
{
                     DiscoCache cache = new DiscoCache(locNode, F.view(topSnapshot, F.remoteNodes(locNode.id())));
 
                     discoCacheHist.put(topVer, cache);
@@ -305,7 +311,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
                         ", evt=" + U.gridEventName(type) + ']';
                 }
 
-                discoWrk.addEvent(type, topVer, node, topSnapshot);
+                discoWrk.addEvent(type, topVer, node, topSnapshot, data);
             }
         });
 
@@ -778,12 +784,11 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
         // Stop receiving notifications.
         getSpi().setListener(null);
 
-        // Stop discovery worker.
+        // Stop discovery worker and metrics updater.
         U.cancel(discoWrk);
-        U.join(discoWrk, log);
-
-        // Stop metrics updater.
         U.cancel(metricsUpdater);
+
+        U.join(discoWrk, log);
         U.join(metricsUpdater, log);
 
         // Stop SPI itself.
@@ -1185,6 +1190,13 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
         ).start();
     }
 
+    /**
+     * @param evt Event.
+     */
+    public void sendCustomEvent(Serializable evt) {
+        getSpi().sendCustomEvent(evt);
+    }
+
     /** Worker for network segment checks. */
     private class SegmentCheckWorker extends GridWorker {
         /** */
@@ -1241,7 +1253,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
 
                     if (!segValid) {
                         discoWrk.addEvent(EVT_NODE_SEGMENTED, 0, getSpi().getLocalNode(),
-                            Collections.<ClusterNode>emptyList());
+                            Collections.<ClusterNode>emptyList(), null);
 
                         lastSegChkRes.set(false);
                     }
@@ -1261,7 +1273,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
     /** Worker for discovery events. */
     private class DiscoveryWorker extends GridWorker {
         /** Event queue. */
-        private final BlockingQueue<GridTuple4<Integer, Long, ClusterNode, Collection<ClusterNode>>>
evts =
+        private final BlockingQueue<GridTuple5<Integer, Long, ClusterNode, Collection<ClusterNode>,
Serializable>> evts =
             new LinkedBlockingQueue<>();
 
         /** Node segmented event fired flag. */
@@ -1292,12 +1304,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
                 evt.eventNode(node);
                 evt.type(type);
 
-                evt.topologySnapshot(topVer, new ArrayList<>(
-                    F.viewReadOnly(topSnapshot, new C1<ClusterNode, ClusterNode>()
{
-                        @Override public ClusterNode apply(ClusterNode e) {
-                            return e;
-                        }
-                    }, daemonFilter)));
+                evt.topologySnapshot(topVer, U.<ClusterNode, ClusterNode>arrayList(topSnapshot,
daemonFilter));
 
                 if (type == EVT_NODE_METRICS_UPDATED)
                     evt.message("Metrics were updated: " + node);
@@ -1327,10 +1334,16 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
          * @param node Node.
          * @param topSnapshot Topology snapshot.
          */
-        void addEvent(int type, long topVer, ClusterNode node, Collection<ClusterNode>
topSnapshot) {
+        void addEvent(
+            int type,
+            long topVer,
+            ClusterNode node,
+            Collection<ClusterNode> topSnapshot,
+            @Nullable Serializable data
+        ) {
             assert node != null;
 
-            evts.add(F.t(type, topVer, node, topSnapshot));
+            evts.add(F.t(type, topVer, node, topSnapshot, data));
         }
 
         /**
@@ -1364,7 +1377,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
         /** @throws InterruptedException If interrupted. */
         @SuppressWarnings("DuplicateCondition")
         private void body0() throws InterruptedException {
-            GridTuple4<Integer, Long, ClusterNode, Collection<ClusterNode>> evt
= evts.take();
+            GridTuple5<Integer, Long, ClusterNode, Collection<ClusterNode>, Serializable>
evt = evts.take();
 
             int type = evt.get1();
 
@@ -1473,6 +1486,22 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
                     break;
                 }
 
+                case EVT_DISCOVERY_CUSTOM_EVT: {
+                    DiscoveryCustomEvent customEvt = new DiscoveryCustomEvent();
+
+                    customEvt.node(ctx.discovery().localNode());
+                    customEvt.eventNode(node);
+                    customEvt.type(type);
+                    customEvt.topologySnapshot(topVer, null);
+                    customEvt.data(evt.get5());
+
+                    assert ctx.event().isRecordable(EVT_DISCOVERY_CUSTOM_EVT);
+
+                    ctx.event().record(customEvt);
+
+                    return;
+                }
+
                 // Don't log metric update to avoid flooding the log.
                 case EVT_NODE_METRICS_UPDATED:
                     break;
@@ -2058,12 +2087,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
          * @param exclNode Node to exclude.
          */
         private void filterNodeMap(ConcurrentMap<String, Collection<ClusterNode>>
map, final ClusterNode exclNode) {
-            IgnitePredicate<ClusterNode> p = new P1<ClusterNode>() {
-                @Override public boolean apply(ClusterNode e) {
-                    return exclNode.equals(e);
-                }
-            };
-
             for (String cacheName : U.cacheNames(exclNode)) {
                 String maskedName = maskNull(cacheName);
 
@@ -2073,7 +2096,10 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi>
{
                     if (oldNodes == null || oldNodes.isEmpty())
                         break;
 
-                    Collection<ClusterNode> newNodes = F.lose(oldNodes, true, p);
+                    Collection<ClusterNode> newNodes = new ArrayList<>(oldNodes);
+
+                    if (!newNodes.remove(exclNode))
+                        break;
 
                     if (map.replace(maskedName, oldNodes, newNodes))
                         break;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index e4b23f1..b9c7596 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -59,7 +59,6 @@ import java.math.*;
 import java.net.*;
 import java.nio.*;
 import java.nio.channels.*;
-import java.nio.channels.spi.*;
 import java.nio.charset.*;
 import java.security.*;
 import java.security.cert.*;
@@ -647,15 +646,7 @@ public abstract class IgniteUtils {
      * @return Nearest power of 2.
      */
     public static int ceilPow2(int v) {
-        v--;
-
-        v |= v >> 1;
-        v |= v >> 2;
-        v |= v >> 4;
-        v |= v >> 8;
-        v |= v >> 16;
-
-        return ++v;
+        return Integer.highestOneBit(v - 1) << 1;
     }
 
     /**
@@ -1633,13 +1624,11 @@ public abstract class IgniteUtils {
                     if (!itf.isLoopback()) {
                         Enumeration<InetAddress> addrs = itf.getInetAddresses();
 
-                        if (addrs != null) {
-                            for (InetAddress addr : asIterable(addrs)) {
-                                String hostAddr = addr.getHostAddress();
+                        for (InetAddress addr : asIterable(addrs)) {
+                            String hostAddr = addr.getHostAddress();
 
-                                if (!addr.isLoopbackAddress() && !ips.contains(hostAddr))
-                                    ips.add(hostAddr);
-                            }
+                            if (!addr.isLoopbackAddress() && !ips.contains(hostAddr))
+                                ips.add(hostAddr);
                         }
                     }
                 }
@@ -2832,15 +2821,8 @@ public abstract class IgniteUtils {
         if (s == null)
             return;
 
-        OutputStream out = null;
-
-        try {
-            out = new FileOutputStream(file, append);
-
-            if (s != null)
-                out.write(s.getBytes(charset));
-        } finally {
-            closeQuiet(out);
+        try (OutputStream out = new FileOutputStream(file, append)) {
+            out.write(s.getBytes(charset));
         }
     }
 
@@ -3311,35 +3293,17 @@ public abstract class IgniteUtils {
     }
 
     /**
-     * Checks for containment of value matching given regular expression in the provided
array.
-     *
-     * @param arr Array of strings.
-     * @param regex Regular expression.
-     * @return {@code true} if string matching given regular expression found, {@code false}
otherwise.
-     */
-    public static boolean containsRegexArray(String[] arr, String regex) {
-        assert arr != null;
-        assert regex != null;
-
-        for (String s : arr)
-            if (s != null && s.matches(regex))
-                return true;
-
-        return false;
-    }
-
-    /**
      * Closes given resource logging possible checked exception.
      *
      * @param rsrc Resource to close. If it's {@code null} - it's no-op.
      * @param log Logger to log possible checked exception with (optional).
      */
-    public static void close(@Nullable Closeable rsrc, @Nullable IgniteLogger log) {
+    public static void close(@Nullable AutoCloseable rsrc, @Nullable IgniteLogger log) {
         if (rsrc != null)
             try {
                 rsrc.close();
             }
-            catch (IOException e) {
+            catch (Exception e) {
                 warn(log, "Failed to close resource: " + e.getMessage());
             }
     }
@@ -3360,114 +3324,6 @@ public abstract class IgniteUtils {
     }
 
     /**
-     * Quietly closes given resource ignoring possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     */
-    public static void closeQuiet(@Nullable Closeable rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
-     * Closes given resource logging possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable Socket rsrc, @Nullable IgniteLogger log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes given resource ignoring possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     */
-    public static void closeQuiet(@Nullable Socket rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
-     * Closes given resource logging possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable ServerSocket rsrc, @Nullable IgniteLogger log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes given resource ignoring possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     */
-    public static void closeQuiet(@Nullable ServerSocket rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
-     * Closes given resource logging possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable AbstractInterruptibleChannel rsrc, @Nullable IgniteLogger
log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes given resource ignoring possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     */
-    public static void closeQuiet(@Nullable AbstractInterruptibleChannel rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
      * Closes given resource logging possible checked exceptions.
      *
      * @param rsrc Resource to close. If it's {@code null} - it's no-op.
@@ -3491,83 +3347,6 @@ public abstract class IgniteUtils {
     }
 
     /**
-     * Closes given resource logging possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable Reader rsrc, @Nullable IgniteLogger log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes given resource ignoring possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     */
-    public static void closeQuiet(@Nullable Reader rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
-     * Quietly closes given resource ignoring possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     */
-    public static void closeQuiet(@Nullable Writer rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
-     * Closes given resource logging possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable ZipFile rsrc, @Nullable IgniteLogger log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes given resource ignoring possible checked exception.
-     *
-     * @param rsrc Resource to close. If it's {@code null} - it's no-op.
-     */
-    public static void closeQuiet(@Nullable ZipFile rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (IOException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
      * Closes given resource.
      *
      * @param rsrc Resource to close. If it's {@code null} - it's no-op.
@@ -3642,99 +3421,6 @@ public abstract class IgniteUtils {
     }
 
     /**
-     * Closes JDBC connection logging possible checked exception.
-     *
-     * @param rsrc JDBC connection to close. If connection is {@code null}, it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable Connection rsrc, @Nullable IgniteLogger log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (SQLException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes JDBC connection ignoring possible checked exception.
-     *
-     * @param rsrc JDBC connection to close. If connection is {@code null}, it's no-op.
-     */
-    public static void closeQuiet(@Nullable Connection rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (SQLException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
-     * Closes JDBC statement logging possible checked exception.
-     *
-     * @param rsrc JDBC statement to close. If statement is {@code null}, it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable Statement rsrc, @Nullable IgniteLogger log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (SQLException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes JDBC statement ignoring possible checked exception.
-     *
-     * @param rsrc JDBC statement to close. If statement is {@code null}, it's no-op.
-     */
-    public static void closeQuiet(@Nullable Statement rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (SQLException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
-     * Closes JDBC result set logging possible checked exception.
-     *
-     * @param rsrc JDBC result set to close. If result set is {@code null}, it's no-op.
-     * @param log Logger to log possible checked exception with (optional).
-     */
-    public static void close(@Nullable ResultSet rsrc, @Nullable IgniteLogger log) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (SQLException e) {
-                warn(log, "Failed to close resource: " + e.getMessage());
-            }
-    }
-
-    /**
-     * Quietly closes JDBC result set ignoring possible checked exception.
-     *
-     * @param rsrc JDBC result set to close. If result set is {@code null}, it's no-op.
-     */
-    public static void closeQuiet(@Nullable ResultSet rsrc) {
-        if (rsrc != null)
-            try {
-                rsrc.close();
-            }
-            catch (SQLException ignored) {
-                // No-op.
-            }
-    }
-
-    /**
      * Closes class loader logging possible checked exception.
      * Note: this issue for problem <a href="http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=5041014">
      * http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=5041014</a>.
@@ -6316,6 +6002,10 @@ public abstract class IgniteUtils {
         return arr;
     }
 
+    /**
+     * @param arr1 Array 1.
+     * @param arr2 Array 2.
+     */
     public static int[] addAll(int[] arr1, int[] arr2) {
         int[] all = new int[arr1.length + arr2.length];
 
@@ -6920,15 +6610,7 @@ public abstract class IgniteUtils {
     public static boolean isPrimitiveArray(Object obj) {
         Class<?> cls = obj.getClass();
 
-        return cls.isArray() && (
-            cls.equals(byte[].class) ||
-                cls.equals(short[].class) ||
-                cls.equals(char[].class) ||
-                cls.equals(int[].class) ||
-                cls.equals(long[].class) ||
-                cls.equals(float[].class) ||
-                cls.equals(double[].class) ||
-                cls.equals(boolean[].class));
+        return cls.isArray() && cls.getComponentType().isPrimitive();
     }
 
     /**
@@ -8620,15 +8302,15 @@ public abstract class IgniteUtils {
      * Converts a hexadecimal character to an integer.
      *
      * @param ch A character to convert to an integer digit
-     * @param index The index of the character in the source
+     * @param idx The index of the character in the source
      * @return An integer
      * @throws IgniteCheckedException Thrown if ch is an illegal hex character
      */
-    public static int toDigit(char ch, int index) throws IgniteCheckedException {
+    public static int toDigit(char ch, int idx) throws IgniteCheckedException {
         int digit = Character.digit(ch, 16);
 
         if (digit == -1)
-            throw new IgniteCheckedException("Illegal hexadecimal character " + ch + " at
index " + index);
+            throw new IgniteCheckedException("Illegal hexadecimal character " + ch + " at
index " + idx);
 
         return digit;
     }
@@ -9028,6 +8710,16 @@ public abstract class IgniteUtils {
 
     /**
      * @param c Collection.
+     * @return Resulting array list.
+     */
+    public static <T extends R, R> List<R> arrayList(Collection<T> c) {
+        assert c != null;
+
+        return new ArrayList<R>(c);
+    }
+
+    /**
+     * @param c Collection.
      * @param cap Initial capacity.
      * @param p Optional filters.
      * @return Resulting array list.
@@ -9037,7 +8729,7 @@ public abstract class IgniteUtils {
         assert c != null;
         assert cap >= 0;
 
-        ArrayList<R> list = new ArrayList<>(cap);
+        List<R> list = new ArrayList<>(cap);
 
         for (T t : c) {
             if (F.isAll(t, p))

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpi.java
index 168ae52..42bad22 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpi.java
@@ -22,6 +22,7 @@ import org.apache.ignite.lang.*;
 import org.apache.ignite.spi.*;
 import org.jetbrains.annotations.*;
 
+import java.io.*;
 import java.util.*;
 
 /**
@@ -137,4 +138,10 @@ public interface DiscoverySpi extends IgniteSpi {
      *         does not support this method.
      */
     public long getGridStartTime();
+
+    /**
+     * Sends custom message across the ring.
+     * @param evt Event.
+     */
+    public void sendCustomEvent(Serializable evt);
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/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 3e0ef02..243aaeb 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
@@ -20,6 +20,7 @@ package org.apache.ignite.spi.discovery;
 import org.apache.ignite.cluster.*;
 import org.jetbrains.annotations.*;
 
+import java.io.*;
 import java.util.*;
 
 /**
@@ -37,7 +38,13 @@ public interface DiscoverySpiListener {
      * @param topSnapshot Topology snapshot after event has been occurred (e.g. if event
is
      *      {@code EVT_NODE_JOINED}, then joined node will be in snapshot).
      * @param topHist Topology snapshots history.
+     * @param data Data for custom event.
      */
-    public void onDiscovery(int type, long topVer, ClusterNode node, Collection<ClusterNode>
topSnapshot,
-        @Nullable Map<Long, Collection<ClusterNode>> topHist);
+    public void onDiscovery(
+        int type,
+        long topVer,
+        ClusterNode node,
+        Collection<ClusterNode> topSnapshot,
+        @Nullable Map<Long, Collection<ClusterNode>> topHist,
+        @Nullable Serializable data);
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/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 51df9db..17d8386 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
@@ -373,6 +373,11 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements
Tcp
         // No-op.
     }
 
+    /** {@inheritDoc} */
+    @Override public void sendCustomEvent(Serializable evt) {
+        throw new UnsupportedOperationException();
+    }
+
     /**
      * @param recon Reconnect flag.
      * @return Whether joined successfully.
@@ -1220,7 +1225,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements
Tcp
                     log.debug("Discovery notification [node=" + node + ", type=" + U.gridEventName(type)
+
                         ", topVer=" + topVer + ']');
 
-                lsnr.onDiscovery(type, topVer, node, top, new TreeMap<>(topHist));
+                lsnr.onDiscovery(type, topVer, node, top, new TreeMap<>(topHist), null);
             }
             else if (log.isDebugEnabled())
                 log.debug("Skipped discovery notification [node=" + node + ", type=" + U.gridEventName(type)
+

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/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 87aebc4..f910556 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
@@ -1028,7 +1028,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
 
                     Map<Long, Collection<ClusterNode>> hist = updateTopologyHistory(topVer,
top);
 
-                    lsnr.onDiscovery(EVT_NODE_FAILED, topVer, n, top, hist);
+                    lsnr.onDiscovery(EVT_NODE_FAILED, topVer, n, top, hist, null);
                 }
             }
         }
@@ -1242,6 +1242,11 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
         this.nodeAuth = nodeAuth;
     }
 
+    /** {@inheritDoc} */
+    @Override public void sendCustomEvent(Serializable evt) {
+        msgWorker.addMessage(new TcpDiscoveryCustomEventMessage(getLocalNodeId(), evt));
+    }
+
     /**
      * Tries to join this node to topology.
      *
@@ -1714,11 +1719,11 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
                 log.debug("Discovery notification [node=" + node + ", spiState=" + spiState
+
                     ", type=" + U.gridEventName(type) + ", topVer=" + topVer + ']');
 
-            Collection<ClusterNode> top = F.<TcpDiscoveryNode, ClusterNode>upcast(ring.visibleNodes());
+            Collection<ClusterNode> top = F.upcast(ring.visibleNodes());
 
             Map<Long, Collection<ClusterNode>> hist = updateTopologyHistory(topVer,
top);
 
-            lsnr.onDiscovery(type, topVer, node, top, hist);
+            lsnr.onDiscovery(type, topVer, node, top, hist, null);
         }
         else if (log.isDebugEnabled())
             log.debug("Skipped discovery notification [node=" + node + ", spiState=" + spiState
+
@@ -2563,6 +2568,9 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
             else if (msg instanceof TcpDiscoveryDiscardMessage)
                 processDiscardMessage((TcpDiscoveryDiscardMessage)msg);
 
+            else if (msg instanceof TcpDiscoveryCustomEventMessage)
+                processCustomMessage((TcpDiscoveryCustomEventMessage)msg);
+
             else
                 assert false : "Unknown message type: " + msg.getClass().getSimpleName();
 
@@ -4429,6 +4437,29 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
             if (ring.hasRemoteNodes())
                 sendMessageAcrossRing(msg);
         }
+
+        /**
+         * @param msg Message.
+         */
+        private void processCustomMessage(TcpDiscoveryCustomEventMessage msg) {
+            if (msg.creatorNodeId().equals(getLocalNodeId())) {
+                if (msg.senderNodeId() != null)
+                    return;
+
+                msg.senderNodeId(getLocalNodeId());
+            }
+
+            DiscoverySpiListener lsnr = TcpDiscoverySpi.this.lsnr;
+
+            TcpDiscoverySpiState spiState = spiStateCopy();
+
+            if (lsnr != null && (spiState == CONNECTED || spiState == DISCONNECTING))
+                lsnr.onDiscovery(EVT_DISCOVERY_CUSTOM_EVT, msg.topologyVersion(), ring.node(msg.creatorNodeId()),
null,
+                    null, msg.message());
+
+            if (ring.hasRemoteNodes())
+                sendMessageAcrossRing(msg);
+        }
     }
 
     /**
@@ -4596,7 +4627,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
                         }
                     }
 
-                    if (!U.bytesEqual(buf, 0, U.IGNITE_HEADER, 0, 4)) {
+                    if (!Arrays.equals(buf, U.IGNITE_HEADER)) {
                         if (log.isDebugEnabled())
                             log.debug("Unknown connection detected (is some other software
connecting to " +
                                 "this Ignite port?) " +

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
index 52156a4..e47e490 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
@@ -990,6 +990,9 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements
Discov
                 log.debug("Message has been added to queue: " + msg);
         }
 
+        /**
+         * @param msg Message.
+         */
         protected abstract void processMessage(TcpDiscoveryAbstractMessage msg);
 
         /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java
new file mode 100644
index 0000000..f83765a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.discovery.tcp.messages;
+
+import java.io.*;
+import java.util.*;
+
+/**
+ * Wrapped for custom message.
+ */
+public class TcpDiscoveryCustomEventMessage extends TcpDiscoveryAbstractMessage {
+    /** */
+    private Serializable msg;
+
+    /**
+     * Public default no-arg constructor for {@link Externalizable} interface.
+     */
+    public TcpDiscoveryCustomEventMessage() {
+        // No-op.
+    }
+
+    /**
+     * @param creatorNodeId Creator node id.
+     */
+    public TcpDiscoveryCustomEventMessage(UUID creatorNodeId, Serializable msg) {
+        super(creatorNodeId);
+
+        this.msg = msg;
+    }
+
+    /**
+     * @return Message.
+     */
+    public Serializable message() {
+        return msg;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        super.writeExternal(out);
+
+        out.writeObject(msg);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+        super.readExternal(in);
+
+        msg = (Serializable)in.readObject();
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
index ce8b3e9..64f939c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridDiscoveryEventSelfTest.java
@@ -417,4 +417,47 @@ public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest
{
             stopAllGrids();
         }
     }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testCustomEvents() throws Exception {
+        try {
+            Ignite g0 = startGrid(0);
+            final Ignite g1 = startGrid(1);
+            Ignite g2 = startGrid(2);
+
+            final CountDownLatch cnt = new CountDownLatch(3);
+
+            IgnitePredicate<DiscoveryCustomEvent> lsnr = new IgnitePredicate<DiscoveryCustomEvent>()
{
+                @Override public boolean apply(DiscoveryCustomEvent evt) {
+                    assert cnt.getCount() > 0;
+
+                    cnt.countDown();
+
+                    return true;
+                }
+            };
+
+            g0.events().localListen(lsnr, EVT_DISCOVERY_CUSTOM_EVT);
+            g1.events().localListen(lsnr, EVT_DISCOVERY_CUSTOM_EVT);
+            g2.events().localListen(lsnr, EVT_DISCOVERY_CUSTOM_EVT);
+
+            ((IgniteKernal)g1).context().discovery().sendCustomEvent("a");
+
+            cnt.await();
+
+            g0.events().localQuery(new IgnitePredicate<DiscoveryCustomEvent>() {
+                @Override public boolean apply(DiscoveryCustomEvent evt) {
+                    assert "a".equals(evt.data());
+                    assert ((IgniteEx)g1).localNode().id().equals(evt.eventNode().id());
+
+                    return true;
+                }
+            }, EVT_DISCOVERY_CUSTOM_EVT);
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2bfb8932/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 7f48f69..8474573 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
@@ -132,7 +132,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) {
+            Map<Long, Collection<ClusterNode>> topHist, Serializable data) {
             if (type == EVT_NODE_METRICS_UPDATED)
                 isMetricsUpdate = true;
         }
@@ -204,7 +204,8 @@ 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) {
+                    Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>>
topHist,
+                    Serializable data) {
                     // If METRICS_UPDATED came from local node
                     if (type == EVT_NODE_METRICS_UPDATED
                         && node.id().equals(spi.getLocalNode().id()))
@@ -368,7 +369,7 @@ 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) {
+                        Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>>
topHist, Serializable data) {
                         info("Discovery event [type=" + type + ", node=" + node + ']');
 
                         synchronized (mux) {


Mime
View raw message