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); diff --git a/p2p/src/main/java/bisq/network/p2p/peers/getdata/RequestDataManager.java b/p2p/src/main/java/bisq/network/p2p/peers/getdata/RequestDataManager.java index 38d2bd3ca4b..5fcc153edab 100644 --- a/p2p/src/main/java/bisq/network/p2p/peers/getdata/RequestDataManager.java +++ b/p2p/src/main/java/bisq/network/p2p/peers/getdata/RequestDataManager.java @@ -31,6 +31,7 @@ import bisq.common.Timer; import bisq.common.UserThread; +import bisq.common.app.Version; import bisq.common.proto.network.NetworkEnvelope; import javax.inject.Inject; @@ -56,9 +57,9 @@ public class RequestDataManager implements MessageListener, ConnectionListener, private static final long RETRY_DELAY_SEC = 10; private static final long CLEANUP_TIMER = 120; // How many seeds we request the PreliminaryGetDataRequest from - private static int NUM_SEEDS_FOR_PRELIMINARY_REQUEST = 2; + private static int NUM_SEEDS_FOR_PRELIMINARY_REQUEST = 16; // how many seeds additional to the first responding PreliminaryGetDataRequest seed we request the GetUpdatedDataRequest from - private static int NUM_ADDITIONAL_SEEDS_FOR_UPDATE_REQUEST = 1; + private static int NUM_ADDITIONAL_SEEDS_FOR_UPDATE_REQUEST = 16; private boolean isPreliminaryDataRequest = true; /////////////////////////////////////////////////////////////////////////////////////////// @@ -261,6 +262,11 @@ public void onMessage(NetworkEnvelope networkEnvelope, Connection connection) { if (peerManager.isSeedNode(connection)) connection.setPeerType(Connection.PeerType.SEED_NODE); + GetDataRequest getDataRequest = (GetDataRequest) networkEnvelope; + if (getDataRequest.getVersion() == null || !Version.isNewVersion(getDataRequest.getVersion(), "1.5.0")) { + connection.shutDown(CloseConnectionReason.MANDATORY_CAPABILITIES_NOT_SUPPORTED); + return; + } final String uid = connection.getUid(); if (!getDataRequestHandlers.containsKey(uid)) { GetDataRequestHandler getDataRequestHandler = new GetDataRequestHandler(networkNode, dataStorage, @@ -284,7 +290,7 @@ public void onFault(String errorMessage, @Nullable Connection connection) { } }); getDataRequestHandlers.put(uid, getDataRequestHandler); - getDataRequestHandler.handle((GetDataRequest) networkEnvelope, connection); + getDataRequestHandler.handle(getDataRequest, connection); } else { log.warn("We have already a GetDataRequestHandler for that connection started. " + "We start a cleanup timer if the handler has not closed by itself in between 2 minutes.");