mark TCP neighbor as stopped when removed, so Replicator Sink & Sourc…

…e thread stop.
iotaledger · Dec 24, 2017 · 3049751 · 3049751
1 parent 5f06cf0
commit 3049751
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 2 deletions.
diff --git a/src/main/java/com/iota/iri/network/TCPNeighbor.java b/src/main/java/com/iota/iri/network/TCPNeighbor.java
@@ -20,6 +20,7 @@ public class TCPNeighbor extends Neighbor {
     private int tcpPort;
 
     private final ArrayBlockingQueue<ByteBuffer> sendQueue = new ArrayBlockingQueue<>(10);
+    private boolean stopped = false;
 
     public TCPNeighbor(InetSocketAddress address, boolean isConfigured, final double limit) {
         super(address, isConfigured, limit);
@@ -35,6 +36,11 @@ public Socket getSource() {
     public void clear() {
         setSource(null);
         setSink(null);
+        this.stopped = true;
+    }
+
+    public boolean isStopped() {
+        return stopped;
     }
 
     public void setSource(Socket source) {

diff --git a/src/main/java/com/iota/iri/network/replicator/ReplicatorSinkProcessor.java b/src/main/java/com/iota/iri/network/replicator/ReplicatorSinkProcessor.java
@@ -76,7 +76,7 @@ public void run() {
                             portAsByteArray, 0, ReplicatorSinkPool.PORT_BYTES);
                     out.write(portAsByteArray);
 
-                    while (!replicatorSinkPool.shutdown) {
+                    while (!replicatorSinkPool.shutdown && !neighbor.isStopped()) {
                         try {
                             ByteBuffer message = neighbor.getNextMessage();
                             if (neighbor.getSink() != null) { 

diff --git a/src/main/java/com/iota/iri/network/replicator/ReplicatorSourceProcessor.java b/src/main/java/com/iota/iri/network/replicator/ReplicatorSourceProcessor.java
@@ -134,7 +134,7 @@ public void run() {
             connection.setSoTimeout(0);  // infinite timeout - blocking read
 
             offset = 0;
-            while (!shutdown) {
+            while (!shutdown && !neighbor.isStopped()) {
 
                 while ( ((count = stream.read(data, offset, (TRANSACTION_PACKET_SIZE - offset + ReplicatorSinkProcessor.CRC32_BYTES))) != -1) 
                         && (offset < (TRANSACTION_PACKET_SIZE + ReplicatorSinkProcessor.CRC32_BYTES))) {