Merge pull request #206 from alvasw/wallet_zeromq

Bitcoind: ZeroMQ Support + Auto-Update Balance in DesktopApp

application/src/main/java/bisq/application/DefaultApplicationService.java

@@ -165,7 +165,7 @@ public DefaultApplicationService(String[] args) {
         protocolService = new ProtocolService(networkService, identityService, persistenceService, openOfferService);
 
         Optional<WalletConfig> walletConfig = !isRegtestRun() ? Optional.empty() : createRegtestWalletConfig();
-        walletService = new WalletService(walletConfig);
+        walletService = new WalletService(persistenceService, walletConfig);
 
         daoBridgeService = new DaoBridgeService(networkService, identityService, getConfig("bisq.oracle.daoBridge"));
     }
@@ -206,12 +206,12 @@ public CompletableFuture<Boolean> initialize() {
                 .thenCompose(result -> setStateAfter(accountAgeWitnessService.initialize(), State.ACCOUNT_AGE_WITNESS_SERVICE_INITIALIZED))
                 .thenCompose(result -> setStateAfterList(protocolService.initialize(), State.PROTOCOL_SERVICE_INITIALIZED))
                 .thenCompose(result -> CompletableFutureUtils.allOf(
-                        walletService.tryAutoInitialization(),
                         userProfileService.initialize()
                                 .thenCompose(res -> chatService.initialize()),
                         openOfferService.initialize(),
                         offerBookService.initialize(),
-                        tradeIntentService.initialize()))
+                        tradeIntentService.initialize(),
+                        walletService.initialize()))
                 // TODO Needs to increase if using embedded I2P router (i2p internal bootstrap timeouts after 5 mins)
                 .orTimeout(5, TimeUnit.MINUTES)
                 .thenApply(list -> list.stream().allMatch(e -> e))

buildSrc/src/main/kotlin/bisq/gradle/Network.kt

@@ -2,6 +2,7 @@ package bisq.gradle
 
 import java.io.IOException
 import java.net.ServerSocket
+import java.util.*
 
 object Network {
     fun isPortFree(port: Int): Boolean =
@@ -12,4 +13,15 @@ object Network {
         } catch (e: IOException) {
             false
         }
+
+    fun findFreeSystemPort(): Int {
+        return try {
+            val server = ServerSocket(0)
+            val port = server.localPort
+            server.close()
+            port
+        } catch (ignored: IOException) {
+            Random().nextInt(10000) + 50000
+        }
+    }
 }

buildSrc/src/main/kotlin/bisq/gradle/tasks/bitcoind/StartBitcoinQtTask.kt

@@ -39,6 +39,7 @@ abstract class StartBitcoinQtTask : DefaultTask() {
     }
 
     private fun spawnBitcoinQtProcess(bitcoindDataDir: File) {
+        val zmqPort = Network.findFreeSystemPort()
         ProcessBuilder(
             listOf(
                 "bitcoin-qt",
@@ -52,6 +53,9 @@ abstract class StartBitcoinQtTask : DefaultTask() {
                 "-rpcuser=bisq",
                 "-rpcpassword=bisq",
 
+                "-zmqpubhashblock=tcp://127.0.0.1:$zmqPort",
+                "-zmqpubrawtx=tcp://127.0.0.1:$zmqPort",
+
                 "-fallbackfee=0.00000001",
                 "-txindex=1"
             )

desktop/src/main/java/bisq/desktop/primary/main/content/wallet/config/WalletConfigPopup.java

@@ -91,7 +91,7 @@ private void onConnectToWallet() {
             model.walletPassphraseProperty.setValue(""); // Wipe passphrase from memory
 
             WalletConfig walletConfig = createWalletConfigFromModel();
-            walletService.initialize(walletConfig, passphrase)
+            walletService.loadOrCreateWallet(walletConfig, passphrase)
                     .whenComplete((__, throwable) -> {
                         if (throwable == null) {
                             UIThread.run(() -> {

desktop/src/main/java/bisq/desktop/primary/main/content/wallet/receive/WalletReceiveController.java

@@ -18,7 +18,8 @@
 package bisq.desktop.primary.main.content.wallet.receive;
 
 import bisq.application.DefaultApplicationService;
-import bisq.desktop.common.threading.UIThread;
+import bisq.common.observable.Pin;
+import bisq.desktop.common.observable.FxBindings;
 import bisq.desktop.common.view.Controller;
 import bisq.wallets.WalletService;
 import lombok.Getter;
@@ -29,6 +30,8 @@ public class WalletReceiveController implements Controller {
     @Getter
     private final WalletReceiveView view;
 
+    private Pin receiveAddressListPin;
+
     public WalletReceiveController(DefaultApplicationService applicationService) {
         this.walletService = applicationService.getWalletService();
         model = new WalletReceiveModel();
@@ -37,14 +40,16 @@ public WalletReceiveController(DefaultApplicationService applicationService) {
 
     @Override
     public void onActivate() {
+        receiveAddressListPin = FxBindings.<String, String>bind(model.getListItems())
+                .to(walletService.getReceiveAddresses());
     }
 
     @Override
     public void onDeactivate() {
+        receiveAddressListPin.unbind();
     }
 
     public void onGenerateNewAddress() {
-        walletService.getNewAddress("")
-                .thenAccept(newAddress -> UIThread.run(() -> model.addNewAddress(newAddress)));
+        walletService.getNewAddress("");
     }
 }

desktop/src/main/java/bisq/desktop/primary/main/content/wallet/receive/WalletReceiveModel.java

@@ -20,15 +20,9 @@
 import bisq.desktop.common.view.Model;
 import javafx.collections.FXCollections;
 import javafx.collections.ObservableList;
+import lombok.Getter;
 
 public class WalletReceiveModel implements Model {
-    private final ObservableList<String> listItems = FXCollections.observableArrayList();
-
-    public void addNewAddress(String address) {
-        listItems.add(address);
-    }
-
-    public ObservableList<String> getListItems() {
-        return listItems;
-    }
+    @Getter
+    final ObservableList<String> listItems = FXCollections.observableArrayList();
 }

wallets/build.gradle

@@ -10,19 +10,24 @@ apply from: '../buildSrc/test-dependencies.gradle'
 ext {
     jsonrpc4jVersion = '1.6.0.bisq.2'
     jacksonDatabindVersion = '2.12.0'
+    jeroMqVersion = '0.5.2'
 }
 
 dependencies {
     api platform(project(':platforms:common-platform'))
 
     implementation project(':common')
+    implementation project(':persistence')
 
     implementation("com.fasterxml.jackson.core:jackson-databind:$jacksonDatabindVersion")
     implementation("com.github.bisq-network:jsonrpc4j:$jsonrpc4jVersion") {
         exclude(module: 'base64')
         exclude(module: 'httpcore-nio')
     }
+
     implementation libs.google.guava
+    implementation libs.protobuf.java
+    implementation("org.zeromq:jeromq:$jeroMqVersion")
 
     testImplementation("org.assertj:assertj-core:3.22.0")
 }

wallets/src/main/java/bisq/wallets/WalletService.java

@@ -19,68 +19,93 @@
 
 import bisq.common.monetary.Coin;
 import bisq.common.observable.Observable;
+import bisq.common.observable.ObservableSet;
+import bisq.persistence.Persistence;
+import bisq.persistence.PersistenceClient;
+import bisq.persistence.PersistenceService;
 import bisq.wallets.bitcoind.BitcoinWallet;
+import bisq.wallets.bitcoind.rpc.BitcoindDaemon;
+import bisq.wallets.bitcoind.zeromq.BitcoindZeroMq;
 import bisq.wallets.elementsd.LiquidWallet;
 import bisq.wallets.exceptions.WalletNotInitializedException;
 import bisq.wallets.model.Transaction;
 import bisq.wallets.model.Utxo;
 import bisq.wallets.rpc.RpcConfig;
+import bisq.wallets.stores.WalletStore;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
 import java.nio.file.Path;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
 @Slf4j
-public class WalletService {
+public class WalletService implements PersistenceClient<WalletStore> {
     private static final String LOCALHOST = "127.0.0.1";
 
+    @Getter
+    private final WalletStore persistableStore = new WalletStore();
+    @Getter
+    private final Persistence<WalletStore> persistence;
+
     private final Optional<WalletConfig> walletConfig;
     @Getter
     private Optional<Wallet> wallet = Optional.empty();
+
+    private Optional<BitcoindZeroMq> bitcoindZeroMq = Optional.empty();
+    private final Set<String> utxoTxIds = new HashSet<>();
+
     @Getter
     private final Observable<Coin> observableBalanceAsCoin = new Observable<>(Coin.of(0, "BTC"));
 
-    public WalletService(Optional<WalletConfig> walletConfig) {
+    public WalletService(PersistenceService persistenceService,
+                         Optional<WalletConfig> walletConfig) {
+        persistence = persistenceService.getOrCreatePersistence(this, persistableStore);
         this.walletConfig = walletConfig;
     }
 
-    public CompletableFuture<Boolean> tryAutoInitialization() {
-        if (walletConfig.isEmpty()) {
-            return CompletableFuture.completedFuture(true);
+    public CompletableFuture<Boolean> initialize() {
+        log.info("initialize");
+
+        if (walletConfig.isPresent()) {
+            return loadOrCreateWallet(walletConfig.get(), Optional.empty());
         }
-        return initialize(walletConfig.get(), Optional.empty());
+
+        return CompletableFuture.completedFuture(true);
     }
 
-    public CompletableFuture<Boolean> initialize(WalletConfig walletConfig, Optional<String> walletPassphrase) {
-        if (wallet.isPresent()) {
-            return CompletableFuture.completedFuture(true);
+    public CompletableFuture<Boolean> loadOrCreateWallet(WalletConfig walletConfig, Optional<String> walletPassphrase) {
+        if (wallet.isEmpty()) {
+            Path walletsDataDir = walletConfig.getWalletsDataDirPath();
+            walletsDataDir.toFile().mkdirs();
+
+            Wallet wallet = switch (walletConfig.getWalletBackend()) {
+                case BITCOIND -> {
+                    var bitcoindWallet = createBitcoinWallet(walletConfig, walletsDataDir);
+                    bitcoindWallet.initialize(walletPassphrase);
+
+                    BitcoindZeroMq bitcoindZeroMq = initializeBitcoindZeroMq(bitcoindWallet.getDaemon());
+                    this.bitcoindZeroMq = Optional.of(bitcoindZeroMq);
+
+                    yield bitcoindWallet;
+                }
+                case ELEMENTSD -> {
+                    var liquidWallet = createLiquidWallet(walletConfig, walletsDataDir);
+                    liquidWallet.initialize(walletPassphrase);
+                    yield liquidWallet;
+                }
+            };
+
+            this.wallet = Optional.of(wallet);
+            log.info("Successfully created/loaded wallet at {}", walletsDataDir);
+
+            updateBalance();
         }
 
-        return CompletableFuture.runAsync(() -> {
-                    Path walletsDataDir = walletConfig.getWalletsDataDirPath();
-                    walletsDataDir.toFile().mkdirs();
-
-                    Wallet wallet = switch (walletConfig.getWalletBackend()) {
-                        case BITCOIND -> {
-                            var bitcoindWallet = createBitcoinWallet(walletConfig, walletsDataDir);
-                            bitcoindWallet.initialize(walletPassphrase);
-                            yield bitcoindWallet;
-                        }
-                        case ELEMENTSD -> {
-                            var liquidWallet = createLiquidWallet(walletConfig, walletsDataDir);
-                            liquidWallet.initialize(walletPassphrase);
-                            yield liquidWallet;
-                        }
-                    };
-
-                    this.wallet = Optional.of(wallet);
-                    log.info("Successfully created wallet at {}", walletsDataDir);
-                })
-                .thenRun(this::getBalance)
-                .thenApply(e -> true);
+        return CompletableFuture.completedFuture(true);
     }
 
     private BitcoinWallet createBitcoinWallet(WalletConfig walletConfig, Path walletsDataDir) {
@@ -95,31 +120,78 @@ private LiquidWallet createLiquidWallet(WalletConfig walletConfig, Path walletsD
         return new LiquidWallet(bitcoindDataDir, rpcConfig);
     }
 
+    private BitcoindZeroMq initializeBitcoindZeroMq(BitcoindDaemon bitcoindDaemon) {
+        var bitcoindZeroMq = new BitcoindZeroMq(bitcoindDaemon);
+        bitcoindZeroMq.initialize();
+        // Update balance when new block gets mined
+        bitcoindZeroMq.getListeners().registerNewBlockMinedListener(unused -> updateBalance());
+
+        // Update balance if a UTXO is spent
+        bitcoindZeroMq.getListeners().registerTransactionIdInInputListener(txId -> {
+            if (utxoTxIds.contains(txId)) {
+                updateBalance();
+            }
+        });
+
+        // Update balance if a receive address is in tx output
+        ObservableSet<String> receiveAddresses = persistableStore.getReceiveAddresses();
+        bitcoindZeroMq.getListeners().registerTxOutputAddressesListener(addresses -> {
+            boolean receiveAddressInTxOutput = addresses.stream().anyMatch(receiveAddresses::contains);
+            if (receiveAddressInTxOutput) {
+                updateBalance();
+            }
+        });
+
+        return bitcoindZeroMq;
+    }
+
     public CompletableFuture<Void> shutdown() {
-        return CompletableFuture.runAsync(() -> wallet.ifPresent(Wallet::shutdown));
+        return CompletableFuture.runAsync(() -> {
+            bitcoindZeroMq.ifPresent(BitcoindZeroMq::shutdown);
+            wallet.ifPresent(Wallet::shutdown);
+        });
     }
 
     public boolean isWalletReady() {
         return walletConfig.isPresent() || wallet.isPresent();
     }
 
-    public CompletableFuture<Long> getBalance() {
-        return CompletableFuture.supplyAsync(() -> {
+    private void updateBalance() {
+        CompletableFuture.runAsync(() -> {
             Wallet wallet = getWalletOrThrowException();
-            double walletBalance = wallet.getBalance();
-            Coin coin = Coin.of(walletBalance, "BTC");
-            observableBalanceAsCoin.set(coin);
-            return coin.getValue();
+            double balance = wallet.getBalance();
+            Coin coin = Coin.of(balance, "BTC");
+
+            // Balance changed?
+            if (!observableBalanceAsCoin.get().equals(coin)) {
+                observableBalanceAsCoin.set(coin);
+
+                listUnspent().thenAccept(utxos -> {
+                    utxoTxIds.clear();
+                    utxos.stream()
+                            .map(Utxo::getTxId)
+                            .forEach(utxoTxIds::add);
+                });
+            }
         });
     }
 
     public CompletableFuture<String> getNewAddress(String label) {
         return CompletableFuture.supplyAsync(() -> {
             Wallet wallet = getWalletOrThrowException();
-            return wallet.getNewAddress(AddressType.BECH32, label);
+            String receiveAddress = wallet.getNewAddress(AddressType.BECH32, label);
+
+            getReceiveAddresses().add(receiveAddress);
+            persist();
+
+            return receiveAddress;
         });
     }
 
+    public ObservableSet<String> getReceiveAddresses() {
+        return persistableStore.getReceiveAddresses();
+    }
+
     public CompletableFuture<String> signMessage(String address, String message) {
         return CompletableFuture.supplyAsync(() -> {
             Wallet wallet = getWalletOrThrowException();

wallets/src/main/java/bisq/wallets/bitcoind/BitcoinWallet.java

@@ -27,6 +27,7 @@
 import bisq.wallets.rpc.RpcClient;
 import bisq.wallets.rpc.RpcClientFactory;
 import bisq.wallets.rpc.RpcConfig;
+import lombok.Getter;
 
 import java.net.MalformedURLException;
 import java.nio.file.Path;
@@ -36,6 +37,7 @@
 public class BitcoinWallet implements Wallet {
     private final Path walletPath;
 
+    @Getter
     private final BitcoindDaemon daemon;
     private final BitcoindWallet wallet;