Update all statistic changes

gui/src/main/java/io/bitsquare/gui/main/settings/network/NetworkSettingsView.java

@@ -26,13 +26,11 @@ import io.bitsquare.gui.common.view.ActivatableViewAndModel;
 import io.bitsquare.gui.common.view.FxmlView;
 import io.bitsquare.gui.main.popups.Popup;
 import io.bitsquare.gui.util.BSFormatter;
-import io.bitsquare.p2p.NodeAddress;
 import io.bitsquare.p2p.P2PService;
-import io.bitsquare.p2p.P2PServiceListener;
 import io.bitsquare.p2p.network.Statistic;
 import io.bitsquare.user.Preferences;
-import javafx.beans.value.ChangeListener;
 import javafx.collections.FXCollections;
+import javafx.collections.ObservableList;
 import javafx.fxml.FXML;
 import javafx.geometry.Insets;
 import javafx.geometry.VPos;
@@ -41,6 +39,7 @@ import javafx.scene.layout.GridPane;
 import javafx.util.StringConverter;
 import org.bitcoinj.core.Peer;
 import org.fxmisc.easybind.EasyBind;
+import org.fxmisc.easybind.Subscription;
 import org.reactfx.util.FxTimer;
 
 import javax.inject.Inject;
@@ -69,16 +68,17 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
     @FXML
     CheckBox useTorCheckBox;
     @FXML
-    TableView<NetworkStatisticListItem> p2PPeerTable;
+    TableView<P2pNetworkListItem> p2PPeerTable;
     @FXML
-    TableColumn<NetworkStatisticListItem, String> onionAddressColumn, connectionTypeColumn, creationDateColumn,
+    TableColumn<P2pNetworkListItem, String> onionAddressColumn, connectionTypeColumn, creationDateColumn,
             lastActivityColumn, sentBytesColumn, receivedBytesColumn, peerTypeColumn;
     /* TableColumn<NetworkStatisticListItem, NetworkStatisticListItem> onionAddressColumn, connectionTypeColumn, creationDateColumn,
              lastActivityColumn, sentBytesColumn, receivedBytesColumn, peerTypeColumn;
  */
-    private P2PServiceListener p2PServiceListener;
+    private Subscription numP2PPeersSubscription;
-    private ChangeListener<Number> numP2PPeersChangeListener;
+    private Subscription bitcoinPeersSubscription;
-    private ChangeListener<List<Peer>> bitcoinPeersChangeListener;
+    private Subscription nodeAddressSubscription;
+    private ObservableList<P2pNetworkListItem> networkListItems = FXCollections.observableArrayList();
 
     @Inject
     public NetworkSettingsView(WalletService walletService, P2PService p2PService, Preferences preferences, Clock clock,
@@ -111,36 +111,6 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
                 return null;
             }
         });
-        p2PServiceListener = new P2PServiceListener() {
-            @Override
-            public void onHiddenServicePublished() {
-                onionAddress.setText(p2PService.getAddress().getFullAddress());
-            }
-
-            @Override
-            public void onRequestingDataCompleted() {
-            }
-
-            @Override
-            public void onNoSeedNodeAvailable() {
-            }
-
-            @Override
-            public void onNoPeersAvailable() {
-            }
-
-            @Override
-            public void onBootstrapComplete() {
-            }
-
-            @Override
-            public void onTorNodeReady() {
-            }
-
-            @Override
-            public void onSetupFailed(Throwable throwable) {
-            }
-        };
 
         p2PPeerTable.setMinHeight(300);
         p2PPeerTable.setColumnResizePolicy(TableView.CONSTRAINED_RESIZE_POLICY);
@@ -156,7 +126,6 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
                 ((Integer) o1.statistic.getSentBytes()).compareTo(((Integer) o2.statistic.getSentBytes())));
         receivedBytesColumn.setComparator((o1, o2) ->
                 ((Integer) o1.statistic.getReceivedBytes()).compareTo(((Integer) o2.statistic.getReceivedBytes())));*/
-
     }
 
     @Override
@@ -175,54 +144,43 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
                         .closeButtonText("Cancel")
                         .onClose(() -> useTorCheckBox.setSelected(!selected))
                         .show();
-
             }
         });
+        bitcoinPeersSubscription = EasyBind.subscribe(walletService.connectedPeersProperty(), connectedPeers -> updateBitcoinPeersTextArea());
 
-        NodeAddress nodeAddress = p2PService.getAddress();
+        nodeAddressSubscription = EasyBind.subscribe(p2PService.getNetworkNode().nodeAddressProperty(),
-        if (nodeAddress == null) {
+                nodeAddress -> onionAddress.setText(nodeAddress == null ? "Not known yet..." : p2PService.getAddress().getFullAddress()));
-            p2PService.addP2PServiceListener(p2PServiceListener);
+        numP2PPeersSubscription = EasyBind.subscribe(p2PService.getNumConnectedPeers(), numPeers -> updateP2PTable());
-        } else {
-            onionAddress.setText(nodeAddress.getFullAddress());
-        }
-
-        bitcoinPeersChangeListener = (observable, oldValue, newValue) -> updateBitcoinPeersTextArea();
-        walletService.connectedPeersProperty().addListener(bitcoinPeersChangeListener);
-        updateBitcoinPeersTextArea();
-
-        numP2PPeersChangeListener = (observable, oldValue, newValue) -> updateP2PStatistics();
-        p2PService.getNumConnectedPeers().addListener(numP2PPeersChangeListener);
-        updateP2PStatistics();
-
         totalTraffic.textProperty().bind(EasyBind.combine(Statistic.totalSentBytesProperty(), Statistic.totalReceivedBytesProperty(),
                 (sent, received) -> "Sent: " + formatter.formatBytes((int) sent) + ", received: " + formatter.formatBytes((int) received)));
+
+        p2PPeerTable.setItems(networkListItems);
+        p2PPeerTable.sort();
     }
 
     @Override
     public void deactivate() {
         useTorCheckBox.setOnAction(null);
 
-        if (p2PServiceListener != null)
+        if (nodeAddressSubscription != null)
-            p2PService.removeP2PServiceListener(p2PServiceListener);
+            nodeAddressSubscription.unsubscribe();
 
-        if (bitcoinPeersChangeListener != null)
+        if (bitcoinPeersSubscription != null)
-            walletService.connectedPeersProperty().removeListener(bitcoinPeersChangeListener);
+            bitcoinPeersSubscription.unsubscribe();
 
-        if (numP2PPeersChangeListener != null)
+        if (numP2PPeersSubscription != null)
-            p2PService.getNumConnectedPeers().removeListener(numP2PPeersChangeListener);
+            numP2PPeersSubscription.unsubscribe();
 
-        p2PPeerTable.getItems().forEach(NetworkStatisticListItem::cleanup);
+        p2PPeerTable.getItems().forEach(P2pNetworkListItem::cleanup);
         totalTraffic.textProperty().unbind();
     }
 
-    private void updateP2PStatistics() {
+    private void updateP2PTable() {
-        p2PPeerTable.getItems().forEach(NetworkStatisticListItem::cleanup);
+        p2PPeerTable.getItems().forEach(P2pNetworkListItem::cleanup);
-
+        networkListItems.clear();
-        List<NetworkStatisticListItem> list = p2PService.getNetworkNode().getConfirmedConnections().stream()
+        networkListItems.setAll(p2PService.getNetworkNode().getAllConnections().stream()
-                .map(connection -> new NetworkStatisticListItem(connection, clock, formatter))
+                .map(connection -> new P2pNetworkListItem(connection, clock, formatter))
-                .collect(Collectors.toList());
+                .collect(Collectors.toList()));
-        p2PPeerTable.setItems(FXCollections.observableArrayList(list));
-        p2PPeerTable.sort();
     }
 
     private void updateBitcoinPeersTextArea() {

network/src/main/java/io/bitsquare/p2p/network/NetworkNode.java

@@ -6,6 +6,9 @@ import io.bitsquare.common.UserThread;
 import io.bitsquare.common.util.Utilities;
 import io.bitsquare.p2p.Message;
 import io.bitsquare.p2p.NodeAddress;
+import javafx.beans.property.ObjectProperty;
+import javafx.beans.property.ReadOnlyObjectProperty;
+import javafx.beans.property.SimpleObjectProperty;
 import org.apache.commons.lang3.StringUtils;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -26,7 +29,7 @@ import java.util.stream.Collectors;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 // Run in UserThread
-public abstract class NetworkNode implements MessageListener, ConnectionListener {
+public abstract class NetworkNode implements MessageListener {
     private static final Logger log = LoggerFactory.getLogger(NetworkNode.class);
     private static final int CREATE_SOCKET_TIMEOUT_MILLIS = 5000;
 
@@ -42,6 +45,7 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
     private volatile boolean shutDownInProgress;
     // accessed from different threads
     private final CopyOnWriteArraySet<OutboundConnection> outBoundConnections = new CopyOnWriteArraySet<>();
+    protected final ObjectProperty<NodeAddress> nodeAddressProperty = new SimpleObjectProperty<>();
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -108,8 +112,30 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
                         existingConnection.sendMessage(message);
                         return existingConnection;
                     } else {
-                        outboundConnection = new OutboundConnection(socket, NetworkNode.this, NetworkNode.this, peersNodeAddress);
+                        outboundConnection = new OutboundConnection(socket,
-                        outBoundConnections.add(outboundConnection);
+                                NetworkNode.this,
+                                new ConnectionListener() {
+                                    @Override
+                                    public void onConnection(Connection connection) {
+                                        outBoundConnections.add((OutboundConnection) connection);
+                                        printOutBoundConnections();
+                                        connectionListeners.stream().forEach(e -> e.onConnection(connection));
+                                    }
+
+                                    @Override
+                                    public void onDisconnect(CloseConnectionReason closeConnectionReason, Connection connection) {
+                                        log.trace("onDisconnect connectionListener\n\tconnection={}" + connection);
+                                        printOutBoundConnections();
+                                        outBoundConnections.remove(connection);
+
+                                        connectionListeners.stream().forEach(e -> e.onDisconnect(closeConnectionReason, connection));
+                                    }
+
+                                    @Override
+                                    public void onError(Throwable throwable) {
+                                        connectionListeners.stream().forEach(e -> e.onError(throwable));
+                                    }
+                                }, peersNodeAddress);
 
                         log.info("\n\n%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%\n" +
                                 "NetworkNode created new outbound connection:"
@@ -204,6 +230,10 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
         return resultFuture;
     }
 
+    public ReadOnlyObjectProperty<NodeAddress> nodeAddressProperty() {
+        return nodeAddressProperty;
+    }
+
     public Set<Connection> getAllConnections() {
         // Can contain inbound and outbound connections with the same peer node address, 
         // as connection hashcode is using uid and port info
@@ -255,33 +285,6 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
     }
 
 
-    ///////////////////////////////////////////////////////////////////////////////////////////
-    // ConnectionListener implementation
-    ///////////////////////////////////////////////////////////////////////////////////////////
-
-    @Override
-    public void onConnection(Connection connection) {
-        connectionListeners.stream().forEach(e -> e.onConnection(connection));
-    }
-
-    @Override
-    public void onDisconnect(CloseConnectionReason closeConnectionReason, Connection connection) {
-        log.trace("onDisconnect connectionListener\n\tconnection={}" + connection);
-        if (inBoundConnections.contains(connection))
-            log.warn("We have the connection in our inBoundConnections. That must not happen as it should be called " +
-                    "from the server listener and get removed from there.");
-        printOutBoundConnections();
-        outBoundConnections.remove(connection);
-        // inbound connections are removed in the listener of the server
-        connectionListeners.stream().forEach(e -> e.onDisconnect(closeConnectionReason, connection));
-    }
-
-    @Override
-    public void onError(Throwable throwable) {
-        connectionListeners.stream().forEach(e -> e.onError(throwable));
-    }
-
-
     ///////////////////////////////////////////////////////////////////////////////////////////
     // MessageListener implementation
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -329,15 +332,18 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
     ///////////////////////////////////////////////////////////////////////////////////////////
 
     void createExecutorService() {
-        executorService = Utilities.getListeningExecutorService("NetworkNode-" + servicePort, 20, 50, 120L);
+        executorService = Utilities.getListeningExecutorService("NetworkNode-" + servicePort, 50, 100, 2 * 60);
     }
 
     void startServer(ServerSocket serverSocket) {
-        ConnectionListener connectionListener = new ConnectionListener() {
+        server = new Server(serverSocket,
+                NetworkNode.this,
+                new ConnectionListener() {
                     @Override
                     public void onConnection(Connection connection) {
                         inBoundConnections.add((InboundConnection) connection);
-                NetworkNode.this.onConnection(connection);
+                        printInboundConnections();
+                        connectionListeners.stream().forEach(e -> e.onConnection(connection));
                     }
 
                     @Override
@@ -345,17 +351,14 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
                         log.trace("onDisconnect at server socket connectionListener\n\tconnection={}" + connection);
                         inBoundConnections.remove(connection);
                         printInboundConnections();
-                NetworkNode.this.onDisconnect(closeConnectionReason, connection);
+                        connectionListeners.stream().forEach(e -> e.onDisconnect(closeConnectionReason, connection));
                     }
 
                     @Override
                     public void onError(Throwable throwable) {
-                NetworkNode.this.onError(throwable);
+                        connectionListeners.stream().forEach(e -> e.onError(throwable));
                     }
-        };
+                });
-        server = new Server(serverSocket,
-                NetworkNode.this,
-                connectionListener);
         executorService.submit(server);
     }
 
@@ -392,5 +395,7 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
     abstract protected Socket createSocket(NodeAddress peersNodeAddress) throws IOException;
 
     @Nullable
-    abstract public NodeAddress getNodeAddress();
+    public NodeAddress getNodeAddress() {
+        return nodeAddressProperty.get();
+    }
 }