Merge pull request #746 from tronprotocol/fix-net-bug

after add to active success then send hello msg

Author: wubin12

src/main/java/org/tron/common/overlay/server/Channel.java

@@ -17,7 +17,6 @@
  */
 package org.tron.common.overlay.server;
 
-import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
@@ -138,7 +137,6 @@ public void publicHandshakeFinished(ChannelHandlerContext ctx, HelloMessage msg)
         setStartTime(msg.getTimestamp());
         setTronState(TronState.HANDSHAKE_FINISHED);
         getNodeStatistics().p2pHandShake.add();
-        channelManager.add(this);
         logger.info("Finish handshake with {}.", ctx.channel().remoteAddress());
     }
 

src/main/java/org/tron/common/overlay/server/ChannelManager.java

@@ -85,7 +85,8 @@ public Set<String> nodesInUse() {
 
   public void disconnect(Channel peer, ReasonCode reason) {
     peer.disconnect(reason);
-    InetSocketAddress socketAddress = (InetSocketAddress)peer.getChannelHandlerContext().channel().remoteAddress();
+    InetSocketAddress socketAddress = (InetSocketAddress) peer.getChannelHandlerContext().channel()
+        .remoteAddress();
     recentlyDisconnected.put(socketAddress.getAddress(), new Date());
   }
 
@@ -115,34 +116,30 @@ public boolean isRecentlyDisconnected(InetAddress peerAddr) {
     }
   }
 
-  public void add(Channel peer) {
-    if (isShouldAddToActivePeers(peer)) {
-      activePeers.put(peer.getNodeIdWrapper(), peer);
-      syncPool.onConnect(peer);
-      logger.info("Add active peer {}, total active peers: {}", peer, activePeers.size());
-    }
-  }
-
-  private boolean isShouldAddToActivePeers(Channel peer) {
+  public synchronized boolean procPeer(Channel peer) {
     if (peer.getNodeStatistics().isPenalized()) {
       disconnect(peer, peer.getNodeStatistics().getDisconnectReason());
       return false;
-    } else if (!peer.isActive() && activePeers.size() >= maxActivePeers) {
+    }
+
+    if (!peer.isActive() && activePeers.size() >= maxActivePeers) {
       disconnect(peer, TOO_MANY_PEERS);
       return false;
-    } else if (activePeers.containsKey(peer.getNodeIdWrapper())) {
+    }
+
+    if (activePeers.containsKey(peer.getNodeIdWrapper())) {
       Channel channel = activePeers.get(peer.getNodeIdWrapper());
       if (channel.getStartTime() > peer.getStartTime()) {
         logger.info("Disconnect connection established later, {}", channel.getNode());
         disconnect(channel, DUPLICATE_PEER);
-        return true;
       } else {
         disconnect(peer, DUPLICATE_PEER);
         return false;
       }
-    } else {
-      return true;
     }
+    logger.info("Add active peer {}, total active peers: {}", peer, activePeers.size());
+    activePeers.put(peer.getNodeIdWrapper(), peer);
+    return true;
   }
 
   public Collection<Channel> getActivePeers() {

src/main/java/org/tron/common/overlay/server/HandshakeHandler.java

@@ -21,7 +21,6 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import java.net.InetSocketAddress;
-import java.net.SocketAddress;
 import java.util.Arrays;
 import java.util.List;
 import org.slf4j.Logger;
@@ -31,14 +30,15 @@
 import org.springframework.context.annotation.Scope;
 import org.springframework.stereotype.Component;
 import org.tron.common.overlay.discover.NodeManager;
-import org.tron.common.overlay.message.*;
-import org.tron.common.utils.ByteArray;
+import org.tron.common.overlay.message.DisconnectMessage;
+import org.tron.common.overlay.message.HelloMessage;
+import org.tron.common.overlay.message.P2pMessage;
+import org.tron.common.overlay.message.P2pMessageFactory;
+import org.tron.common.overlay.message.ReasonCode;
 import org.tron.core.config.args.Args;
 import org.tron.core.db.Manager;
 import org.tron.core.net.peer.PeerConnection;
 
-import static org.tron.common.overlay.message.StaticMessages.PONG_MESSAGE;
-
 @Component
 @Scope("prototype")
 public class HandshakeHandler extends ByteToMessageDecoder {
@@ -51,13 +51,20 @@ public class HandshakeHandler extends ByteToMessageDecoder {
 
   private final NodeManager nodeManager;
 
+  private final ChannelManager channelManager;
+
   private Manager manager;
 
   private  P2pMessageFactory messageFactory = new P2pMessageFactory();
+  
+  @Autowired
+  private SyncPool syncPool;
 
   @Autowired
-  public HandshakeHandler(final NodeManager nodeManager, final Manager manager) {
+  public HandshakeHandler(final NodeManager nodeManager, final ChannelManager channelManager,
+      final Manager manager) {
     this.nodeManager = nodeManager;
+    this.channelManager = channelManager;
     this.manager = manager;
   }
 
@@ -140,14 +147,19 @@ private void handleHelloMsg(ChannelHandlerContext ctx, HelloMessage msg) {
       return;
     }
 
-    if (remoteId.length != 64) {
-      sendHelloMsg(ctx, msg.getTimestamp());
-    }
-
     ((PeerConnection)channel).setHelloMessage(msg);
 
     channel.getNodeStatistics().p2pInHello.add();
 
     channel.publicHandshakeFinished(ctx, msg);
+    if (!channelManager.procPeer(channel)) {
+      return;
+    }
+
+    if (remoteId.length != 64) {
+      sendHelloMsg(ctx, msg.getTimestamp());
+    }
+
+    syncPool.onConnect(channel);
   }
 }