Merge pull request #749 from tronprotocol/sync_msg

sync: mdf fetch item logic

Author: lvs007

src/main/java/org/tron/core/net/node/NodeImpl.java

@@ -335,7 +335,7 @@ public void setPool(SyncPool pool) {
   public void broadcast(Message msg) {
     InventoryType type;
     if (msg instanceof BlockMessage) {
-      logger.info("Ready to broadcast a block, Its hash is " + msg.getMessageId());
+      logger.info("Ready to broadcast block {}", ((BlockMessage) msg).getBlockId());
       freshBlockId.offer(((BlockMessage) msg).getBlockId());
       BlockCache.put(msg.getMessageId(), (BlockMessage) msg);
       type = InventoryType.BLOCK;
@@ -473,25 +473,22 @@ private void consumerAdvObjToFetch() {
       }
     }
     InvToSend sendPackage = new InvToSend();
-    //AtomicLong batchFecthResponseSize = new AtomicLong(0);
-
-    advObjToFetch.values().stream()
-        .sorted(PriorItem::compareTo)
-        .forEach(idToFetch ->
-      filterActivePeer.stream()
-          .filter(peer -> peer.getAdvObjSpreadToUs().containsKey(idToFetch.getHash())
-              && sendPackage.getSize(peer) < MAX_TRX_PER_PEER)
-          .sorted(Comparator.comparingInt(peer -> sendPackage.getSize(peer)))
-          .findFirst().ifPresent(peer -> {
-            long now = Time.getCurrentMillis();
-        if (idToFetch.getTime() > now - MSG_CACHE_DURATION_IN_BLOCKS * BLOCK_PRODUCED_INTERVAL) {
+    long now = Time.getCurrentMillis();
+    advObjToFetch.values().stream().sorted(PriorItem::compareTo).forEach(idToFetch -> {
+      if (idToFetch.getTime() < now - MSG_CACHE_DURATION_IN_BLOCKS * BLOCK_PRODUCED_INTERVAL) {
+        logger.info("This obj is too late to fetch: " + idToFetch);
+        advObjToFetch.remove(idToFetch.getHash());
+      }else {
+        filterActivePeer.stream()
+                .filter(peer -> peer.getAdvObjSpreadToUs().containsKey(idToFetch.getHash()) && sendPackage.getSize(peer) < MAX_TRX_PER_PEER)
+                .sorted(Comparator.comparingInt(peer -> sendPackage.getSize(peer)))
+                .findFirst().ifPresent(peer -> {
           sendPackage.add(idToFetch, peer);
           peer.getAdvObjWeRequested().put(idToFetch.getItem(), now);
-        } else {
-          logger.info("This obj is too late to fetch: " + idToFetch);
-        }
-        advObjToFetch.remove(idToFetch.getHash());
-      }));
+          advObjToFetch.remove(idToFetch.getHash());
+        });
+      }
+    });
 
     sendPackage.sendFetch();
   }
@@ -548,22 +545,17 @@ private synchronized void handleSyncBlock() {
       pool.forEach(msg -> {
         final boolean[] isFound = {false};
         getActivePeer().stream()
-            .filter(peer ->
-                !peer.getSyncBlockToFetch().isEmpty()
-                    && peer.getSyncBlockToFetch().peek().equals(msg.getBlockId()))
-            .forEach(peer -> {
-              peer.getSyncBlockToFetch().pop();
-              peer.getBlockInProc().add(msg.getBlockId());
-              isFound[0] = true;
-            });
+                .filter(peer -> !peer.getSyncBlockToFetch().isEmpty() && peer.getSyncBlockToFetch().peek().equals(msg.getBlockId()))
+                .forEach(peer -> {
+                  peer.getSyncBlockToFetch().pop();
+                  peer.getBlockInProc().add(msg.getBlockId());
+                  isFound[0] = true;
+                });
 
         if (isFound[0]) {
           if (!freshBlockId.contains(msg.getBlockId())) {
             blockWaitToProc.remove(msg);
-            //TODO: blockWaitToProc and handle thread.
-            BlockCapsule block = msg.getBlockCapsule();
-            //handleBackLogBlocksPool.execute(() -> processSyncBlock(block));
-            processSyncBlock(block);
+            processSyncBlock(msg.getBlockCapsule());
             isBlockProc[0] = true;
           }
         }