diff --git a/p2p/src/main/java/bisq/network/p2p/network/Connection.java b/p2p/src/main/java/bisq/network/p2p/network/Connection.java index 3a03fd0231f..c3f6f222215 100644 --- a/p2p/src/main/java/bisq/network/p2p/network/Connection.java +++ b/p2p/src/main/java/bisq/network/p2p/network/Connection.java @@ -68,9 +68,13 @@ import java.io.StreamCorruptedException; import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.Queue; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; @@ -421,11 +425,36 @@ private boolean violatesThrottleLimit(long now, int seconds, int messageCountLim @Override public void onMessage(NetworkEnvelope networkEnvelope, Connection connection) { checkArgument(connection.equals(this)); - + int accountAgeWitnessEntries = 0; if (networkEnvelope instanceof BundleOfEnvelopes) { - for (NetworkEnvelope current : ((BundleOfEnvelopes) networkEnvelope).getEnvelopes()) { - UserThread.execute(() -> messageListeners.forEach(e -> e.onMessage(current, connection))); + Map> map = new HashMap<>(); + Set set = new HashSet<>(); + + List networkEnvelopes = ((BundleOfEnvelopes) networkEnvelope).getEnvelopes(); + for (NetworkEnvelope current : networkEnvelopes) { + String simpleName = current.getClass().getSimpleName(); + boolean isAccountAgeWitness = false; + if (current instanceof AddPersistableNetworkPayloadMessage) { + PersistableNetworkPayload persistableNetworkPayload = ((AddPersistableNetworkPayloadMessage) current).getPersistableNetworkPayload(); + simpleName = "AddPersistableNetworkPayloadMessage." + persistableNetworkPayload.getClass().getSimpleName(); + if (simpleName.equals("AddPersistableNetworkPayloadMessage.AccountAgeWitness")) { + accountAgeWitnessEntries++; + isAccountAgeWitness = true; + } + } + map.putIfAbsent(simpleName, new ArrayList<>()); + map.get(simpleName).add(current); + if (!isAccountAgeWitness || accountAgeWitnessEntries < 20) { + set.add(current); + } } + map.forEach((key, value) -> log.info("BundleOfEnvelope with {} items of {}, from {}", + value.size(), key, connection.getPeersNodeAddressOptional())); + + log.info("We forward {} items. All received items: {}", set.size(), networkEnvelopes.size()); + + set.forEach(envelope -> UserThread.execute(() -> + messageListeners.forEach(listener -> listener.onMessage(envelope, connection)))); } else { UserThread.execute(() -> messageListeners.forEach(e -> e.onMessage(networkEnvelope, connection))); } @@ -718,7 +747,6 @@ public void run() { lastReadTimeStamp, now, elapsed); Thread.sleep(20); } - // Reading the protobuffer message from the inputStream protobuf.NetworkEnvelope proto = protobuf.NetworkEnvelope.parseDelimitedFrom(protoInputStream);