add P2PNetworkReadyListener

core/src/main/java/io/bitsquare/arbitration/ArbitratorManager.java

@@ -24,8 +24,8 @@ import io.bitsquare.common.crypto.KeyRing;
 import io.bitsquare.common.handlers.ErrorMessageHandler;
 import io.bitsquare.common.handlers.ResultHandler;
 import io.bitsquare.p2p.Address;
+import io.bitsquare.p2p.P2PNetworkReadyListener;
 import io.bitsquare.p2p.P2PService;
-import io.bitsquare.p2p.P2PServiceListener;
 import io.bitsquare.p2p.storage.HashMapChangedListener;
 import io.bitsquare.p2p.storage.data.ProtectedData;
 import io.bitsquare.user.User;
@@ -86,7 +86,7 @@ public class ArbitratorManager {
     ));
     private static final String publicKeyForTesting = "027a381b5333a56e1cc3d90d3a7d07f26509adf7029ed06fc997c656621f8da1ee";
     private final boolean isDevTest;
-    private P2PServiceListener p2PServiceListener;
+    private P2PNetworkReadyListener p2PNetworkReadyListener;
 
     @Inject
     public ArbitratorManager(@Named(ProgramArguments.DEV_TEST) boolean isDevTest, KeyRing keyRing, ArbitratorService arbitratorService, User user) {
@@ -113,29 +113,13 @@ public class ArbitratorManager {
 
             P2PService p2PService = arbitratorService.getP2PService();
             if (!p2PService.isAuthenticated()) {
-                p2PServiceListener = new P2PServiceListener() {
+                p2PNetworkReadyListener = new P2PNetworkReadyListener() {
                     @Override
-                    public void onTorNodeReady() {
+                    public void onFirstPeerAuthenticated() {
-                    }
-
-                    @Override
-                    public void onHiddenServicePublished() {
-                    }
-
-                    @Override
-                    public void onSetupFailed(Throwable throwable) {
-                    }
-
-                    @Override
-                    public void onRequestingDataCompleted() {
-                    }
-
-                    @Override
-                    public void onAuthenticated() {
                         republishArbitrator();
                     }
                 };
-                p2PService.addP2PServiceListener(p2PServiceListener);
+                p2PService.addP2PServiceListener(p2PNetworkReadyListener);
 
             } else {
                 republishArbitrator();
@@ -152,7 +136,8 @@ public class ArbitratorManager {
     }
 
     private void republishArbitrator() {
-        if (p2PServiceListener != null) arbitratorService.getP2PService().removeP2PServiceListener(p2PServiceListener);
+        if (p2PNetworkReadyListener != null)
+            arbitratorService.getP2PService().removeP2PServiceListener(p2PNetworkReadyListener);
 
         Arbitrator registeredArbitrator = user.getRegisteredArbitrator();
         if (registeredArbitrator != null) {

core/src/main/java/io/bitsquare/arbitration/DisputeManager.java

@@ -28,8 +28,8 @@ import io.bitsquare.common.crypto.KeyRing;
 import io.bitsquare.common.crypto.PubKeyRing;
 import io.bitsquare.p2p.Address;
 import io.bitsquare.p2p.Message;
+import io.bitsquare.p2p.P2PNetworkReadyListener;
 import io.bitsquare.p2p.P2PService;
-import io.bitsquare.p2p.P2PServiceListener;
 import io.bitsquare.p2p.messaging.DecryptedMsgWithPubKey;
 import io.bitsquare.p2p.messaging.SendMailboxMessageListener;
 import io.bitsquare.storage.Storage;
@@ -67,7 +67,7 @@ public class DisputeManager {
     private final DisputeList<Dispute> disputes;
     transient private final ObservableList<Dispute> disputesObservableList;
     private final String disputeInfo;
-    private final P2PServiceListener p2PServiceListener;
+    private final P2PNetworkReadyListener p2PNetworkReadyListener;
     private final CopyOnWriteArraySet<DecryptedMsgWithPubKey> decryptedMailboxMessageWithPubKeys = new CopyOnWriteArraySet<>();
     private final CopyOnWriteArraySet<DecryptedMsgWithPubKey> decryptedMailMessageWithPubKeys = new CopyOnWriteArraySet<>();
 
@@ -115,29 +115,13 @@ public class DisputeManager {
                 applyMessages();
         });
 
-        p2PServiceListener = new P2PServiceListener() {
+        p2PNetworkReadyListener = new P2PNetworkReadyListener() {
             @Override
-            public void onTorNodeReady() {
+            public void onFirstPeerAuthenticated() {
-            }
-
-            @Override
-            public void onHiddenServicePublished() {
-            }
-
-            @Override
-            public void onSetupFailed(Throwable throwable) {
-            }
-
-            @Override
-            public void onRequestingDataCompleted() {
-            }
-
-            @Override
-            public void onAuthenticated() {
                 applyMessages();
             }
         };
-        p2PService.addP2PServiceListener(p2PServiceListener);
+        p2PService.addP2PServiceListener(p2PNetworkReadyListener);
     }
 
     private void applyMessages() {
@@ -159,7 +143,7 @@ public class DisputeManager {
         });
         decryptedMailboxMessageWithPubKeys.clear();
 
-        p2PService.removeP2PServiceListener(p2PServiceListener);
+        p2PService.removeP2PServiceListener(p2PNetworkReadyListener);
     }
 
 

core/src/main/java/io/bitsquare/trade/TradeManager.java

@@ -27,8 +27,8 @@ import io.bitsquare.common.handlers.FaultHandler;
 import io.bitsquare.common.handlers.ResultHandler;
 import io.bitsquare.p2p.Address;
 import io.bitsquare.p2p.Message;
+import io.bitsquare.p2p.P2PNetworkReadyListener;
 import io.bitsquare.p2p.P2PService;
-import io.bitsquare.p2p.P2PServiceListener;
 import io.bitsquare.p2p.messaging.DecryptedMailListener;
 import io.bitsquare.p2p.messaging.DecryptedMailboxListener;
 import io.bitsquare.p2p.messaging.DecryptedMsgWithPubKey;
@@ -81,7 +81,7 @@ public class TradeManager {
     private final Storage<TradableList<Trade>> tradableListStorage;
     private final TradableList<Trade> trades;
     private final BooleanProperty pendingTradesInitialized = new SimpleBooleanProperty();
-    private P2PServiceListener p2PServiceListener;
+    private P2PNetworkReadyListener p2PNetworkReadyListener;
 
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -148,29 +148,13 @@ public class TradeManager {
             }
         });
 
-        p2PServiceListener = new P2PServiceListener() {
+        p2PNetworkReadyListener = new P2PNetworkReadyListener() {
             @Override
-            public void onTorNodeReady() {
+            public void onFirstPeerAuthenticated() {
-            }
-
-            @Override
-            public void onHiddenServicePublished() {
-            }
-
-            @Override
-            public void onSetupFailed(Throwable throwable) {
-            }
-
-            @Override
-            public void onRequestingDataCompleted() {
-            }
-
-            @Override
-            public void onAuthenticated() {
                 initPendingTrades();
             }
         };
-        p2PService.addP2PServiceListener(p2PServiceListener);
+        p2PService.addP2PServiceListener(p2PNetworkReadyListener);
     }
 
 
@@ -185,7 +169,7 @@ public class TradeManager {
     }
 
     private void initPendingTrades() {
-        if (p2PServiceListener != null) p2PService.removeP2PServiceListener(p2PServiceListener);
+        if (p2PNetworkReadyListener != null) p2PService.removeP2PServiceListener(p2PNetworkReadyListener);
 
         List<Trade> failedTrades = new ArrayList<>();
         for (Trade trade : trades) {

core/src/main/java/io/bitsquare/trade/offer/OpenOfferManager.java

@@ -27,8 +27,8 @@ import io.bitsquare.common.handlers.ResultHandler;
 import io.bitsquare.common.util.Utilities;
 import io.bitsquare.p2p.Address;
 import io.bitsquare.p2p.Message;
+import io.bitsquare.p2p.P2PNetworkReadyListener;
 import io.bitsquare.p2p.P2PService;
-import io.bitsquare.p2p.P2PServiceListener;
 import io.bitsquare.p2p.messaging.SendMailMessageListener;
 import io.bitsquare.storage.Storage;
 import io.bitsquare.trade.TradableList;
@@ -68,7 +68,7 @@ public class OpenOfferManager {
     private final TradableList<OpenOffer> openOffers;
     private final Storage<TradableList<OpenOffer>> openOffersStorage;
     private boolean shutDownRequested;
-    private P2PServiceListener p2PServiceListener;
+    private P2PNetworkReadyListener p2PNetworkReadyListener;
     private final Timer timer = new Timer();
 
     ///////////////////////////////////////////////////////////////////////////////////////////
@@ -128,29 +128,13 @@ public class OpenOfferManager {
         // If offer removal at shutdown fails we don't want to have long term dangling dead offers, so we set TTL quite short and use re-publish as 
         // strategy. Offerers need to be online anyway.
         if (!p2PService.isAuthenticated()) {
-            p2PServiceListener = new P2PServiceListener() {
+            p2PNetworkReadyListener = new P2PNetworkReadyListener() {
                 @Override
-                public void onTorNodeReady() {
+                public void onFirstPeerAuthenticated() {
-                }
-
-                @Override
-                public void onHiddenServicePublished() {
-                }
-
-                @Override
-                public void onSetupFailed(Throwable throwable) {
-                }
-
-                @Override
-                public void onRequestingDataCompleted() {
-                }
-
-                @Override
-                public void onAuthenticated() {
                     startRePublishThread();
                 }
             };
-            p2PService.addP2PServiceListener(p2PServiceListener);
+            p2PService.addP2PServiceListener(p2PNetworkReadyListener);
 
         } else {
             startRePublishThread();
@@ -158,8 +142,8 @@ public class OpenOfferManager {
     }
 
     private void startRePublishThread() {
-        if (p2PServiceListener != null)
+        if (p2PNetworkReadyListener != null)
-            p2PService.removeP2PServiceListener(p2PServiceListener);
+            p2PService.removeP2PServiceListener(p2PNetworkReadyListener);
 
         long period = (long) (Offer.TTL * 0.8);
         TimerTask timerTask = new TimerTask() {

gui/src/main/java/io/bitsquare/gui/main/MainViewModel.java

@@ -220,7 +220,7 @@ class MainViewModel implements ViewModel {
             }
 
             @Override
-            public void onAuthenticated() {
+            public void onFirstPeerAuthenticated() {
                 p2pNetworkInfoFooter.set("Authenticated in P2P network.");
             }
 

jsocks/src/main/java/com/runjva/sourceforge/jsocks/protocol/Socks5Message.java

@@ -95,7 +95,7 @@ class Socks5Message extends ProxyMessage {
         this.host = hostName;
         this.version = SOCKS_VERSION;
 
-        log.debug("Doing ATYP_DOMAINNAME");
+        //log.debug("Doing ATYP_DOMAINNAME");
 
         addrType = SOCKS_ATYP_DOMAINNAME;
         final byte addr[] = hostName.getBytes();

jtorproxy/src/main/java/com/msopentech/thali/toronionproxy/OnionProxyManager.java

@@ -323,7 +323,7 @@ public abstract class OnionProxyManager {
         }
 
         if (phase != null && phase.contains("PROGRESS=100")) {
-            LOG.info("Tor has already bootstrapped");
+            LOG.trace("Tor has already bootstrapped");
             return true;
         }
 

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

@@ -0,0 +1,23 @@
+package io.bitsquare.p2p;
+
+
+public abstract class P2PNetworkReadyListener implements P2PServiceListener {
+    @Override
+    public void onTorNodeReady() {
+    }
+
+    @Override
+    public void onHiddenServicePublished() {
+    }
+
+    @Override
+    public void onSetupFailed(Throwable throwable) {
+    }
+
+    @Override
+    public void onRequestingDataCompleted() {
+    }
+
+    @Override
+    abstract public void onFirstPeerAuthenticated();
+}

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

@@ -141,7 +141,7 @@ public class P2PService implements SetupListener {
                 authenticated.set(true);
 
                 dataStorage.setAuthenticated();
-                p2pServiceListeners.stream().forEach(e -> e.onAuthenticated());
+                p2pServiceListeners.stream().forEach(e -> e.onFirstPeerAuthenticated());
             }
 
             @Override
@@ -660,7 +660,7 @@ public class P2PService implements SetupListener {
     ///////////////////////////////////////////////////////////////////////////////////////////
 
     private HashSet<ProtectedData> getDataSet() {
-        Log.traceCall();
+        // Log.traceCall();
         return new HashSet<>(dataStorage.getMap().values());
     }
 

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

@@ -7,5 +7,5 @@ public interface P2PServiceListener extends SetupListener {
 
     void onRequestingDataCompleted();
 
-    void onAuthenticated();
+    void onFirstPeerAuthenticated();
 }

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

@@ -5,6 +5,7 @@ import com.google.common.util.concurrent.Uninterruptibles;
 import io.bitsquare.app.Log;
 import io.bitsquare.common.ByteArrayUtils;
 import io.bitsquare.common.UserThread;
+import io.bitsquare.common.util.Utilities;
 import io.bitsquare.p2p.Address;
 import io.bitsquare.p2p.Message;
 import io.bitsquare.p2p.Utils;
@@ -18,6 +19,8 @@ import java.net.Socket;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.util.Date;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
@@ -33,7 +36,8 @@ public class Connection {
     private static final Logger log = LoggerFactory.getLogger(Connection.class);
     private static final int MAX_MSG_SIZE = 5 * 1024 * 1024;         // 5 MB of compressed data
     private static final int MAX_ILLEGAL_REQUESTS = 5;
-    private static final int SOCKET_TIMEOUT = 10 * 60 * 1000;        // 10 min. //TODO set shorter
+    private static final int SEND_MESSAGE_TIMEOUT = 10 * 1000;        // 10 sec.
+    private static final int SOCKET_TIMEOUT = 30 * 60 * 1000;        // 30 min.
     private InputHandler inputHandler;
     private volatile boolean isAuthenticated;
 
@@ -69,9 +73,12 @@ public class Connection {
 
     public Connection(Socket socket, MessageListener messageListener, ConnectionListener connectionListener) {
         Log.traceCall();
-        portInfo = "localPort=" + socket.getLocalPort() + "/port=" + socket.getPort();
         uid = UUID.randomUUID().toString();
-
+        if (socket.getLocalPort() == 0)
+            portInfo = "port=" + socket.getPort();
+        else
+            portInfo = "localPort=" + socket.getLocalPort() + "/port=" + socket.getPort();
+        
         init(socket, messageListener, connectionListener);
     }
 
@@ -88,6 +95,7 @@ public class Connection {
             objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
             ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
 
+
             // We create a thread for handling inputStream data
             inputHandler = new InputHandler(sharedSpace, objectInputStream, portInfo);
             singleThreadExecutor.submit(inputHandler);
@@ -121,6 +129,16 @@ public class Connection {
         Log.traceCall();
         if (!stopped) {
             try {
+                Timer timeoutTimer = new Timer();
+                timeoutTimer.schedule(new TimerTask() {
+                    @Override
+                    public void run() {
+                        Utilities.setThreadName("SendMessageTimerTask");
+                        throw new RuntimeException("Timeout occurred: Send message " + message
+                                + " on connection with port " + portInfo + " failed.");
+                    }
+                }, SEND_MESSAGE_TIMEOUT);
+
                 log.info("writeObject " + message + " on connection with port " + portInfo);
                 Object objectToWrite;
                 if (useCompression) {
@@ -140,6 +158,7 @@ public class Connection {
                     }
                     sharedSpace.updateLastActivityDate();
                 }
+                timeoutTimer.cancel();
             } catch (IOException e) {
                 // an exception lead to a shutdown
                 sharedSpace.handleConnectionException(e);
@@ -223,16 +242,13 @@ public class Connection {
 
             log.trace("ShutDown connection requested. Connection=" + this.toString());
 
-            stopped = true;
-            sharedSpace.stop();
-            if (inputHandler != null)
-                inputHandler.stop();
-
             if (sendCloseConnectionMessage) {
                 new Thread(() -> {
                     Thread.currentThread().setName("Connection:SendCloseConnectionMessage-" + this.objectId);
+                    Log.traceCall("sendCloseConnectionMessage");
                     try {
                         sendMessage(new CloseConnectionMessage());
+                        // TODO increase delay
                         Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
                     } catch (Throwable t) {
                         t.printStackTrace();
@@ -249,6 +265,11 @@ public class Connection {
 
     private void continueShutDown(@Nullable Runnable shutDownCompleteHandler) {
         Log.traceCall();
+
+        stopped = true;
+        sharedSpace.stop();
+        if (inputHandler != null)
+            inputHandler.stop();
         ConnectionListener.Reason shutDownReason = sharedSpace.getShutDownReason();
         if (shutDownReason == null)
             shutDownReason = ConnectionListener.Reason.SHUT_DOWN;
@@ -365,7 +386,8 @@ public class Connection {
         }
 
         public void handleConnectionException(Exception e) {
-            Log.traceCall();
+            Log.traceCall(e.toString());
+            log.warn("Exception might be expected: " + e.toString());
             if (e instanceof SocketException) {
                 if (socket.isClosed())
                     shutDownReason = ConnectionListener.Reason.SOCKET_CLOSED;
@@ -388,12 +410,12 @@ public class Connection {
         }
 
         public void onMessage(Message message) {
-            Log.traceCall();
+            //Log.traceCall();
             UserThread.execute(() -> messageListener.onMessage(message, connection));
         }
 
         public boolean useCompression() {
-            Log.traceCall();
+            //Log.traceCall();
             return useCompression;
         }
 
@@ -403,7 +425,7 @@ public class Connection {
         }
 
         public synchronized ConnectionListener getConnectionListener() {
-            Log.traceCall();
+            // Log.traceCall();
             return connectionListener;
         }
 
@@ -413,7 +435,7 @@ public class Connection {
         }
 
         public String getConnectionId() {
-            Log.traceCall();
+            //Log.traceCall();
             return connection.objectId;
         }
 
@@ -423,7 +445,7 @@ public class Connection {
         }
 
         public synchronized ConnectionListener.Reason getShutDownReason() {
-            Log.traceCall();
+            //Log.traceCall();
             return shutDownReason;
         }
 

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

@@ -105,7 +105,6 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
                             + "\n############################################################\n");
 
                     newConnection.sendMessage(message);
-
                     return newConnection; // can take a while when using tor
                 } catch (Throwable throwable) {
                     if (!(throwable instanceof ConnectException || throwable instanceof IOException)) {
@@ -138,7 +137,6 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
         ListenableFuture<Connection> future = executorService.submit(() -> {
             Thread.currentThread().setName("NetworkNode:SendMessage-to-" + connection.objectId);
             try {
-                log.debug("## connection.sendMessage");
                 connection.sendMessage(message);
                 return connection;
             } catch (Throwable t) {

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

@@ -228,14 +228,12 @@ public class TorNetworkNode extends NetworkNode {
         });
         Futures.addCallback(future, new FutureCallback<TorNode<JavaOnionProxyManager, JavaOnionProxyContext>>() {
             public void onSuccess(TorNode<JavaOnionProxyManager, JavaOnionProxyContext> torNode) {
-                Log.traceCall();
                 UserThread.execute(() -> {
                     resultHandler.accept(torNode);
                 });
             }
 
             public void onFailure(@NotNull Throwable throwable) {
-                Log.traceCall();
                 UserThread.execute(() -> {
                     log.error("TorNode creation failed with exception: " + throwable.getMessage());
                     restartTor();

network/src/main/java/io/bitsquare/p2p/peers/PeerGroup.java

@@ -42,8 +42,8 @@ public class PeerGroup {
         MAX_CONNECTIONS = maxConnections;
     }
 
-    private static final int SEND_PING_INTERVAL = 100000000;// new Random().nextInt(2 * 60 * 1000) + 2 * 60 * 1000; // 2-4 min.
+    private static final int SEND_PING_INTERVAL = new Random().nextInt(2 * 60 * 1000) + 2 * 60 * 1000; // 2-4 min.
-    private static final int GET_PEERS_INTERVAL = 100000000;//new Random().nextInt(1 * 60 * 1000) + 1 * 60 * 1000; // 1-2 min.
+    private static final int GET_PEERS_INTERVAL = new Random().nextInt(1 * 60 * 1000) + 1 * 60 * 1000; // 1-2 min.
     private static final int PING_AFTER_CONNECTION_INACTIVITY = 30 * 1000;
     private static final int MAX_REPORTED_PEERS = 1000;
 
@@ -472,7 +472,7 @@ public class PeerGroup {
                         });
                     }, 5, 10));
         } else {
-            log.debug("No peers available for requesting.");
+            log.debug("No peers available for requesting data.");
         }
     }
 
@@ -641,7 +641,7 @@ public class PeerGroup {
     }
 
     private Address getMyAddress() {
-        Log.traceCall();
+        // Log.traceCall();
         return networkNode.getAddress();
     }
 

network/src/main/java/io/bitsquare/p2p/storage/ProtectedExpirableDataStorage.java

@@ -211,7 +211,7 @@ public class ProtectedExpirableDataStorage {
     }
 
     public Map<BigInteger, ProtectedData> getMap() {
-        Log.traceCall();
+        //Log.traceCall();
         return map;
     }
 
@@ -362,7 +362,7 @@ public class ProtectedExpirableDataStorage {
     }
 
     private BigInteger getHashAsBigInteger(ExpirablePayload payload) {
-        Log.traceCall();
+        //Log.traceCall();
         return new BigInteger(Hash.getHash(payload));
     }
 }

network/src/test/java/io/bitsquare/p2p/TestUtils.java

@@ -86,7 +86,7 @@ public class TestUtils {
             }
 
             @Override
-            public void onAuthenticated() {
+            public void onFirstPeerAuthenticated() {
             }
 
             @Override
@@ -129,7 +129,7 @@ public class TestUtils {
             }
 
             @Override
-            public void onAuthenticated() {
+            public void onFirstPeerAuthenticated() {
                 latch.countDown();
             }
 

network/src/test/java/io/bitsquare/p2p/routing/PeerGroupTest.java

@@ -95,7 +95,7 @@ public class PeerGroupTest {
             }
 
             @Override
-            public void onAuthenticated() {
+            public void onFirstPeerAuthenticated() {
             }
 
             @Override
@@ -139,7 +139,7 @@ public class PeerGroupTest {
             }
 
             @Override
-            public void onAuthenticated() {
+            public void onFirstPeerAuthenticated() {
                 latch.countDown();
             }
 
@@ -170,7 +170,7 @@ public class PeerGroupTest {
             }
 
             @Override
-            public void onAuthenticated() {
+            public void onFirstPeerAuthenticated() {
                 latch.countDown();
             }
 
@@ -397,7 +397,7 @@ public class PeerGroupTest {
             }
 
             @Override
-            public void onAuthenticated() {
+            public void onFirstPeerAuthenticated() {
             }
 
             @Override