handle connection errors and incoming notifications

4 years ago · 50ef1c1a07
5 changed files with 251 additions and 47 deletions
--- a/src/main/java/com/sparrowwallet/sparrow/AppController.java
+++ b/src/main/java/com/sparrowwallet/sparrow/AppController.java
@ -24,10 +24,7 @@ import com.sparrowwallet.sparrow.transaction.TransactionController;
 import com.sparrowwallet.sparrow.wallet.WalletController;
 import com.sparrowwallet.sparrow.wallet.WalletForm;
 import de.codecentric.centerdevice.MenuToolkit;
-import javafx.animation.Animation;
-import javafx.animation.KeyFrame;
-import javafx.animation.KeyValue;
-import javafx.animation.Timeline;
+import javafx.animation.*;
 import javafx.concurrent.Worker;
 import javafx.event.ActionEvent;
 import javafx.fxml.FXML;
@ -53,7 +50,6 @@ public class AppController implements Initializable {
    private static final int SERVER_PING_PERIOD = 10 * 1000;
    private static final int ENUMERATE_HW_PERIOD = 30 * 1000;

-    private static final String TRANSACTION_TAB_TYPE = "transaction";
    public static final String DRAG_OVER_CLASS = "drag-over";

    @FXML
@ -77,9 +73,12 @@ public class AppController implements Initializable {
    @FXML
    private UnlabeledToggleSwitch serverToggle;

+    //Determines if a change in serverToggle changes the offline/online mode
+    private boolean changeMode = true;
+
    private Timeline statusTimeline;

-    private ElectrumServer.PingService pingService;
+    private ElectrumServer.ConnectionService connectionService;

    public static boolean showTxHexProperty;

@ -89,7 +88,7 @@ public class AppController implements Initializable {
    }

    void initializeView() {
-        setOsxApplicationMenu();
+        //setOsxApplicationMenu();

        rootStack.setOnDragOver(event -> {
            if(event.getGestureSource() != rootStack && event.getDragboard().hasFiles()) {
@ -140,44 +139,52 @@ public class AppController implements Initializable {
        exportWallet.setDisable(true);

        serverToggle.selectedProperty().addListener((observable, oldValue, newValue) -> {
-            Config.get().setMode(newValue ? Mode.ONLINE : Mode.OFFLINE);
-            if(newValue) {
-                if(pingService.getState() == Worker.State.CANCELLED) {
-                    pingService.reset();
-                }
+            if(changeMode) {
+                Config.get().setMode(newValue ? Mode.ONLINE : Mode.OFFLINE);
+                if(newValue) {
+                    if(connectionService.getState() == Worker.State.CANCELLED) {
+                        connectionService.reset();
+                    }

-                if(!pingService.isRunning()) {
-                    pingService.start();
+                    if(!connectionService.isRunning()) {
+                        connectionService.start();
+                    }
+                } else {
+                    connectionService.cancel();
                }
-            } else {
-                pingService.cancel();
            }
        });

-        pingService = createPingService();
+        connectionService = createConnectionService();
        Config config = Config.get();
        if(config.getMode() == Mode.ONLINE && config.getElectrumServer() != null && !config.getElectrumServer().isEmpty()) {
-            pingService.start();
+            connectionService.start();
        }
    }

-    private ElectrumServer.PingService createPingService() {
-        ElectrumServer.PingService pingService = new ElectrumServer.PingService();
-        pingService.setPeriod(new Duration(SERVER_PING_PERIOD));
-        pingService.setOnSucceeded(successEvent -> {
+    private ElectrumServer.ConnectionService createConnectionService() {
+        ElectrumServer.ConnectionService connectionService = new ElectrumServer.ConnectionService();
+        connectionService.setPeriod(new Duration(SERVER_PING_PERIOD));
+        connectionService.setRestartOnFailure(true);
+
+        connectionService.setOnSucceeded(successEvent -> {
+            changeMode = false;
            serverToggle.setSelected(true);
-            if(pingService.getValue() != null) {
-                statusBar.setText("Connected: " + pingService.getValue().split(System.lineSeparator(), 2)[0]);
-            } else {
-                statusBar.setText("");
+            changeMode = true;
+
+            if(connectionService.getValue() != null) {
+                EventManager.get().post(connectionService.getValue());
            }
        });
-        pingService.setOnFailed(failEvent -> {
+        connectionService.setOnFailed(failEvent -> {
+            changeMode = false;
            serverToggle.setSelected(false);
-            statusBar.setText(failEvent.getSource().getException().getMessage());
+            changeMode = true;
+
+            EventManager.get().post(new ConnectionFailedEvent(failEvent.getSource().getException()));
        });

-        return pingService;
+        return connectionService;
    }

    private void setOsxApplicationMenu() {
@ -594,6 +601,19 @@ public class AppController implements Initializable {
        exportWallet.setDisable(!event.getWallet().isValid());
    }

+    @Subscribe
+    public void statusUpdated(StatusEvent event) {
+        statusBar.setText(event.getStatus());
+
+        PauseTransition wait = new PauseTransition(Duration.seconds(10));
+        wait.setOnFinished((e) -> {
+            if(statusBar.getText().equals(event.getStatus())) {
+                statusBar.setText("");
+            }
+        });
+        wait.play();
+    }
+
    @Subscribe
    public void timedWorker(TimedEvent event) {
        if(event.getTimeMills() == 0) {
@ -643,4 +663,18 @@ public class AppController implements Initializable {
            usbStatus.setDevices(event.getDevices());
        }
    }
+
+    @Subscribe
+    public void newConnection(ConnectionEvent event) {
+        String banner = event.getServerBanner();
+        String status = "Connected: " + (banner == null ? "Server" : banner.split(System.lineSeparator(), 2)[0]) + " at height " + event.getBlockHeight();
+        EventManager.get().post(new StatusEvent(status));
+    }
+
+    @Subscribe
+    public void connectionFailed(ConnectionFailedEvent event) {
+        String reason = event.getException().getCause() != null ? event.getException().getCause().getMessage() : event.getException().getMessage();
+        String status = "Connection error: " + reason;
+        EventManager.get().post(new StatusEvent(status));
+    }
 }
--- a/src/main/java/com/sparrowwallet/sparrow/event/ConnectionEvent.java
+++ b/src/main/java/com/sparrowwallet/sparrow/event/ConnectionEvent.java
@ -0,0 +1,35 @@
+package com.sparrowwallet.sparrow.event;
+
+import com.sparrowwallet.drongo.protocol.BlockHeader;
+
+import java.util.List;
+
+public class ConnectionEvent {
+    private final List<String> serverVersion;
+    private final String serverBanner;
+    private final int blockHeight;
+    private final BlockHeader blockHeader;
+
+    public ConnectionEvent(List<String> serverVersion, String serverBanner, int blockHeight, BlockHeader blockHeader) {
+        this.serverVersion = serverVersion;
+        this.serverBanner = serverBanner;
+        this.blockHeight = blockHeight;
+        this.blockHeader = blockHeader;
+    }
+
+    public List<String> getServerVersion() {
+        return serverVersion;
+    }
+
+    public String getServerBanner() {
+        return serverBanner;
+    }
+
+    public int getBlockHeight() {
+        return blockHeight;
+    }
+
+    public BlockHeader getBlockHeader() {
+        return blockHeader;
+    }
+}
--- a/src/main/java/com/sparrowwallet/sparrow/event/ConnectionFailedEvent.java
+++ b/src/main/java/com/sparrowwallet/sparrow/event/ConnectionFailedEvent.java
@ -0,0 +1,13 @@
+package com.sparrowwallet.sparrow.event;
+
+public class ConnectionFailedEvent {
+    private final Throwable exception;
+
+    public ConnectionFailedEvent(Throwable exception) {
+        this.exception = exception;
+    }
+
+    public Throwable getException() {
+        return exception;
+    }
+}
--- a/src/main/java/com/sparrowwallet/sparrow/event/StatusEvent.java
+++ b/src/main/java/com/sparrowwallet/sparrow/event/StatusEvent.java
@ -0,0 +1,13 @@
+package com.sparrowwallet.sparrow.event;
+
+public class StatusEvent {
+    private final String status;
+
+    public StatusEvent(String status) {
+        this.status = status;
+    }
+
+    public String getStatus() {
+        return status;
+    }
+}
--- a/src/main/java/com/sparrowwallet/sparrow/io/ElectrumServer.java
+++ b/src/main/java/com/sparrowwallet/sparrow/io/ElectrumServer.java
@ -7,6 +7,7 @@ import com.sparrowwallet.drongo.KeyPurpose;
 import com.sparrowwallet.drongo.Utils;
 import com.sparrowwallet.drongo.protocol.*;
 import com.sparrowwallet.drongo.wallet.*;
+import com.sparrowwallet.sparrow.event.ConnectionEvent;
 import javafx.concurrent.ScheduledService;
 import javafx.concurrent.Service;
 import javafx.concurrent.Task;
@ -24,6 +25,7 @@ import java.security.cert.CertificateException;
 import java.security.cert.CertificateFactory;
 import java.security.cert.X509Certificate;
 import java.util.*;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;

 public class ElectrumServer {
@ -75,6 +77,11 @@ public class ElectrumServer {
        return transport;
    }

+    public void connect() throws ServerException {
+        TcpTransport tcpTransport = (TcpTransport)getTransport();
+        tcpTransport.connect();
+    }
+
    public void ping() throws ServerException {
        JsonRpcClient client = new JsonRpcClient(getTransport());
        client.createRequest().method("server.ping").id(1).executeNullable();
@ -90,6 +97,11 @@ public class ElectrumServer {
        return client.createRequest().returnAs(String.class).method("server.banner").id(1).execute();
    }

+    public BlockHeaderTip subscribeBlockHeaders() throws ServerException {
+        JsonRpcClient client = new JsonRpcClient(getTransport());
+        return client.createRequest().returnAs(BlockHeaderTip.class).method("blockchain.headers.subscribe").id(1).execute();
+    }
+
    public static synchronized void closeActiveConnection() throws ServerException {
        try {
            if(transport != null) {
@ -368,6 +380,16 @@ public class ElectrumServer {
        }
    }

+    private static class BlockHeaderTip {
+        public int height;
+        public String hex;
+
+        public BlockHeader getBlockHeader() {
+            byte[] blockHeaderBytes = Utils.hexToBytes(hex);
+            return new BlockHeader(blockHeaderBytes);
+        }
+    }
+
    public static class TcpTransport implements Transport, Closeable {
        public static final int DEFAULT_PORT = 50001;

@ -376,6 +398,12 @@ public class ElectrumServer {

        private Socket socket;

+        private String response;
+
+        private final ReentrantLock clientRequestLock = new ReentrantLock();
+        private boolean running = false;
+        private boolean reading = true;
+
        public TcpTransport(HostAndPort server) {
            this.server = server;
            this.socketFactory = SocketFactory.getDefault();
@ -383,27 +411,62 @@ public class ElectrumServer {

        @Override
        public @NotNull String pass(@NotNull String request) throws IOException {
-            if(socket == null) {
-                socket = createSocket();
-            }
-
+            clientRequestLock.lock();
            try {
-                writeRequest(socket, request);
-            } catch (IOException e) {
-                socket = createSocket();
-                writeRequest(socket, request);
+                writeRequest(request);
+                return readResponse();
+            } finally {
+                clientRequestLock.unlock();
            }
-
-            return readResponse(socket);
        }

-        private void writeRequest(Socket socket, String request) throws IOException {
+        private void writeRequest(String request) throws IOException {
            PrintWriter out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())));
            out.println(request);
            out.flush();
        }

-        private String readResponse(Socket socket) throws IOException {
+        private synchronized String readResponse() {
+            while(reading) {
+                try {
+                    wait();
+                } catch (InterruptedException e) {
+                    //Restore interrupt status and continue
+                    Thread.currentThread().interrupt();
+                }
+            }
+
+            reading = true;
+
+            notifyAll();
+            return response;
+        }
+
+        public synchronized void readInputLoop() throws ServerException {
+            while(running) {
+                try {
+                    String received = readInputStream();
+                    if(received.contains("method")) {
+                        //Handle notification
+                        System.out.println("Notification: " + received);
+                    } else {
+                        response = received;
+                        reading = false;
+                        notifyAll();
+                        wait();
+                    }
+                } catch(InterruptedException e) {
+                    //Restore interrupt status and continue
+                    Thread.currentThread().interrupt();
+                } catch(IOException e) {
+                    if(running) {
+                        throw new ServerException(e);
+                    }
+                }
+            }
+        }
+
+        protected String readInputStream() throws IOException {
            BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String response = in.readLine();

@ -414,6 +477,15 @@ public class ElectrumServer {
            return response;
        }

+        public void connect() throws ServerException {
+            try {
+                socket = createSocket();
+                running = true;
+            } catch (IOException e) {
+                throw new ServerException(e);
+            }
+        }
+
        protected Socket createSocket() throws IOException {
            return socketFactory.createSocket(server.getHost(), server.getPortOrDefault(DEFAULT_PORT));
        }
@ -421,6 +493,7 @@ public class ElectrumServer {
        @Override
        public void close() throws IOException {
            if(socket != null) {
+                running = false;
                socket.close();
            }
        }
@ -527,20 +600,38 @@ public class ElectrumServer {
        }
    }

-    public static class PingService extends ScheduledService<String> {
+    public static class ConnectionService extends ScheduledService<ConnectionEvent> implements Thread.UncaughtExceptionHandler {
        private boolean firstCall = true;
+        private Thread reader;
+        private Throwable lastReaderException;

        @Override
-        protected Task<String> createTask() {
+        protected Task<ConnectionEvent> createTask() {
            return new Task<>() {
-                protected String call() throws ServerException {
+                protected ConnectionEvent call() throws ServerException {
                    ElectrumServer electrumServer = new ElectrumServer();
                    if(firstCall) {
-                        electrumServer.getServerVersion();
+                        electrumServer.connect();
+
+                        reader = new Thread(new ReadRunnable());
+                        reader.setDaemon(true);
+                        reader.setUncaughtExceptionHandler(ConnectionService.this);
+                        reader.start();
+
+                        List<String> serverVersion = electrumServer.getServerVersion();
                        firstCall = false;
-                        return electrumServer.getServerBanner();
+
+                        BlockHeaderTip tip = electrumServer.subscribeBlockHeaders();
+                        String banner = electrumServer.getServerBanner();
+
+                        return new ConnectionEvent(serverVersion, banner, tip.height, tip.getBlockHeader());
                    } else {
-                        electrumServer.ping();
+                        if(reader.isAlive()) {
+                            electrumServer.ping();
+                        } else {
+                            firstCall = true;
+                            throw new ServerException("Connection to server failed", lastReaderException);
+                        }
                    }

                    return null;
@ -563,6 +654,24 @@ public class ElectrumServer {
        public void reset() {
            super.reset();
            firstCall = true;
+            lastReaderException = null;
+        }
+
+        @Override
+        public void uncaughtException(Thread t, Throwable e) {
+            this.lastReaderException = e;
+        }
+    }
+
+    public static class ReadRunnable implements Runnable {
+        @Override
+        public void run() {
+            try {
+                TcpTransport tcpTransport = (TcpTransport)getTransport();
+                tcpTransport.readInputLoop();
+            } catch (ServerException e) {
+                throw new RuntimeException(e.getCause() != null ? e.getCause() : e);
+            }
        }
    }