Improve btc monitor

Signed-off-by: HenrikJannsen <boilingfrog@gmx.com>

btcnodemonitor/src/main/java/bisq/btcnodemonitor/BtcNodeMonitor.java

@@ -54,8 +54,7 @@ public class BtcNodeMonitor {
         return simpleHttpServer.start()
                 .thenCompose(nil -> proxySetup.createSocksProxy())
                 .thenAccept(peerGroupService::applySocks5Proxy)
-                .thenCompose(nil -> peerGroupService.start())
+                .thenCompose(nil -> peerGroupService.start());
-                .thenRun(peerGroupService::connectToAll);
     }
 
     public CompletableFuture<Void> shutdown() {

btcnodemonitor/src/main/java/bisq/btcnodemonitor/BtcNodeMonitorMain.java

@@ -51,6 +51,7 @@ public class BtcNodeMonitorMain implements GracefulShutDownHandler {
 
     @Override
     public void gracefulShutDown(ResultHandler resultHandler) {
+        log.info("gracefulShutDown");
         btcNodeMonitor.shutdown().join();
         System.exit(0);
         resultHandler.handleResult();

btcnodemonitor/src/main/java/bisq/btcnodemonitor/btc/PeerConncetionInfo.java

@@ -64,7 +64,9 @@ public class PeerConncetionInfo {
     }
 
     public String getShortId() {
-        return getAddress().substring(0, 12) + "...";
+        String address = getAddress();
+        int endIndex = Math.min(address.length(), 12);
+        return address.substring(0, endIndex) + "...";
     }
 
     public int getNumConnectionAttempts() {

btcnodemonitor/src/main/java/bisq/btcnodemonitor/btc/PeerConnection.java

@@ -17,8 +17,6 @@
 
 package bisq.btcnodemonitor.btc;
 
-import bisq.common.Timer;
-import bisq.common.UserThread;
 import bisq.common.util.SingleThreadExecutorUtils;
 
 import org.bitcoinj.core.Context;
@@ -27,6 +25,8 @@ import org.bitcoinj.core.Peer;
 import org.bitcoinj.core.PeerAddress;
 import org.bitcoinj.core.Utils;
 import org.bitcoinj.core.VersionMessage;
+import org.bitcoinj.core.listeners.PeerConnectedEventListener;
+import org.bitcoinj.core.listeners.PeerDisconnectedEventListener;
 import org.bitcoinj.net.BlockingClientManager;
 import org.bitcoinj.net.ClientConnectionManager;
 
@@ -39,6 +39,8 @@ import java.net.SocketAddress;
 
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import lombok.extern.slf4j.Slf4j;
@@ -55,9 +57,15 @@ public class PeerConnection {
     private final int connectTimeoutMillis;
     private final int vMinRequiredProtocolVersion;
     private final PeerConncetionInfo peerConncetionInfo;
-    private Optional<Timer> disconnectScheduler = Optional.empty();
-    private Optional<Timer> reconnectScheduler = Optional.empty();
     private final AtomicBoolean shutdownCalled = new AtomicBoolean();
+    private final ExecutorService connectionExecutor;
+    private final ExecutorService onConnectionExecutor;
+    private final ExecutorService onDisConnectionExecutor;
+    private Optional<PeerConnectedEventListener> peerConnectedEventListener = Optional.empty();
+    private Optional<PeerDisconnectedEventListener> peerDisconnectedEventListener = Optional.empty();
+    private Optional<CompletableFuture<Void>> connectAndDisconnectFuture = Optional.empty();
+    private Optional<CompletableFuture<Void>> innerConnectAndDisconnectFuture = Optional.empty();
+    private Optional<CompletableFuture<SocketAddress>> openConnectionFuture = Optional.empty();
 
     public PeerConnection(Context context,
                           PeerConncetionInfo peerConncetionInfo,
@@ -72,37 +80,110 @@ public class PeerConnection {
         this.disconnectIntervalSec = disconnectIntervalSec;
         this.reconnectIntervalSec = reconnectIntervalSec;
 
+        connectionExecutor = SingleThreadExecutorUtils.getSingleThreadExecutor("connection-" + peerConncetionInfo.getShortId());
+        onConnectionExecutor = SingleThreadExecutorUtils.getSingleThreadExecutor("onConnection-" + peerConncetionInfo.getShortId());
+        onDisConnectionExecutor = SingleThreadExecutorUtils.getSingleThreadExecutor("onDisConnection-" + peerConncetionInfo.getShortId());
+
         this.params = context.getParams();
         vMinRequiredProtocolVersion = params.getProtocolVersionNum(NetworkParameters.ProtocolVersion.BLOOM_FILTER);
     }
 
+    public void start() {
+        CompletableFuture.runAsync(() -> {
+            do {
+                connectAndDisconnectFuture = Optional.of(connectAndDisconnect());
+                try {
+                    connectAndDisconnectFuture.get().join();
+                } catch (Exception ignore) {
+                }
+            }
+            while (!shutdownCalled.get() && !Thread.currentThread().isInterrupted());
+            log.info("Exiting startConnectAndDisconnectLoop loop. Expected at shutdown");
+        }, connectionExecutor);
+    }
+
     public CompletableFuture<Void> shutdown() {
+        log.info("Shutdown");
         shutdownCalled.set(true);
-        disconnectScheduler.ifPresent(Timer::stop);
+        peerConncetionInfo.getCurrentConnectionAttempt().ifPresent(connectionAttempt -> {
-        reconnectScheduler.ifPresent(Timer::stop);
+            Peer peer = connectionAttempt.getPeer();
+            peerDisconnectedEventListener.ifPresent(peer::removeDisconnectedEventListener);
+            peerDisconnectedEventListener = Optional.empty();
+            peerConnectedEventListener.ifPresent(peer::removeConnectedEventListener);
+            peerConnectedEventListener = Optional.empty();
+        });
+
         return CompletableFuture.runAsync(() -> {
-            log.info("shutdown {}", peerConncetionInfo);
             Context.propagate(context);
-            disconnect();
+            peerConncetionInfo.getCurrentConnectionAttempt()
+                    .ifPresent(currentConnectionAttempt -> currentConnectionAttempt.getPeer().close());
+
+            connectAndDisconnectFuture.ifPresent(connectFuture -> connectFuture.complete(null));
+            innerConnectAndDisconnectFuture.ifPresent(connectFuture -> connectFuture.complete(null));
+
+            connectionExecutor.shutdownNow();
+            onConnectionExecutor.shutdownNow();
+            onDisConnectionExecutor.shutdownNow();
         }, SingleThreadExecutorUtils.getSingleThreadExecutor("shutdown-" + peerConncetionInfo.getShortId()));
     }
 
-    public void connect() {
+    private CompletableFuture<Void> connectAndDisconnect() {
-        CompletableFuture.runAsync(() -> {
+        CompletableFuture<Void> future = new CompletableFuture<>();
-            log.info("connect {}", peerConncetionInfo);
+        innerConnectAndDisconnectFuture = Optional.of(CompletableFuture.runAsync(() -> {
+            log.info("\n>> Connect to {}", peerConncetionInfo.getAddress());
             Context.propagate(context);
             Peer peer = createPeer(peerConncetionInfo.getPeerAddress());
             PeerConncetionInfo.ConnectionAttempt connectionAttempt = peerConncetionInfo.newConnectionAttempt(peer);
             long ts = System.currentTimeMillis();
             connectionAttempt.setConnectionStartedTs(ts);
-            try {
+
-                peer.addConnectedEventListener((peer1, peerCount) -> {
+            peerConnectedEventListener = Optional.of((p, peerCount) -> {
-                    connectionAttempt.setDurationUntilConnection(System.currentTimeMillis() - ts);
+                peerConnectedEventListener.ifPresent(peer::removeConnectedEventListener);
-                    connectionAttempt.setConnectionSuccessTs(System.currentTimeMillis());
+                peerConnectedEventListener = Optional.empty();
+                if (shutdownCalled.get()) {
+                    return;
+                }
+                try {
+                    log.info("\n## Successfully connected to {}", peer.getAddress());
+                    long now = System.currentTimeMillis();
+                    connectionAttempt.setDurationUntilConnection(now - ts);
+                    connectionAttempt.setConnectionSuccessTs(now);
                     connectionAttempt.onConnected();
-                    startAutoDisconnectAndReconnect();
+
-                });
+                    try {
-                peer.addDisconnectedEventListener((peer1, peerCount) -> {
+                        Thread.sleep(disconnectIntervalSec * 1000L); // 2 sec
+                    } catch (InterruptedException ignore) {
+                    }
+                    if (shutdownCalled.get()) {
+                        return;
+                    }
+                    log.info("Close peer {}", peer.getAddress());
+                    peer.close();
+                } catch (Exception exception) {
+                    log.warn("Exception at onConnection handler. {}", exception.toString());
+                    handleException(exception, peer, connectionAttempt, ts, future);
+                }
+            });
+            peer.addConnectedEventListener(onConnectionExecutor, peerConnectedEventListener.get());
+
+            peerDisconnectedEventListener = Optional.of((p, peerCount) -> {
+                // At socket timeouts we get called twice from Bitcoinj
+                if (peerDisconnectedEventListener.isEmpty()) {
+                    log.error("We got called twice at socketimeout from BitcoinJ and ignore the 2nd call.");
+                    return;
+                }
+                peerDisconnectedEventListener.ifPresent(peer::removeDisconnectedEventListener);
+                peerDisconnectedEventListener = Optional.empty();
+                if (shutdownCalled.get()) {
+                    return;
+                }
+                if (openConnectionFuture.isPresent() && !openConnectionFuture.get().isDone()) {
+                    // BitcoinJ calls onDisconnect at sockettimeout without throwing an error on the open connection future.
+                    openConnectionFuture.get().completeExceptionally(new TimeoutException("Open connection failed due timeout at PeerSocketHandler"));
+                    return;
+                }
+                try {
+                    log.info("\n<< Disconnected from {}", peer.getAddress());
                     long passed = System.currentTimeMillis() - ts;
                     // Timeout is not handled as error in bitcoinj, but it simply disconnects
                     // If we had a successful connect before we got versionMessage set, otherwise its from an error.
@@ -113,52 +194,62 @@ public class PeerConnection {
                         connectionAttempt.setDurationUntilDisConnection(passed);
                         connectionAttempt.onDisconnected();
                     }
-                    startAutoDisconnectAndReconnect();
+                    try {
-                });
+                        Thread.sleep(reconnectIntervalSec * 2000L); // 120 sec
-                openConnection(peer).join();
+                    } catch (InterruptedException ignore) {
+                    }
+                    if (shutdownCalled.get()) {
+                        return;
+                    }
+                    future.complete(null);
+                } catch (Exception exception) {
+                    log.warn("Exception at onDisconnection handler. {}", exception.toString());
+                    handleException(exception, peer, connectionAttempt, ts, future);
+                }
+            });
+            peer.addDisconnectedEventListener(onDisConnectionExecutor, peerDisconnectedEventListener.get());
+
+            try {
+                openConnectionFuture = Optional.of(openConnection(peer));
+                openConnectionFuture.get().join();
             } catch (Exception exception) {
-                log.warn("Error at opening connection to peer {}", peerConncetionInfo, exception);
+                log.warn("Error at opening connection to peer {}. {}", peerConncetionInfo, exception.toString());
-                connectionAttempt.setDurationUntilFailure(System.currentTimeMillis() - ts);
+                handleException(exception, peer, connectionAttempt, ts, future);
-                connectionAttempt.onException(exception);
-                startAutoDisconnectAndReconnect();
             }
-        }, SingleThreadExecutorUtils.getSingleThreadExecutor("connect-" + peerConncetionInfo.getShortId()));
+        }, MoreExecutors.directExecutor()));
+
+        return future;
     }
 
-    private CompletableFuture<Void> disconnect() {
+    private void handleException(Throwable throwable,
-        return peerConncetionInfo.getCurrentConnectionAttempt()
+                                 Peer peer,
-                .map(currentConnectionAttempt -> CompletableFuture.runAsync(() -> {
+                                 PeerConncetionInfo.ConnectionAttempt connectionAttempt,
-                            log.info("disconnect {}", peerConncetionInfo);
+                                 long ts,
-                            Context.propagate(context);
+                                 CompletableFuture<Void> future) {
-                            currentConnectionAttempt.getPeer().close();
+        peerDisconnectedEventListener.ifPresent(peer::removeDisconnectedEventListener);
-                        },
+        peerDisconnectedEventListener = Optional.empty();
-                        SingleThreadExecutorUtils.getSingleThreadExecutor("disconnect-" + peerConncetionInfo.getShortId())))
+        peerConnectedEventListener.ifPresent(peer::removeConnectedEventListener);
-                .orElse(CompletableFuture.completedFuture(null));
+        peerConnectedEventListener = Optional.empty();
-    }
-
-    private void startAutoDisconnectAndReconnect() {
         if (shutdownCalled.get()) {
             return;
         }
-        disconnectScheduler.ifPresent(Timer::stop);
+        connectionAttempt.setDurationUntilFailure(System.currentTimeMillis() - ts);
-        disconnectScheduler = Optional.of(UserThread.runAfter(() -> {
+        connectionAttempt.onException(throwable);
-            if (shutdownCalled.get()) {
+        try {
-                return;
+            // Try disconnect
-            }
+            log.info("Try close peer {}", peer.getAddress());
-            disconnect()
+            peer.close();
-                    .thenRun(() -> {
+        } catch (Exception ignore) {
-                        if (shutdownCalled.get()) {
+        }
-                            return;
+
-                        }
+        try {
-                        reconnectScheduler.ifPresent(Timer::stop);
+            Thread.sleep(reconnectIntervalSec * 1000L); // 120 sec
-                        reconnectScheduler = Optional.of(UserThread.runAfter(() -> {
+        } catch (InterruptedException ignore) {
-                            if (shutdownCalled.get()) {
+        }
-                                return;
+        if (shutdownCalled.get()) {
-                            }
+            return;
-                            connect();
+        }
-                        }, reconnectIntervalSec));
+        future.completeExceptionally(throwable);
-                    });
-        }, disconnectIntervalSec));
     }
 
     private Peer createPeer(PeerAddress address) {

btcnodemonitor/src/main/java/bisq/btcnodemonitor/btc/PeerGroupService.java

@@ -119,21 +119,24 @@ public class PeerGroupService {
             Context.propagate(context);
             blockingClientManager.startAsync();
             blockingClientManager.awaitRunning();
-        }, SingleThreadExecutorUtils.getSingleThreadExecutor("start"));
-    }
 
-    public void connectToAll() {
+            peerConnections.forEach(PeerConnection::start);
-        peerConnections.forEach(PeerConnection::connect);
+        }, SingleThreadExecutorUtils.getSingleThreadExecutor("start"));
     }
 
     public CompletableFuture<Void> shutdown() {
         return CompletableFuture.runAsync(() -> {
-            log.info("shutdown");
+            log.info("Shutdown all peerConnections");
             Context.propagate(context);
             CountDownLatch latch = new CountDownLatch(peerConnections.size());
-            peerConnections.forEach(e -> e.shutdown().thenRun(latch::countDown));
+            peerConnections.forEach(peerConnection -> peerConnection.shutdown()
+                    .thenRun(latch::countDown));
             try {
-                latch.await(5, TimeUnit.SECONDS);
+                if (latch.await(3, TimeUnit.SECONDS)) {
+                    log.info("All peerConnections shut down.");
+                } else {
+                    log.info("Shutdown of peerConnections not completed in time.");
+                }
                 blockingClientManager.stopAsync();
                 blockingClientManager.awaitTerminated(Duration.ofSeconds(2));
             } catch (Exception e) {

btcnodemonitor/src/main/java/bisq/btcnodemonitor/server/SimpleHttpServer.java

@@ -56,28 +56,24 @@ public class SimpleHttpServer {
     private final static String CLOSE_TAG = "</font><br/>";
     private final static String WARNING_ICON = "&#9888; ";
     private final static String ALERT_ICON = "&#9760; "; // &#9889;  &#9889;
-    private final Config config;
     @Getter
     private final List<BtcNodes.BtcNode> providedBtcNodes;
     private final Map<String, BtcNodes.BtcNode> btcNodeByAddress;
     private final int port;
     private final PeerConncetionModel peerConncetionModel;
     private final String started;
-
+    private final String networkInfo;
     private String html;
-    private int requestCounter;
+
-    private String networkInfo;
 
     ///////////////////////////////////////////////////////////////////////////////////////////
     // Constructor
     ///////////////////////////////////////////////////////////////////////////////////////////
 
     public SimpleHttpServer(Config config, PeerConncetionModel peerConncetionModel) {
-        this.config = config;
         this.peerConncetionModel = peerConncetionModel;
         started = new Date().toString();
-
+        providedBtcNodes = peerConncetionModel.getProvidedBtcNodes();
-        this.providedBtcNodes = peerConncetionModel.getProvidedBtcNodes();
 
         BaseCurrencyNetwork network = config.baseCurrencyNetwork;
         if (config.useTorForBtcMonitor) {
@@ -121,9 +117,9 @@ public class SimpleHttpServer {
 
     public CompletableFuture<Void> shutdown() {
         return CompletableFuture.runAsync(() -> {
-            log.info("shutDown");
+            log.info("stop Spark server");
             Spark.stop();
-            log.info("shutDown completed");
+            log.info("Spark server stopped");
         });
     }
 

btcnodemonitor/src/main/java/bisq/btcnodemonitor/socksProxy/ProxySetup.java

@@ -81,12 +81,12 @@ public class ProxySetup {
     }
 
     public CompletableFuture<Void> shutdown() {
-        log.warn("start shutdown");
+        log.info("Shutdown tor");
         return CompletableFuture.runAsync(() -> {
                     if (tor != null) {
                         tor.shutdown();
+                        log.info("Tor shutdown completed");
                     }
-                    log.warn("shutdown completed");
                 })
                 .orTimeout(2, TimeUnit.SECONDS);
     }