ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sevdoki...@apache.org
Subject [2/3] incubator-ignite git commit: # IGNITE-943 Move TcpDiscoverySpiAdapter.MessageWorkerAdapter to TcpDiscoverySpi
Date Tue, 26 May 2015 16:17:28 GMT
# IGNITE-943 Move TcpDiscoverySpiAdapter.MessageWorkerAdapter to TcpDiscoverySpi


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

Branch: refs/heads/ignite-943
Commit: f43cbbbb915da375fa89edf3120a0c8d52c26bfe
Parents: e5849fe
Author: sevdokimov <sevdokimov@gridgain.com>
Authored: Tue May 26 16:57:22 2015 +0300
Committer: sevdokimov <sevdokimov@gridgain.com>
Committed: Tue May 26 16:57:22 2015 +0300

----------------------------------------------------------------------
 .../spi/discovery/tcp/TcpDiscoverySpi.java      | 91 ++++++++++++++++++++
 .../discovery/tcp/TcpDiscoverySpiAdapter.java   | 90 -------------------
 2 files changed, 91 insertions(+), 90 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f43cbbbb/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 2123afc..37a07d6 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
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.events.*;
 import org.apache.ignite.internal.processors.security.*;
 import org.apache.ignite.internal.util.*;
 import org.apache.ignite.internal.util.future.*;
+import org.apache.ignite.internal.util.io.*;
 import org.apache.ignite.internal.util.lang.*;
 import org.apache.ignite.internal.util.tostring.*;
 import org.apache.ignite.internal.util.typedef.*;
@@ -5321,4 +5322,94 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
             U.closeQuiet(sock);
         }
     }
+
+    /**
+     * Base class for message workers.
+     */
+    protected abstract class MessageWorkerAdapter extends IgniteSpiThread {
+        /** Pre-allocated output stream (100K). */
+        private final GridByteArrayOutputStream bout = new GridByteArrayOutputStream(100
* 1024);
+
+        /** Message queue. */
+        private final BlockingDeque<TcpDiscoveryAbstractMessage> queue = new LinkedBlockingDeque<>();
+
+        /** Backed interrupted flag. */
+        private volatile boolean interrupted;
+
+        /**
+         * @param name Thread name.
+         */
+        protected MessageWorkerAdapter(String name) {
+            super(gridName, name, log);
+
+            setPriority(threadPri);
+        }
+
+        /** {@inheritDoc} */
+        @Override protected void body() throws InterruptedException {
+            if (log.isDebugEnabled())
+                log.debug("Message worker started [locNodeId=" + getLocalNodeId() + ']');
+
+            while (!isInterrupted()) {
+                TcpDiscoveryAbstractMessage msg = queue.poll(2000, TimeUnit.MILLISECONDS);
+
+                if (msg == null)
+                    continue;
+
+                processMessage(msg);
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public void interrupt() {
+            interrupted = true;
+
+            super.interrupt();
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean isInterrupted() {
+            return interrupted || super.isInterrupted();
+        }
+
+        /**
+         * @return Current queue size.
+         */
+        int queueSize() {
+            return queue.size();
+        }
+
+        /**
+         * Adds message to queue.
+         *
+         * @param msg Message to add.
+         */
+        void addMessage(TcpDiscoveryAbstractMessage msg) {
+            if (msg.highPriority())
+                queue.addFirst(msg);
+            else
+                queue.add(msg);
+
+            if (log.isDebugEnabled())
+                log.debug("Message has been added to queue: " + msg);
+        }
+
+        /**
+         * @param msg Message.
+         */
+        protected abstract void processMessage(TcpDiscoveryAbstractMessage msg);
+
+        /**
+         * @param sock Socket.
+         * @param msg Message.
+         * @throws IOException If IO failed.
+         * @throws IgniteCheckedException If marshalling failed.
+         */
+        protected final void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg)
+            throws IOException, IgniteCheckedException {
+            bout.reset();
+
+            TcpDiscoverySpi.this.writeToSocket(sock, msg, bout);
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f43cbbbb/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 d5ea9f2..4349ebc 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
@@ -1185,96 +1185,6 @@ public abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter
implements
     }
 
     /**
-     * Base class for message workers.
-     */
-    protected abstract class MessageWorkerAdapter extends IgniteSpiThread {
-        /** Pre-allocated output stream (100K). */
-        private final GridByteArrayOutputStream bout = new GridByteArrayOutputStream(100
* 1024);
-
-        /** Message queue. */
-        private final BlockingDeque<TcpDiscoveryAbstractMessage> queue = new LinkedBlockingDeque<>();
-
-        /** Backed interrupted flag. */
-        private volatile boolean interrupted;
-
-        /**
-         * @param name Thread name.
-         */
-        protected MessageWorkerAdapter(String name) {
-            super(gridName, name, log);
-
-            setPriority(threadPri);
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void body() throws InterruptedException {
-            if (log.isDebugEnabled())
-                log.debug("Message worker started [locNodeId=" + getLocalNodeId() + ']');
-
-            while (!isInterrupted()) {
-                TcpDiscoveryAbstractMessage msg = queue.poll(2000, TimeUnit.MILLISECONDS);
-
-                if (msg == null)
-                    continue;
-
-                processMessage(msg);
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override public void interrupt() {
-            interrupted = true;
-
-            super.interrupt();
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean isInterrupted() {
-            return interrupted || super.isInterrupted();
-        }
-
-        /**
-         * @return Current queue size.
-         */
-        int queueSize() {
-            return queue.size();
-        }
-
-        /**
-         * Adds message to queue.
-         *
-         * @param msg Message to add.
-         */
-        void addMessage(TcpDiscoveryAbstractMessage msg) {
-            if (msg.highPriority())
-                queue.addFirst(msg);
-            else
-                queue.add(msg);
-
-            if (log.isDebugEnabled())
-                log.debug("Message has been added to queue: " + msg);
-        }
-
-        /**
-         * @param msg Message.
-         */
-        protected abstract void processMessage(TcpDiscoveryAbstractMessage msg);
-
-        /**
-         * @param sock Socket.
-         * @param msg Message.
-         * @throws IOException If IO failed.
-         * @throws IgniteCheckedException If marshalling failed.
-         */
-        protected final void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg)
-            throws IOException, IgniteCheckedException {
-            bout.reset();
-
-            TcpDiscoverySpiAdapter.this.writeToSocket(sock, msg, bout);
-        }
-    }
-
-    /**
      * Allow to connect to addresses parallel.
      */
     protected class SocketMultiConnector implements AutoCloseable {


Mime
View raw message