Remove TradeProtocol from Trade

We keep the TradeProtocol now in TradeManager in a hashmap.
As TradeProtocol constructors get called now earlier we need to add an onInitialized method which signals that the TradeProtocol is ready.
The processModel needs to get set the transient fields after construction.

core/src/main/java/bisq/core/support/dispute/mediation/MediationManager.java

@@ -234,7 +234,7 @@ public final class MediationManager extends DisputeManager<MediationDisputeList>
         ProcessModel processModel = trade.getProcessModel();
         processModel.setBuyerPayoutAmountFromMediation(buyerPayoutAmount.value);
         processModel.setSellerPayoutAmountFromMediation(sellerPayoutAmount.value);
-        DisputeProtocol tradeProtocol = (DisputeProtocol) trade.getTradeProtocol();
+        DisputeProtocol tradeProtocol = (DisputeProtocol) tradeManager.getTradeProtocol(trade);
 
         trade.setMediationResultState(MediationResultState.MEDIATION_RESULT_ACCEPTED);
 

core/src/main/java/bisq/core/trade/BuyerAsMakerTrade.java

@@ -20,7 +20,6 @@ package bisq.core.trade;
 import bisq.core.btc.wallet.BtcWalletService;
 import bisq.core.offer.Offer;
 import bisq.core.proto.CoreProtoResolver;
-import bisq.core.trade.protocol.BuyerAsMakerProtocol;
 import bisq.core.trade.protocol.ProcessModel;
 
 import bisq.network.p2p.NodeAddress;
@@ -100,14 +99,4 @@ public final class BuyerAsMakerTrade extends BuyerTrade implements MakerTrade {
                 proto,
                 coreProtoResolver);
     }
-
-
-    ///////////////////////////////////////////////////////////////////////////////////////////
-    // API
-    ///////////////////////////////////////////////////////////////////////////////////////////
-
-    @Override
-    protected void createTradeProtocol() {
-        tradeProtocol = new BuyerAsMakerProtocol(this);
-    }
 }

core/src/main/java/bisq/core/trade/BuyerAsTakerTrade.java

@@ -20,7 +20,6 @@ package bisq.core.trade;
 import bisq.core.btc.wallet.BtcWalletService;
 import bisq.core.offer.Offer;
 import bisq.core.proto.CoreProtoResolver;
-import bisq.core.trade.protocol.BuyerAsTakerProtocol;
 import bisq.core.trade.protocol.ProcessModel;
 
 import bisq.network.p2p.NodeAddress;
@@ -104,14 +103,4 @@ public final class BuyerAsTakerTrade extends BuyerTrade implements TakerTrade {
                 proto,
                 coreProtoResolver);
     }
-
-
-    ///////////////////////////////////////////////////////////////////////////////////////////
-    // API
-    ///////////////////////////////////////////////////////////////////////////////////////////
-
-    @Override
-    protected void createTradeProtocol() {
-        tradeProtocol = new BuyerAsTakerProtocol(this);
-    }
 }

core/src/main/java/bisq/core/trade/SellerAsMakerTrade.java

@@ -21,7 +21,6 @@ import bisq.core.btc.wallet.BtcWalletService;
 import bisq.core.offer.Offer;
 import bisq.core.proto.CoreProtoResolver;
 import bisq.core.trade.protocol.ProcessModel;
-import bisq.core.trade.protocol.SellerAsMakerProtocol;
 
 import bisq.network.p2p.NodeAddress;
 
@@ -101,14 +100,4 @@ public final class SellerAsMakerTrade extends SellerTrade implements MakerTrade
                 proto,
                 coreProtoResolver);
     }
-
-
-    ///////////////////////////////////////////////////////////////////////////////////////////
-    // API
-    ///////////////////////////////////////////////////////////////////////////////////////////
-
-    @Override
-    protected void createTradeProtocol() {
-        tradeProtocol = new SellerAsMakerProtocol(this);
-    }
 }

core/src/main/java/bisq/core/trade/SellerAsTakerTrade.java

@@ -21,7 +21,6 @@ import bisq.core.btc.wallet.BtcWalletService;
 import bisq.core.offer.Offer;
 import bisq.core.proto.CoreProtoResolver;
 import bisq.core.trade.protocol.ProcessModel;
-import bisq.core.trade.protocol.SellerAsTakerProtocol;
 
 import bisq.network.p2p.NodeAddress;
 
@@ -104,14 +103,4 @@ public final class SellerAsTakerTrade extends SellerTrade implements TakerTrade
                 proto,
                 coreProtoResolver);
     }
-
-
-    ///////////////////////////////////////////////////////////////////////////////////////////
-    // API
-    ///////////////////////////////////////////////////////////////////////////////////////////
-
-    @Override
-    protected void createTradeProtocol() {
-        tradeProtocol = new SellerAsTakerProtocol(this);
-    }
 }

core/src/main/java/bisq/core/trade/Trade.java

@@ -31,7 +31,6 @@ import bisq.core.support.dispute.refund.RefundResultState;
 import bisq.core.support.messages.ChatMessage;
 import bisq.core.trade.protocol.ProcessModel;
 import bisq.core.trade.protocol.ProcessModelServiceProvider;
-import bisq.core.trade.protocol.TradeProtocol;
 import bisq.core.trade.txproof.AssetTxProofResult;
 
 import bisq.network.p2p.DecryptedMessageWithPubKey;
@@ -384,8 +383,6 @@ public abstract class Trade implements Tradable, Model {
     transient final private StringProperty errorMessageProperty = new SimpleStringProperty();
 
     //  Mutable
-    @Getter
-    transient protected TradeProtocol tradeProtocol;
     @Nullable
     transient private Transaction depositTx;
     @Getter
@@ -402,6 +399,7 @@ public abstract class Trade implements Tradable, Model {
 
     transient private ObjectProperty<Coin> tradeAmountProperty;
     transient private ObjectProperty<Volume> tradeVolumeProperty;
+    @Getter
     final transient private Set<DecryptedMessageWithPubKey> decryptedMessageWithPubKeySet = new HashSet<>();
 
     // Added in v1.1.6
@@ -631,10 +629,6 @@ public abstract class Trade implements Tradable, Model {
         this.btcWalletService = btcWalletService;
     }
 
-    public void setupProcessModel(ProcessModelServiceProvider serviceProvider, TradeManager tradeManager) {
-        processModel.applyTransient(serviceProvider, tradeManager, offer);
-    }
-
     public void init(ProcessModelServiceProvider serviceProvider) {
         serviceProvider.getArbitratorManager().getDisputeAgentByNodeAddress(arbitratorNodeAddress).ifPresent(arbitrator -> {
             arbitratorBtcPubKey = arbitrator.getBtcPubKey();
@@ -652,15 +646,6 @@ public abstract class Trade implements Tradable, Model {
             persist();
         });
 
-        createTradeProtocol();
-
-        // If we have already received a msg we apply it.
-        // removeDecryptedMsgWithPubKey will be called synchronous after apply. We don't have threaded context
-        // or async calls there.
-        // Clone to avoid ConcurrentModificationException. We remove items at the applyMailboxMessage call...
-        HashSet<DecryptedMessageWithPubKey> set = new HashSet<>(decryptedMessageWithPubKeySet);
-        set.forEach(msg -> tradeProtocol.applyMailboxMessage(msg));
-
         isInitialized = true;
     }
 
@@ -727,21 +712,6 @@ public abstract class Trade implements Tradable, Model {
         return delayedPayoutTx;
     }
 
-    // We don't need to persist the msg as if we don't apply it it will not be removed from the P2P network and we
-    // will receive it again on next startup. This might happen in edge cases when the user shuts down after we
-    // received the msg but before the init is called.
-    void addDecryptedMessageWithPubKey(DecryptedMessageWithPubKey decryptedMessageWithPubKey) {
-        if (!decryptedMessageWithPubKeySet.contains(decryptedMessageWithPubKey)) {
-            decryptedMessageWithPubKeySet.add(decryptedMessageWithPubKey);
-
-            // If we have already initialized we apply.
-            // removeDecryptedMsgWithPubKey will be called synchronous after apply. We don't have threaded context
-            // or async calls there.
-            if (tradeProtocol != null)
-                tradeProtocol.applyMailboxMessage(decryptedMessageWithPubKey);
-        }
-    }
-
     public void removeDecryptedMessageWithPubKey(DecryptedMessageWithPubKey decryptedMessageWithPubKey) {
         decryptedMessageWithPubKeySet.remove(decryptedMessageWithPubKey);
     }
@@ -795,8 +765,6 @@ public abstract class Trade implements Tradable, Model {
     // Abstract
     ///////////////////////////////////////////////////////////////////////////////////////////
 
-    protected abstract void createTradeProtocol();
-
     public abstract Coin getPayoutAmount();
 
 
@@ -1234,7 +1202,6 @@ public abstract class Trade implements Tradable, Model {
                 ",\n     disputeStateProperty=" + disputeStateProperty +
                 ",\n     tradePeriodStateProperty=" + tradePeriodStateProperty +
                 ",\n     errorMessageProperty=" + errorMessageProperty +
-                ",\n     tradeProtocol=" + tradeProtocol +
                 ",\n     depositTx=" + depositTx +
                 ",\n     delayedPayoutTx=" + delayedPayoutTx +
                 ",\n     payoutTx=" + payoutTx +

core/src/main/java/bisq/core/trade/TradeManager.java

@@ -39,6 +39,8 @@ import bisq.core.trade.protocol.MakerProtocol;
 import bisq.core.trade.protocol.ProcessModel;
 import bisq.core.trade.protocol.ProcessModelServiceProvider;
 import bisq.core.trade.protocol.TakerProtocol;
+import bisq.core.trade.protocol.TradeProtocol;
+import bisq.core.trade.protocol.TradeProtocolFactory;
 import bisq.core.trade.statistics.TradeStatisticsManager;
 import bisq.core.user.User;
 import bisq.core.util.Validator;
@@ -84,6 +86,8 @@ import javafx.collections.ObservableList;
 import org.bouncycastle.crypto.params.KeyParameter;
 
 import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
@@ -124,9 +128,10 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
     private final ClockWatcher clockWatcher;
 
     private final Storage<TradableList<Trade>> tradableListStorage;
+    private final Map<String, TradeProtocol> tradeProtocolByTradeId = new HashMap<>();
     private TradableList<Trade> tradableList;
     @Getter
-    private final BooleanProperty pendingTradesInitialized = new SimpleBooleanProperty();
+    private final BooleanProperty persistedTradesInitialized = new SimpleBooleanProperty();
     @Setter
     @Nullable
     private ErrorMessageHandler takeOfferRequestErrorMessageHandler;
@@ -210,9 +215,13 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
     @Override
     public void onDirectMessage(DecryptedMessageWithPubKey message, NodeAddress peer) {
         NetworkEnvelope networkEnvelope = message.getNetworkEnvelope();
-        // The maker received a TakeOfferRequest
         if (networkEnvelope instanceof TakeOfferRequest) {
-            TakeOfferRequest takeOfferRequest = (TakeOfferRequest) networkEnvelope;
+            handleTakeOfferRequest(peer, (TakeOfferRequest) networkEnvelope);
+        }
+    }
+
+    // The maker received a TakeOfferRequest
+    private void handleTakeOfferRequest(NodeAddress peer, TakeOfferRequest takeOfferRequest) {
         log.info("Received TakeOfferRequest from {} with tradeId {} and uid {}",
                 peer, takeOfferRequest.getTradeId(), takeOfferRequest.getUid());
 
@@ -235,18 +244,9 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
 
         Offer offer = openOffer.getOffer();
         openOfferManager.reserveOpenOffer(openOffer);
-            Trade trade = offer.isBuyOffer() ?
+        Trade trade;
-                    new BuyerAsMakerTrade(offer,
+        if (offer.isBuyOffer()) {
-                            Coin.valueOf(takeOfferRequest.getTxFee()),
+            trade = new BuyerAsMakerTrade(offer,
-                            Coin.valueOf(takeOfferRequest.getTakerFee()),
-                            takeOfferRequest.isCurrencyForTakerFeeBtc(),
-                            openOffer.getArbitratorNodeAddress(),
-                            openOffer.getMediatorNodeAddress(),
-                            openOffer.getRefundAgentNodeAddress(),
-                            tradableListStorage,
-                            btcWalletService,
-                            getNewProcessModel(offer)) :
-                    new SellerAsMakerTrade(offer,
                     Coin.valueOf(takeOfferRequest.getTxFee()),
                     Coin.valueOf(takeOfferRequest.getTakerFee()),
                     takeOfferRequest.isCurrencyForTakerFeeBtc(),
@@ -256,15 +256,28 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
                     tradableListStorage,
                     btcWalletService,
                     getNewProcessModel(offer));
-
+        } else {
-            initNewMakerTrade(trade);
+            trade = new SellerAsMakerTrade(offer,
+                    Coin.valueOf(takeOfferRequest.getTxFee()),
+                    Coin.valueOf(takeOfferRequest.getTakerFee()),
+                    takeOfferRequest.isCurrencyForTakerFeeBtc(),
+                    openOffer.getArbitratorNodeAddress(),
+                    openOffer.getMediatorNodeAddress(),
+                    openOffer.getRefundAgentNodeAddress(),
+                    tradableListStorage,
+                    btcWalletService,
+                    getNewProcessModel(offer));
+        }
+        TradeProtocol tradeProtocol = TradeProtocolFactory.getNewTradeProtocol(trade);
+        tradeProtocolByTradeId.put(trade.getId(), tradeProtocol);
         tradableList.add(trade);
-            ((MakerProtocol) trade.getTradeProtocol()).handleTakeOfferRequest(takeOfferRequest, peer, errorMessage -> {
+        initTradeAndProtocol(trade, tradeProtocol);
+
+        ((MakerProtocol) tradeProtocol).handleTakeOfferRequest(takeOfferRequest, peer, errorMessage -> {
             if (takeOfferRequestErrorMessageHandler != null)
                 takeOfferRequestErrorMessageHandler.handleErrorMessage(errorMessage);
         });
     }
-    }
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -277,9 +290,23 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
         NetworkEnvelope networkEnvelope = message.getNetworkEnvelope();
         if (networkEnvelope instanceof TradeMessage) {
             TradeMessage tradeMessage = (TradeMessage) networkEnvelope;
-            // The mailbox message will be removed inside the tasks after they are processed successfully
             getTradeById(tradeMessage.getTradeId())
-                    .ifPresent(trade -> trade.addDecryptedMessageWithPubKey(message));
+                    .ifPresent(trade -> {
+                        // We don't need to persist the msg as if we don't processes the message it will not be
+                        // removed from the P2P network and we will receive it again on next startup.
+                        // This might happen in edge cases when the user shuts down after we received the msg but
+                        // before it is processed.
+                        //TODO
+                        Set<DecryptedMessageWithPubKey> decryptedMessageWithPubKeySet = trade.getDecryptedMessageWithPubKeySet();
+                        if (!decryptedMessageWithPubKeySet.contains(message)) {
+                            decryptedMessageWithPubKeySet.add(message);
+
+                            // The message will be removed after processed
+                            TradeProtocol tradeProtocol = getTradeProtocol(trade);
+                            tradeProtocol.applyMailboxMessage(message);
+                        }
+
+                    });
         } else if (networkEnvelope instanceof AckMessage) {
             AckMessage ackMessage = (AckMessage) networkEnvelope;
             if (ackMessage.getSourceType() == AckMessageSourceType.TRADE_MESSAGE) {
@@ -322,6 +349,16 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
         tradableList.persist();
     }
 
+    public TradeProtocol getTradeProtocol(Trade trade) {
+        if (tradeProtocolByTradeId.containsKey(trade.getId())) {
+            return tradeProtocolByTradeId.get(trade.getId());
+        } else {
+            TradeProtocol tradeProtocol = TradeProtocolFactory.getNewTradeProtocol(trade);
+            tradeProtocolByTradeId.put(trade.getId(), tradeProtocol);
+            return tradeProtocol;
+        }
+    }
+
 
     ///////////////////////////////////////////////////////////////////////////////////////////
     // Init pending trade
@@ -329,27 +366,16 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
 
     private void initPersistedTrades() {
         tradableList.forEach(this::initPersistedTrade);
-        pendingTradesInitialized.set(true);
+        persistedTradesInitialized.set(true);
     }
 
     private void initPersistedTrade(Trade trade) {
-        initTrade(trade);
+        initTradeAndProtocol(trade, getTradeProtocol(trade));
         trade.updateDepositTxFromWallet();
     }
 
-    private void initNewMakerTrade(Trade trade) {
+    private void initTradeAndProtocol(Trade trade, TradeProtocol tradeProtocol) {
-        initTrade(trade);
+        tradeProtocol.initialize(processModelServiceProvider, this, trade.getOffer());
-    }
-
-    private void initNewTakerTrade(Trade trade, boolean useSavingsWallet, Coin fundsNeededForTrade) {
-        initTrade(trade);
-
-        trade.getProcessModel().setUseSavingsWallet(useSavingsWallet);
-        trade.getProcessModel().setFundsNeededForTradeAsLong(fundsNeededForTrade.value);
-    }
-
-    private void initTrade(Trade trade) {
-        trade.setupProcessModel(processModelServiceProvider, this);
         trade.init(processModelServiceProvider);
     }
 
@@ -391,49 +417,9 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
         offer.checkOfferAvailability(model,
                 () -> {
                     if (offer.getState() == Offer.State.AVAILABLE) {
-                        Trade trade = getNewTrade(amount,
+                        Trade trade;
-                                txFee,
+                        if (offer.isBuyOffer()) {
-                                takerFee,
+                            trade = new SellerAsTakerTrade(offer,
-                                isCurrencyForTakerFeeBtc,
-                                tradePrice,
-                                fundsNeededForTrade,
-                                offer,
-                                paymentAccountId,
-                                useSavingsWallet,
-                                model);
-                        tradableList.add(trade);
-                        ((TakerProtocol) trade.getTradeProtocol()).onTakeOffer();
-                        tradeResultHandler.handleResult(trade);
-                    }
-                },
-                errorMessageHandler);
-    }
-
-    private Trade getNewTrade(Coin amount,
-                              Coin txFee,
-                              Coin takerFee,
-                              boolean isCurrencyForTakerFeeBtc,
-                              long tradePrice,
-                              Coin fundsNeededForTrade,
-                              Offer offer,
-                              String paymentAccountId,
-                              boolean useSavingsWallet,
-                              OfferAvailabilityModel model) {
-        Trade trade = offer.isBuyOffer() ?
-                new SellerAsTakerTrade(offer,
-                        amount,
-                        txFee,
-                        takerFee,
-                        isCurrencyForTakerFeeBtc,
-                        tradePrice,
-                        model.getPeerNodeAddress(),
-                        model.getSelectedArbitrator(),
-                        model.getSelectedMediator(),
-                        model.getSelectedRefundAgent(),
-                        tradableListStorage,
-                        btcWalletService,
-                        getNewProcessModel(offer)) :
-                new BuyerAsTakerTrade(offer,
                                     amount,
                                     txFee,
                                     takerFee,
@@ -446,9 +432,36 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
                                     tradableListStorage,
                                     btcWalletService,
                                     getNewProcessModel(offer));
+                        } else {
+                            trade = new BuyerAsTakerTrade(offer,
+                                    amount,
+                                    txFee,
+                                    takerFee,
+                                    isCurrencyForTakerFeeBtc,
+                                    tradePrice,
+                                    model.getPeerNodeAddress(),
+                                    model.getSelectedArbitrator(),
+                                    model.getSelectedMediator(),
+                                    model.getSelectedRefundAgent(),
+                                    tradableListStorage,
+                                    btcWalletService,
+                                    getNewProcessModel(offer));
+                        }
+                        trade.getProcessModel().setUseSavingsWallet(useSavingsWallet);
+                        trade.getProcessModel().setFundsNeededForTradeAsLong(fundsNeededForTrade.value);
                         trade.setTakerPaymentAccountId(paymentAccountId);
-        initNewTakerTrade(trade, useSavingsWallet, fundsNeededForTrade);
+
-        return trade;
+                        TradeProtocol tradeProtocol = TradeProtocolFactory.getNewTradeProtocol(trade);
+                        tradeProtocolByTradeId.put(trade.getId(), tradeProtocol);
+                        tradableList.add(trade);
+
+                        initTradeAndProtocol(trade, tradeProtocol);
+
+                        ((TakerProtocol) tradeProtocol).onTakeOffer();
+                        tradeResultHandler.handleResult(trade);
+                    }
+                },
+                errorMessageHandler);
     }
 
     private ProcessModel getNewProcessModel(Offer offer) {
@@ -483,7 +496,7 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
                     log.debug("onWithdraw onSuccess tx ID:" + transaction.getTxId().toString());
                     onTradeCompleted(trade);
                     trade.setState(Trade.State.WITHDRAW_COMPLETED);
-                    trade.getTradeProtocol().onWithdrawCompleted();
+                    getTradeProtocol(trade).onWithdrawCompleted();
                     resultHandler.handleResult();
                 }
             }
@@ -664,8 +677,8 @@ public class TradeManager implements PersistedDataHost, DecryptedDirectMessageLi
         return tradableList.getList();
     }
 
-    public BooleanProperty pendingTradesInitializedProperty() {
+    public BooleanProperty persistedTradesInitializedProperty() {
-        return pendingTradesInitialized;
+        return persistedTradesInitialized;
     }
 
     public boolean isMyOffer(Offer offer) {

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

@@ -62,7 +62,11 @@ public class BuyerAsTakerProtocol extends BuyerProtocol implements TakerProtocol
 
     public BuyerAsTakerProtocol(BuyerAsTakerTrade trade) {
         super(trade);
+    }
 
+    @Override
+    protected void onInitialized() {
+        super.onInitialized();
         Offer offer = checkNotNull(trade.getOffer());
         processModel.getTradingPeer().setPubKeyRing(offer.getPubKeyRing());
     }

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

@@ -54,7 +54,11 @@ public abstract class BuyerProtocol extends DisputeProtocol {
 
     public BuyerProtocol(BuyerTrade trade) {
         super(trade);
+    }
 
+    @Override
+    protected void onInitialized() {
+        super.onInitialized();
         // We get called the constructor with any possible state and phase. As we don't want to log an error for such
         // cases we use the alternative 'given' method instead of 'expect'.
         given(phase(Trade.Phase.TAKER_FEE_PUBLISHED)

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

@@ -57,7 +57,11 @@ public class SellerAsTakerProtocol extends SellerProtocol implements TakerProtoc
 
     public SellerAsTakerProtocol(SellerAsTakerTrade trade) {
         super(trade);
+    }
 
+    @Override
+    protected void onInitialized() {
+        super.onInitialized();
         Offer offer = checkNotNull(trade.getOffer());
         processModel.getTradingPeer().setPubKeyRing(offer.getPubKeyRing());
     }

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

@@ -17,7 +17,9 @@
 
 package bisq.core.trade.protocol;
 
+import bisq.core.offer.Offer;
 import bisq.core.trade.Trade;
+import bisq.core.trade.TradeManager;
 import bisq.core.trade.messages.CounterCurrencyTransferStartedMessage;
 import bisq.core.trade.messages.DepositTxAndDelayedPayoutTxMessage;
 import bisq.core.trade.messages.TradeMessage;
@@ -36,6 +38,8 @@ import bisq.common.crypto.PubKeyRing;
 import bisq.common.proto.network.NetworkEnvelope;
 import bisq.common.taskrunner.Task;
 
+import java.util.HashSet;
+
 import lombok.extern.slf4j.Slf4j;
 
 import javax.annotation.Nullable;
@@ -46,6 +50,7 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener {
     protected final ProcessModel processModel;
     protected final Trade trade;
     private Timer timeoutTimer;
+    private boolean initialized;
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -55,10 +60,6 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener {
     public TradeProtocol(Trade trade) {
         this.trade = trade;
         this.processModel = trade.getProcessModel();
-
-        if (!trade.isWithdrawn()) {
-            processModel.getP2PService().addDecryptedDirectMessageListener(this);
-        }
     }
 
 
@@ -66,12 +67,28 @@ public abstract class TradeProtocol implements DecryptedDirectMessageListener {
     // API
     ///////////////////////////////////////////////////////////////////////////////////////////
 
+    public void initialize(ProcessModelServiceProvider serviceProvider, TradeManager tradeManager, Offer offer) {
+        processModel.applyTransient(serviceProvider, tradeManager, offer);
+        initialized = true;
+        onInitialized();
+    }
+
+    protected void onInitialized() {
+        if (!trade.isWithdrawn()) {
+            processModel.getP2PService().addDecryptedDirectMessageListener(this);
+        }
+
+        // Apply mailbox messages
+        // Clone to avoid ConcurrentModificationException. We remove items at the applyMailboxMessage call...
+        new HashSet<>(trade.getDecryptedMessageWithPubKeySet()).forEach(this::applyMailboxMessage);
+    }
+
     public void onWithdrawCompleted() {
         cleanup();
     }
 
     public void applyMailboxMessage(DecryptedMessageWithPubKey message) {
-        if (isPubKeyValid(message)) {
+        if (initialized && isPubKeyValid(message)) {
             NetworkEnvelope networkEnvelope = message.getNetworkEnvelope();
             if (networkEnvelope instanceof MailboxMessage &&
                     networkEnvelope instanceof TradeMessage) {

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

@@ -0,0 +1,40 @@
+/*
+ * This file is part of Bisq.
+ *
+ * Bisq is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or (at
+ * your option) any later version.
+ *
+ * Bisq is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
+ * License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Bisq. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package bisq.core.trade.protocol;
+
+import bisq.core.trade.BuyerAsMakerTrade;
+import bisq.core.trade.BuyerAsTakerTrade;
+import bisq.core.trade.SellerAsMakerTrade;
+import bisq.core.trade.SellerAsTakerTrade;
+import bisq.core.trade.Trade;
+
+public class TradeProtocolFactory {
+    public static TradeProtocol getNewTradeProtocol(Trade trade) {
+        if (trade instanceof BuyerAsMakerTrade) {
+            return new BuyerAsMakerProtocol((BuyerAsMakerTrade) trade);
+        } else if (trade instanceof BuyerAsTakerTrade) {
+            return new BuyerAsTakerProtocol((BuyerAsTakerTrade) trade);
+        } else if (trade instanceof SellerAsMakerTrade) {
+            return new SellerAsMakerProtocol((SellerAsMakerTrade) trade);
+        } else if (trade instanceof SellerAsTakerTrade) {
+            return new SellerAsTakerProtocol((SellerAsTakerTrade) trade);
+        } else {
+            throw new IllegalStateException("Trade not of expected type. Trade=" + trade);
+        }
+    }
+}

core/src/main/java/bisq/core/trade/txproof/xmr/XmrTxProofService.java

@@ -264,7 +264,7 @@ public class XmrTxProofService implements AssetTxProofService {
                         log.info("We auto-confirm trade {} as our all our services for the tx proof completed successfully", trade.getShortId());
                         log.info("###########################################################################################");
 
-                        ((SellerProtocol) trade.getTradeProtocol()).onPaymentReceived(() -> {
+                        ((SellerProtocol) tradeManager.getTradeProtocol(trade)).onPaymentReceived(() -> {
                         }, errorMessage -> {
                         });
                     }

desktop/src/main/java/bisq/desktop/main/MainViewModel.java

@@ -214,7 +214,8 @@ public class MainViewModel implements ViewModel, BisqSetup.BisqSetupListener {
     @Override
     public void onSetupComplete() {
         // We handle the trade period here as we display a global popup if we reached dispute time
-        tradesAndUIReady = EasyBind.combine(isSplashScreenRemoved, tradeManager.pendingTradesInitializedProperty(), (a, b) -> a && b);
+        tradesAndUIReady = EasyBind.combine(isSplashScreenRemoved, tradeManager.persistedTradesInitializedProperty(),
+                (a, b) -> a && b);
         tradesAndUIReady.subscribe((observable, oldValue, newValue) -> {
             if (newValue) {
                 tradeManager.applyTradePeriodState();

desktop/src/main/java/bisq/desktop/main/portfolio/pendingtrades/PendingTradesDataModel.java

@@ -189,13 +189,14 @@ public class PendingTradesDataModel extends ActivatableDataModel {
         Trade trade = getTrade();
         checkNotNull(trade, "trade must not be null");
         checkArgument(trade instanceof BuyerTrade, "Check failed: trade instanceof BuyerTrade");
-        ((BuyerProtocol) trade.getTradeProtocol()).onPaymentStarted(resultHandler, errorMessageHandler);
+        ((BuyerProtocol) tradeManager.getTradeProtocol(trade)).onPaymentStarted(resultHandler, errorMessageHandler);
     }
 
     public void onFiatPaymentReceived(ResultHandler resultHandler, ErrorMessageHandler errorMessageHandler) {
-        checkNotNull(getTrade(), "trade must not be null");
+        Trade trade = getTrade();
-        checkArgument(getTrade() instanceof SellerTrade, "Trade must be instance of SellerTrade");
+        checkNotNull(trade, "trade must not be null");
-        ((SellerProtocol) getTrade().getTradeProtocol()).onPaymentReceived(resultHandler, errorMessageHandler);
+        checkArgument(trade instanceof SellerTrade, "Trade must be instance of SellerTrade");
+        ((SellerProtocol) tradeManager.getTradeProtocol(trade)).onPaymentReceived(resultHandler, errorMessageHandler);
     }
 
     public void onWithdrawRequest(String toAddress,
@@ -643,8 +644,7 @@ public class PendingTradesDataModel extends ActivatableDataModel {
 
             trade.setDisputeState(Trade.DisputeState.REFUND_REQUESTED);
 
-            //todo add UI spinner as it can take a bit if peer is offline
+            ((DisputeProtocol) tradeManager.getTradeProtocol(trade)).onPublishDelayedPayoutTx(() -> {
-            ((DisputeProtocol) trade.getTradeProtocol()).onPublishDelayedPayoutTx(() -> {
                         log.info("DelayedPayoutTx published and message sent to peer");
                         disputeManager.sendOpenNewDisputeMessage(dispute,
                                 false,

desktop/src/main/java/bisq/desktop/main/portfolio/pendingtrades/steps/TradeStepView.java

@@ -222,17 +222,17 @@ public abstract class TradeStepView extends AnchorPane {
             infoLabel.setText(getInfoText());
         }
 
-        BooleanProperty pendingTradesInitialized = model.dataModel.tradeManager.getPendingTradesInitialized();
+        BooleanProperty initialized = model.dataModel.tradeManager.getPersistedTradesInitialized();
-        if (pendingTradesInitialized.get()) {
+        if (initialized.get()) {
             onPendingTradesInitialized();
         } else {
             pendingTradesInitializedListener = (observable, oldValue, newValue) -> {
                 if (newValue) {
                     onPendingTradesInitialized();
-                    UserThread.execute(() -> pendingTradesInitialized.removeListener(pendingTradesInitializedListener));
+                    UserThread.execute(() -> initialized.removeListener(pendingTradesInitializedListener));
                 }
             };
-            pendingTradesInitialized.addListener(pendingTradesInitializedListener);
+            initialized.addListener(pendingTradesInitializedListener);
         }
     }