Bundle messages at broadcast

core/src/main/java/bisq/core/dao/monitoring/network/StateNetworkService.java

@@ -156,7 +156,7 @@ public void requestHashesFromAllConnectedSeedNodes(int fromHeight) {
 
     public void broadcastMyStateHash(StH myStateHash) {
         NewStateHashMessage newStateHashMessage = getNewStateHashMessage(myStateHash);
-        broadcaster.broadcast(newStateHashMessage, networkNode.getNodeAddress(), null);
+        broadcaster.broadcast(newStateHashMessage, networkNode.getNodeAddress());
     }
 
     public void requestHashes(int fromHeight, String peersAddress) {

core/src/main/java/bisq/core/dao/node/full/network/FullNodeNetworkService.java

@@ -105,7 +105,7 @@ public void publishNewBlock(Block block) {
         log.info("Publish new block at height={} and block hash={}", block.getHeight(), block.getHash());
         RawBlock rawBlock = RawBlock.fromBlock(block);
         NewBlockBroadcastMessage newBlockBroadcastMessage = new NewBlockBroadcastMessage(rawBlock);
-        broadcaster.broadcast(newBlockBroadcastMessage, networkNode.getNodeAddress(), null);
+        broadcaster.broadcast(newBlockBroadcastMessage, networkNode.getNodeAddress());
     }
 
 

core/src/main/java/bisq/core/dao/node/lite/network/LiteNodeNetworkService.java

@@ -238,7 +238,7 @@ public void onMessage(NetworkEnvelope networkEnvelope, Connection connection) {
                 log.debug("We received a new message from peer {} and broadcast it to our peers. extBlockId={}",
                         connection.getPeersNodeAddressOptional().orElse(null), extBlockId);
                 receivedBlocks.add(extBlockId);
-                broadcaster.broadcast(newBlockBroadcastMessage, connection.getPeersNodeAddressOptional().orElse(null), null);
+                broadcaster.broadcast(newBlockBroadcastMessage, connection.getPeersNodeAddressOptional().orElse(null));
                 listeners.forEach(listener -> listener.onNewBlockReceived(newBlockBroadcastMessage));
             } else {
                 log.debug("We had that message already and do not further broadcast it. extBlockId={}", extBlockId);

p2p/src/main/java/bisq/network/p2p/BundleOfEnvelopes.java

@@ -17,6 +17,7 @@
 
 package bisq.network.p2p;
 
+import bisq.network.p2p.storage.messages.BroadcastMessage;
 import bisq.network.p2p.storage.payload.CapabilityRequiringPayload;
 
 import bisq.common.app.Capabilities;
@@ -36,14 +37,18 @@
 
 @EqualsAndHashCode(callSuper = true)
 @Value
-public final class BundleOfEnvelopes extends NetworkEnvelope implements ExtendedDataSizePermission, CapabilityRequiringPayload {
+public final class BundleOfEnvelopes extends BroadcastMessage implements ExtendedDataSizePermission, CapabilityRequiringPayload {
 
     private final List<NetworkEnvelope> envelopes;
 
     public BundleOfEnvelopes() {
         this(new ArrayList<>(), Version.getP2PMessageVersion());
     }
 
+    public BundleOfEnvelopes(List<NetworkEnvelope> envelopes) {
+        this(envelopes, Version.getP2PMessageVersion());
+    }
+
     public void add(NetworkEnvelope networkEnvelope) {
         envelopes.add(networkEnvelope);
     }
@@ -67,7 +72,9 @@ public protobuf.NetworkEnvelope toProtoNetworkEnvelope() {
                 .build();
     }
 
-    public static BundleOfEnvelopes fromProto(protobuf.BundleOfEnvelopes proto, NetworkProtoResolver resolver, int messageVersion) {
+    public static BundleOfEnvelopes fromProto(protobuf.BundleOfEnvelopes proto,
+                                              NetworkProtoResolver resolver,
+                                              int messageVersion) {
         List<NetworkEnvelope> envelopes = proto.getEnvelopesList()
                 .stream()
                 .map(envelope -> {

p2p/src/main/java/bisq/network/p2p/P2PService.java

@@ -37,7 +37,6 @@
 import bisq.network.p2p.storage.HashMapChangedListener;
 import bisq.network.p2p.storage.P2PDataStorage;
 import bisq.network.p2p.storage.messages.AddDataMessage;
-import bisq.network.p2p.storage.messages.BroadcastMessage;
 import bisq.network.p2p.storage.messages.RefreshOfferMessage;
 import bisq.network.p2p.storage.payload.CapabilityRequiringPayload;
 import bisq.network.p2p.storage.payload.MailboxStoragePayload;
@@ -53,7 +52,6 @@
 import bisq.common.proto.ProtobufferException;
 import bisq.common.proto.network.NetworkEnvelope;
 import bisq.common.proto.persistable.PersistedDataHost;
-import bisq.common.util.Utilities;
 
 import com.google.inject.Inject;
 
@@ -77,6 +75,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -122,9 +121,6 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
     private final BooleanProperty preliminaryDataReceived = new SimpleBooleanProperty();
     private final IntegerProperty numConnectedPeers = new SimpleIntegerProperty(0);
 
-    private volatile boolean shutDownInProgress;
-    @Getter
-    private boolean shutDownComplete;
     private final Subscription networkReadySubscription;
     private boolean isBootstrapped;
     private final KeepAliveManager keepAliveManager;
@@ -212,48 +208,48 @@ public void onAllServicesInitialized() {
     }
 
     public void shutDown(Runnable shutDownCompleteHandler) {
-        if (!shutDownInProgress) {
-            shutDownInProgress = true;
-
-            shutDownResultHandlers.add(shutDownCompleteHandler);
+        shutDownResultHandlers.add(shutDownCompleteHandler);
 
-            if (p2PDataStorage != null)
-                p2PDataStorage.shutDown();
+        // We need to make sure queued up messages are flushed out before we continue shut down other network
+        // services
+        if (broadcaster != null) {
+            broadcaster.shutDown(this::doShutDown);
+        } else {
+            doShutDown();
+        }
+    }
 
-            if (peerManager != null)
-                peerManager.shutDown();
+    private void doShutDown() {
+        if (p2PDataStorage != null) {
+            p2PDataStorage.shutDown();
+        }
 
-            if (broadcaster != null)
-                broadcaster.shutDown();
+        if (peerManager != null) {
+            peerManager.shutDown();
+        }
 
-            if (requestDataManager != null)
-                requestDataManager.shutDown();
+        if (requestDataManager != null) {
+            requestDataManager.shutDown();
+        }
 
-            if (peerExchangeManager != null)
-                peerExchangeManager.shutDown();
+        if (peerExchangeManager != null) {
+            peerExchangeManager.shutDown();
+        }
 
-            if (keepAliveManager != null)
-                keepAliveManager.shutDown();
+        if (keepAliveManager != null) {
+            keepAliveManager.shutDown();
+        }
 
-            if (networkReadySubscription != null)
-                networkReadySubscription.unsubscribe();
+        if (networkReadySubscription != null) {
+            networkReadySubscription.unsubscribe();
+        }
 
-            if (networkNode != null) {
-                networkNode.shutDown(() -> {
-                    shutDownResultHandlers.stream().forEach(Runnable::run);
-                    shutDownComplete = true;
-                });
-            } else {
-                shutDownResultHandlers.stream().forEach(Runnable::run);
-                shutDownComplete = true;
-            }
+        if (networkNode != null) {
+            networkNode.shutDown(() -> {
+                shutDownResultHandlers.forEach(Runnable::run);
+            });
         } else {
-            log.debug("shutDown already in progress");
-            if (shutDownComplete) {
-                shutDownCompleteHandler.run();
-            } else {
-                shutDownResultHandlers.add(shutDownCompleteHandler);
-            }
+            shutDownResultHandlers.forEach(Runnable::run);
         }
     }
 
@@ -448,7 +444,7 @@ public void onAdded(Collection<ProtectedStorageEntry> protectedStorageEntries) {
 
     @Override
     public void onRemoved(Collection<ProtectedStorageEntry> protectedStorageEntries) {
-        // not handled
+        // not used
     }
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -672,43 +668,21 @@ private void addMailboxData(MailboxStoragePayload expirableMailboxStoragePayload
 
                     BroadcastHandler.Listener listener = new BroadcastHandler.Listener() {
                         @Override
-                        public void onBroadcasted(BroadcastMessage message, int numOfCompletedBroadcasts) {
+                        public void onSufficientlyBroadcast(List<Broadcaster.BroadcastRequest> broadcastRequests) {
+                            broadcastRequests.stream()
+                                    .filter(broadcastRequest -> broadcastRequest.getMessage() instanceof AddDataMessage)
+                                    .filter(broadcastRequest -> {
+                                        AddDataMessage addDataMessage = (AddDataMessage) broadcastRequest.getMessage();
+                                        return addDataMessage.getProtectedStorageEntry().equals(protectedMailboxStorageEntry);
+                                    })
+                                    .forEach(e -> sendMailboxMessageListener.onStoredInMailbox());
                         }
 
                         @Override
-                        public void onBroadcastedToFirstPeer(BroadcastMessage message) {
-                            // The reason for that check was to separate different callback for different send calls.
-                            // We only want to notify our sendMailboxMessageListener for the calls he is interested in.
-                            if (message instanceof AddDataMessage &&
-                                    ((AddDataMessage) message).getProtectedStorageEntry().equals(protectedMailboxStorageEntry)) {
-                                // We delay a bit to give more time for sufficient propagation in the P2P network.
-                                // This should help to avoid situations where a user closes the app too early and the msg
-                                // does not arrive.
-                                // We could use onBroadcastCompleted instead but it might take too long if one peer
-                                // is very badly connected.
-                                // TODO We could check for a certain threshold of no. of incoming network_messages of the same msg
-                                // to see how well it is propagated. BitcoinJ uses such an approach for tx propagation.
-                                UserThread.runAfter(() -> {
-                                    log.info("Broadcasted to first peer (3 sec. ago):  Message = {}", Utilities.toTruncatedString(message));
-                                    sendMailboxMessageListener.onStoredInMailbox();
-                                }, 3);
-                            }
-                        }
-
-                        @Override
-                        public void onBroadcastCompleted(BroadcastMessage message,
-                                                         int numOfCompletedBroadcasts,
-                                                         int numOfFailedBroadcasts) {
-                            log.info("Broadcast completed: Sent to {} peers (failed: {}). Message = {}",
-                                    numOfCompletedBroadcasts, numOfFailedBroadcasts, Utilities.toTruncatedString(message));
-                            if (numOfCompletedBroadcasts == 0)
-                                sendMailboxMessageListener.onFault("Broadcast completed without any successful broadcast");
-                        }
-
-                        @Override
-                        public void onBroadcastFailed(String errorMessage) {
-                            // TODO investigate why not sending sendMailboxMessageListener.onFault. Related probably
-                            // to the logic from BroadcastHandler.sendToPeer
+                        public void onNotSufficientlyBroadcast(int numOfCompletedBroadcasts, int numOfFailedBroadcast) {
+                            sendMailboxMessageListener.onFault("Message was not sufficiently broadcast.\n" +
+                                    "numOfCompletedBroadcasts: " + numOfCompletedBroadcasts + ".\n" +
+                                    "numOfFailedBroadcast=" + numOfFailedBroadcast);
                         }
                     };
                     boolean result = p2PDataStorage.addProtectedStorageEntry(protectedMailboxStorageEntry, networkNode.getNodeAddress(), listener);
@@ -721,7 +695,7 @@ public void onBroadcastFailed(String errorMessage) {
                         log.error("Unexpected state: adding mailbox message that already exists.");
                     }
                 } catch (CryptoException e) {
-                    log.error("Signing at getDataWithSignedSeqNr failed. That should never happen.");
+                    log.error("Signing at getMailboxDataWithSignedSeqNr failed.");
                 }
             } else {
                 sendMailboxMessageListener.onFault("There are no P2P network nodes connected. " +

p2p/src/main/java/bisq/network/p2p/network/Connection.java

@@ -225,7 +225,7 @@ public Capabilities getCapabilities() {
 
     // Called from various threads
     public void sendMessage(NetworkEnvelope networkEnvelope) {
-        log.debug(">> Send networkEnvelope of type: " + networkEnvelope.getClass().getSimpleName());
+        log.debug(">> Send networkEnvelope of type: {}", networkEnvelope.getClass().getSimpleName());
 
         if (!stopped) {
             if (noCapabilityRequiredOrCapabilityIsSupported(networkEnvelope)) {
@@ -319,6 +319,8 @@ public void sendMessage(NetworkEnvelope networkEnvelope) {
         }
     }
 
+    // TODO: If msg is BundleOfEnvelopes we should check each individual message for capability and filter out those
+    //  which fail.
     public boolean noCapabilityRequiredOrCapabilityIsSupported(Proto msg) {
         boolean result;
         if (msg instanceof AddDataMessage) {
@@ -408,12 +410,13 @@ private boolean violatesThrottleLimit(long now, int seconds, int messageCountLim
     public void onMessage(NetworkEnvelope networkEnvelope, Connection connection) {
         checkArgument(connection.equals(this));
 
-        if (networkEnvelope instanceof BundleOfEnvelopes)
+        if (networkEnvelope instanceof BundleOfEnvelopes) {
             for (NetworkEnvelope current : ((BundleOfEnvelopes) networkEnvelope).getEnvelopes()) {
                 UserThread.execute(() -> messageListeners.forEach(e -> e.onMessage(current, connection)));
             }
-        else
+        } else {
             UserThread.execute(() -> messageListeners.forEach(e -> e.onMessage(networkEnvelope, connection)));
+        }
     }