process messages on user thread and protocol tasks off user thread

Makefile

@@ -68,7 +68,7 @@ monerod-local1:
 		--no-zmq \
 		--add-exclusive-node 127.0.0.1:28080 \
 		--rpc-access-control-origins http://localhost:8080 \
-		--fixed-difficulty 400
+		--fixed-difficulty 800
 
 monerod-local2:
 	./.localnet/monerod \
@@ -82,7 +82,7 @@ monerod-local2:
 		--confirm-external-bind \
 		--add-exclusive-node 127.0.0.1:48080 \
 		--rpc-access-control-origins http://localhost:8080 \
-		--fixed-difficulty 400
+		--fixed-difficulty 800
 
 funding-wallet-local:
 	./.localnet/monero-wallet-rpc \

core/src/main/java/bisq/core/offer/OpenOfferManager.java

@@ -621,10 +621,11 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
 
     private void processUnpostedOffers(TransactionResultHandler resultHandler, // TODO (woodser): transaction not needed with result handler
                                        ErrorMessageHandler errorMessageHandler) {
+        new Thread(() -> {
             List<String> errorMessages = new ArrayList<String>();
             for (OpenOffer scheduledOffer : openOffers.getObservableList()) {
                 if (scheduledOffer.getState() != OpenOffer.State.SCHEDULED) continue;
-            CountDownLatch latch = new CountDownLatch(openOffers.list.size());
+                CountDownLatch latch = new CountDownLatch(1);
                 processUnpostedOffer(scheduledOffer, (transaction) -> {
                     latch.countDown();
                 }, errorMessage -> {
@@ -636,9 +637,11 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
             requestPersistence();
             if (errorMessages.size() > 0) errorMessageHandler.handleErrorMessage(errorMessages.toString());
             else resultHandler.handleResult(null);
+        }).start();
     }
 
     private void processUnpostedOffer(OpenOffer openOffer, TransactionResultHandler resultHandler, ErrorMessageHandler errorMessageHandler) {
+        new Thread(() -> {
             try {
 
                 // done processing if wallet not initialized
@@ -704,6 +707,7 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
                 e.printStackTrace();
                 errorMessageHandler.handleErrorMessage(e.getMessage());
             }
+        }).start();
     }
 
     private BigInteger getScheduledAmount() {
@@ -790,6 +794,7 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
         log.info("Received SignOfferRequest from {} with offerId {} and uid {}",
                 peer, request.getOfferId(), request.getUid());
         
+        boolean result = false;
         String errorMessage = null;
         try {
             
@@ -845,9 +850,6 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
             signedOffers.add(signedOffer);
             requestPersistence();
 
-            // send ack
-            sendAckMessage(request.getClass(), peer, request.getPubKeyRing(), request.getOfferId(), request.getUid(), true, errorMessage);
-
             // send response with signature
             SignOfferResponse response = new SignOfferResponse(request.getOfferId(),
                     UUID.randomUUID().toString(),
@@ -874,11 +876,13 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
                                     errorMessage);
                         }
                     });
+            result = true;
         } catch (Exception e) {
             e.printStackTrace();
             errorMessage = "Exception at handleSignOfferRequest " + e.getMessage();
             log.error(errorMessage);
-            sendAckMessage(request.getClass(), peer, request.getPubKeyRing(), request.getOfferId(), request.getUid(), false, errorMessage);
+        } finally {
+            sendAckMessage(request.getClass(), peer, request.getPubKeyRing(), request.getOfferId(), request.getUid(), result, errorMessage);
         }
     }
     

core/src/main/java/bisq/core/offer/placeoffer/tasks/MakerSendSignOfferRequest.java

@@ -93,6 +93,10 @@ public class MakerSendSignOfferRequest extends Task<PlaceOfferModel> {
 
     private void sendSignOfferRequests(SignOfferRequest request, ResultHandler resultHandler, ErrorMessageHandler errorMessageHandler) {
         Arbitrator leastUsedArbitrator = DisputeAgentSelection.getLeastUsedArbitrator(model.getTradeStatisticsManager(), model.getArbitratorManager());
+        if (leastUsedArbitrator == null) {
+            errorMessageHandler.handleErrorMessage("Could not get least used arbitrator");
+            return;
+        }
         sendSignOfferRequests(request, leastUsedArbitrator.getNodeAddress(), new HashSet<NodeAddress>(), resultHandler, errorMessageHandler);
     }
 
@@ -140,7 +144,7 @@ public class MakerSendSignOfferRequest extends Task<PlaceOfferModel> {
 
         // get registered arbitrator
         Arbitrator arbitrator = model.getUser().getAcceptedArbitratorByAddress(arbitratorNodeAddress);
-        if (arbitrator == null) throw new RuntimeException("Node address " + arbitratorNodeAddress + " is not a registered arbitrator");
+        if (arbitrator == null) throw new RuntimeException("Node address " + arbitratorNodeAddress + " is not a registered arbitrator"); // TODO: use error handler
         request.getOfferPayload().setArbitratorSigner(arbitratorNodeAddress);
 
         // send request to arbitrator

core/src/main/java/bisq/core/trade/protocol/ArbitratorProtocol.java

@@ -51,6 +51,7 @@ public class ArbitratorProtocol extends DisputeProtocol {
 
   public void handleInitTradeRequest(InitTradeRequest message, NodeAddress peer, ErrorMessageHandler errorMessageHandler) {
       System.out.println("ArbitratorProtocol.handleInitTradeRequest()");
+      new Thread(() -> {
           synchronized (trade) {
               latchTrade();
               this.errorMessageHandler = errorMessageHandler;
@@ -75,6 +76,7 @@ public class ArbitratorProtocol extends DisputeProtocol {
                       .executeTasks(true);
               awaitTradeLatch();
           }
+      }).start();
   }
   
   @Override
@@ -84,6 +86,7 @@ public class ArbitratorProtocol extends DisputeProtocol {
   
   public void handleDepositRequest(DepositRequest request, NodeAddress sender) {
     System.out.println("ArbitratorProtocol.handleDepositRequest() " + trade.getId());
+    new Thread(() -> {
         synchronized (trade) {
             latchTrade();
             Validator.checkTradeId(processModel.getOfferId(), request);
@@ -108,6 +111,7 @@ public class ArbitratorProtocol extends DisputeProtocol {
                 .executeTasks(true);
             awaitTradeLatch();
         }
+    }).start();
   }
   
   @Override
@@ -117,6 +121,7 @@ public class ArbitratorProtocol extends DisputeProtocol {
   
   public void handlePaymentAccountKeyRequest(PaymentAccountKeyRequest request, NodeAddress sender) {
       System.out.println("ArbitratorProtocol.handlePaymentAccountKeyRequest() " + trade.getId());
+      new Thread(() -> {
           synchronized (trade) {
               latchTrade();
               Validator.checkTradeId(processModel.getOfferId(), request);
@@ -138,6 +143,7 @@ public class ArbitratorProtocol extends DisputeProtocol {
                   .executeTasks(true);
               awaitTradeLatch();
           }
+      }).start();
   }
   
   protected void handle(PayoutTxPublishedMessage request, NodeAddress peer) {

core/src/main/java/bisq/core/trade/protocol/BuyerAsMakerProtocol.java

@@ -54,6 +54,7 @@ public class BuyerAsMakerProtocol extends BuyerProtocol implements MakerProtocol
                                        NodeAddress peer,
                                        ErrorMessageHandler errorMessageHandler) {
         System.out.println(getClass().getCanonicalName() + ".handleInitTradeRequest()");
+        new Thread(() -> {
             synchronized (trade) {
                 latchTrade();
                 this.errorMessageHandler = errorMessageHandler;
@@ -77,6 +78,7 @@ public class BuyerAsMakerProtocol extends BuyerProtocol implements MakerProtocol
                         .executeTasks(true);
                 awaitTradeLatch();
             }
+        }).start();
     }
 
     @Override

core/src/main/java/bisq/core/trade/protocol/BuyerAsTakerProtocol.java

@@ -65,6 +65,7 @@ public class BuyerAsTakerProtocol extends BuyerProtocol implements TakerProtocol
     public void onTakeOffer(TradeResultHandler tradeResultHandler,
                             ErrorMessageHandler errorMessageHandler) {
       System.out.println(getClass().getCanonicalName() + ".onTakeOffer()");
+      new Thread(() -> {
           synchronized (trade) {
               latchTrade();
               this.tradeResultHandler = tradeResultHandler;
@@ -88,6 +89,7 @@ public class BuyerAsTakerProtocol extends BuyerProtocol implements TakerProtocol
                       .executeTasks(true);
               awaitTradeLatch();
           }
+      }).start();
     }
 
     @Override

core/src/main/java/bisq/core/trade/protocol/SellerAsMakerProtocol.java

@@ -55,6 +55,7 @@ public class SellerAsMakerProtocol extends SellerProtocol implements MakerProtoc
                                        NodeAddress peer,
                                        ErrorMessageHandler errorMessageHandler) {
         System.out.println(getClass().getCanonicalName() + ".handleInitTradeRequest()");
+        new Thread(() -> {
             synchronized (trade) {
                 latchTrade();
                 this.errorMessageHandler = errorMessageHandler;
@@ -78,6 +79,7 @@ public class SellerAsMakerProtocol extends SellerProtocol implements MakerProtoc
                         .executeTasks(true);
                 awaitTradeLatch();
             }
+        }).start();
     }
 
     @Override

core/src/main/java/bisq/core/trade/protocol/SellerAsTakerProtocol.java

@@ -63,6 +63,7 @@ public class SellerAsTakerProtocol extends SellerProtocol implements TakerProtoc
     public void onTakeOffer(TradeResultHandler tradeResultHandler,
                             ErrorMessageHandler errorMessageHandler) {
       System.out.println(getClass().getCanonicalName() + ".onTakeOffer()");
+      new Thread(() -> {
           synchronized (trade) {
               latchTrade();
               this.tradeResultHandler = tradeResultHandler;
@@ -86,6 +87,7 @@ public class SellerAsTakerProtocol extends SellerProtocol implements TakerProtoc
                       .executeTasks(true);
               awaitTradeLatch();
           }
+      }).start();
     }
 
     @Override

core/src/main/java/bisq/core/trade/protocol/TradeProtocol.java

@@ -229,6 +229,7 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener, D
 
     public void handleInitMultisigRequest(InitMultisigRequest request, NodeAddress sender) {
         System.out.println(getClass().getCanonicalName() + ".handleInitMultisigRequest()");
+        new Thread(() -> {
             synchronized (trade) {
                 latchTrade();
                 Validator.checkTradeId(processModel.getOfferId(), request);
@@ -251,10 +252,12 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener, D
                         .executeTasks(true);
                 awaitTradeLatch();
             }
+        }).start();
     }
 
     public void handleSignContractRequest(SignContractRequest message, NodeAddress sender) {
         System.out.println(getClass().getCanonicalName() + ".handleSignContractRequest() " + trade.getId());
+        new Thread(() -> {
             synchronized (trade) {
                 Validator.checkTradeId(processModel.getOfferId(), message);
                 if (trade.getState() == Trade.State.MULTISIG_COMPLETED || trade.getState() == Trade.State.CONTRACT_SIGNATURE_REQUESTED) {
@@ -285,10 +288,12 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener, D
                     });
                 }
             }
+        }).start();
     }
 
     public void handleSignContractResponse(SignContractResponse message, NodeAddress sender) {
         System.out.println(getClass().getCanonicalName() + ".handleSignContractResponse() " + trade.getId());
+        new Thread(() -> {
             synchronized (trade) {
                 Validator.checkTradeId(processModel.getOfferId(), message);
                 if (trade.getState() == Trade.State.CONTRACT_SIGNED) {
@@ -320,10 +325,12 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener, D
                     });
                 }
             }
+        }).start();
     }
 
     public void handleDepositResponse(DepositResponse response, NodeAddress sender) {
         System.out.println(getClass().getCanonicalName() + ".handleDepositResponse()");
+        new Thread(() -> {
             synchronized (trade) {
                 latchTrade();
                 Validator.checkTradeId(processModel.getOfferId(), response);
@@ -348,6 +355,7 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener, D
                         .executeTasks(true);
                 awaitTradeLatch();
             }
+        }).start();
     }
 
     // TODO (woodser): update to use fluent for consistency

desktop/src/main/java/bisq/desktop/main/market/offerbook/OfferBookChartViewModel.java

@@ -42,6 +42,7 @@ import bisq.core.offer.Offer;
 import bisq.core.offer.OfferDirection;
 import bisq.core.offer.OpenOfferManager;
 import bisq.core.provider.price.PriceFeedService;
+import bisq.core.trade.TradeUtils;
 import bisq.core.user.Preferences;
 import bisq.core.util.VolumeUtil;
 
@@ -66,6 +67,7 @@ import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
 import java.util.stream.Collectors;
 
 class OfferBookChartViewModel extends ActivatableViewModel {
@@ -125,7 +127,6 @@ class OfferBookChartViewModel extends ActivatableViewModel {
 
         offerBookListItems = offerBook.getOfferBookListItems();
         offerBookListItemsListener = c -> {
-            UserThread.execute(() -> {
             c.next();
             if (c.wasAdded() || c.wasRemoved()) {
                 ArrayList<OfferBookListItem> list = new ArrayList<>(c.getRemoved());
@@ -137,7 +138,6 @@ class OfferBookChartViewModel extends ActivatableViewModel {
             }
 
             fillTradeCurrencies();
-            });
         };
 
         currenciesUpdatedListener = (observable, oldValue, newValue) -> {

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

@@ -432,19 +432,13 @@ public class Connection implements HasCapabilities, Runnable, MessageListener {
     // Only receive non - CloseConnectionMessage network_messages
     @Override
     public void onMessage(NetworkEnvelope networkEnvelope, Connection connection) {
-        Connection that = this;
+        checkArgument(connection.equals(this));
-        connectionThreadPool.submit(new Runnable() {
-            @Override
-            public void run() {
-                checkArgument(connection.equals(that));
         if (networkEnvelope instanceof BundleOfEnvelopes) {
             onBundleOfEnvelopes((BundleOfEnvelopes) networkEnvelope, connection);
         } else {
-                    messageListeners.forEach(e -> e.onMessage(networkEnvelope, connection));
+            UserThread.execute(() -> messageListeners.forEach(e -> e.onMessage(networkEnvelope, connection)));
         }
     }
-        });
-    }
 
     private void onBundleOfEnvelopes(BundleOfEnvelopes bundleOfEnvelopes, Connection connection) {
         Map<P2PDataStorage.ByteArray, Set<NetworkEnvelope>> itemsByHash = new HashMap<>();
@@ -475,8 +469,8 @@ public class Connection implements HasCapabilities, Runnable, MessageListener {
                 envelopesToProcess.add(networkEnvelope);
             }
         }
-        envelopesToProcess.forEach(envelope ->
+        envelopesToProcess.forEach(envelope -> UserThread.execute(() ->
-                messageListeners.forEach(listener -> listener.onMessage(envelope, connection)));
+                messageListeners.forEach(listener -> listener.onMessage(envelope, connection))));
     }