[enh] Allow to close() when listening network peers

duniter4j-core-client/src/main/java/org/duniter/core/client/service/local/NetworkService.java

@@ -25,6 +25,7 @@ package org.duniter.core.client.service.local;
 import org.duniter.core.beans.Service;
 import org.duniter.core.client.model.local.Peer;
 
+import java.io.Closeable;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.List;
@@ -86,9 +87,9 @@ public interface NetworkService extends Service {
 
     Comparator<Peer> peerComparator(Sort sort);
 
-    void addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener);
+    Closeable addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener);
 
-    void addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener,
+    Closeable addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener,
                                 final Filter filter, final Sort sort, final boolean autoreconnect,
                                 final ExecutorService executor);
 

duniter4j-core-client/src/main/java/org/duniter/core/client/service/local/NetworkServiceImpl.java

@@ -43,9 +43,11 @@ import org.duniter.core.util.*;
 import org.duniter.core.util.CollectionUtils;
 import org.duniter.core.util.concurrent.CompletableFutures;
 import org.duniter.core.util.http.InetAddressUtils;
+import org.duniter.core.util.websocket.WebsocketClientEndpoint;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.*;
@@ -365,7 +367,7 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
         return peers;
     }
 
-    public void addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener) {
+    public Closeable addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener) {
 
         BlockchainParameters parameters = blockchainRemoteService.getParameters(mainPeer);
         fillCurrentBlock(mainPeer);
@@ -386,13 +388,13 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
         Sort sortDef = new Sort();
         sortDef.sortType = null;
 
-        addPeersChangeListener(mainPeer, listener, filterDef, sortDef, true, null);
+        return addPeersChangeListener(mainPeer, listener, filterDef, sortDef, true, null);
 
     }
 
-    public void addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener,
-                                       final Filter filter, final Sort sort, final boolean autoreconnect,
-                                       final ExecutorService executor) {
+    public Closeable addPeersChangeListener(final Peer mainPeer, final PeersChangeListener listener,
+                                            final Filter filter, final Sort sort, final boolean autoreconnect,
+                                            final ExecutorService executor) {
 
         final String currency = filter != null && filter.currency != null ? filter.currency :
                 blockchainRemoteService.getParameters(mainPeer).getCurrency();
@@ -507,7 +509,7 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
         };
 
         // Manage new block event
-        blockchainRemoteService.addBlockListener(mainPeer, json -> {
+        WebsocketClientEndpoint.MessageListener blockListener = json -> {
             log.debug("Received new block event");
             try {
                 BlockchainBlock block = readValue(json, BlockchainBlock.class);
@@ -522,10 +524,11 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
             } catch(IOException e) {
                 log.error("Could not parse peer received by WS: " + e.getMessage(), e);
             }
-        }, autoreconnect);
+        };
+        WebsocketClientEndpoint wsBlockEndpoint = blockchainRemoteService.addBlockListener(mainPeer, blockListener, autoreconnect);
 
         // Manage new peer event
-        networkRemoteService.addPeerListener(mainPeer, json -> {
+        WebsocketClientEndpoint.MessageListener peerListsner = json -> {
 
             log.debug("Received new peer event");
             try {
@@ -536,11 +539,17 @@ public class NetworkServiceImpl extends BaseRemoteServiceImpl implements Network
             } catch(IOException e) {
                 log.error("Could not parse peer received by WS: " + e.getMessage(), e);
             }
-        }, autoreconnect);
+        };
+        WebsocketClientEndpoint wsPeerEndpoint = networkRemoteService.addPeerListener(mainPeer, peerListsner, autoreconnect);
 
         // Default action: Load all peers
         pool.submit(loadAllPeers);
 
+        // Return the tear down logic
+        return () -> {
+            wsBlockEndpoint.unregisterListener(blockListener);
+            wsPeerEndpoint.unregisterListener(peerListsner);
+        };
     }
 
     public String getVersion(final Peer peer) {

duniter4j-core-shared/src/main/java/org/duniter/core/util/websocket/WebsocketClientEndpoint.java

@@ -148,6 +148,16 @@ public class WebsocketClientEndpoint implements Closeable {
     public void unregisterListener(MessageListener listener) {
         synchronized (messageListeners) {
             this.messageListeners.remove(listener);
+
+            // If no more listener, close the WS client
+            if (this.messageListeners.size() == 0) {
+                try {
+                    close();
+                }
+                catch(IOException e) {
+                    // Silent
+                }
+            }
         }
     }