Git-Mirrors/haveno
1
0
Fork 0
mirror of https://github.com/haveno-dex/haveno.git synced 2025-07-22 14:41:08 -04:00
Code Issues Projects Releases Packages Wiki Activity

Use sets instead of lists, fix threading issues

Browse source
This commit is contained in:
Manfred Karrer 2015-11-05 23:23:15 +01:00
parent a1993a5d9a
commit 81bf4b65d4
22 changed files with 107 additions and 852 deletions
Download patch file Download diff file Expand all files Collapse all files

6
core/src/main/java/io/bitsquare/arbitration/DisputeManager.java
View file

@ -51,7 +51,7 @@ import java.io.File;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet;
import java.util.stream.Collectors; import java.util.stream.Collectors;
public class DisputeManager { public class DisputeManager {
@ -68,8 +68,8 @@ public class DisputeManager {
transient private final ObservableList<Dispute> disputesObservableList; transient private final ObservableList<Dispute> disputesObservableList;
private final String disputeInfo; private final String disputeInfo;
private final P2PServiceListener p2PServiceListener; private final P2PServiceListener p2PServiceListener;
private final List<DecryptedMsgWithPubKey> decryptedMailboxMessageWithPubKeys = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<DecryptedMsgWithPubKey> decryptedMailboxMessageWithPubKeys = new CopyOnWriteArraySet<>();
private final List<DecryptedMsgWithPubKey> decryptedMailMessageWithPubKeys = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<DecryptedMsgWithPubKey> decryptedMailMessageWithPubKeys = new CopyOnWriteArraySet<>();
/////////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////////

8
core/src/main/java/io/bitsquare/btc/WalletService.java
View file

@ -52,7 +52,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.time.Duration; import java.time.Duration;
import java.util.*; import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException; import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -68,9 +68,9 @@ public class WalletService {
public static final String PREFIX_KEY = "wallet.prefix"; public static final String PREFIX_KEY = "wallet.prefix";
private static final long STARTUP_TIMEOUT = 60 * 1000; private static final long STARTUP_TIMEOUT = 60 * 1000;
private final List<AddressConfidenceListener> addressConfidenceListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<AddressConfidenceListener> addressConfidenceListeners = new CopyOnWriteArraySet<>();
private final List<TxConfidenceListener> txConfidenceListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<TxConfidenceListener> txConfidenceListeners = new CopyOnWriteArraySet<>();
private final List<BalanceListener> balanceListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<BalanceListener> balanceListeners = new CopyOnWriteArraySet<>();
private final DownloadListener downloadListener = new DownloadListener(); private final DownloadListener downloadListener = new DownloadListener();
private final WalletEventListener walletEventListener = new BitsquareWalletEventListener(); private final WalletEventListener walletEventListener = new BitsquareWalletEventListener();

10
core/src/main/resources/logback.xml
View file

@ -32,16 +32,6 @@
<logger name="org.bitcoinj.core.BitcoinSerializer" level="ERROR"/> <logger name="org.bitcoinj.core.BitcoinSerializer" level="ERROR"/>
<logger name="org.bitcoinj.core.Peer" level="ERROR"/>--> <logger name="org.bitcoinj.core.Peer" level="ERROR"/>-->
<!-- <logger name="net.tomp2p.message.Encoder" level="WARN"/>
<logger name="net.tomp2p.message.Decoder" level="WARN"/>
<logger name="net.tomp2p.message.MessageHeaderCodec" level="WARN"/>
<logger name="io.netty.util" level="WARN"/>
<logger name="io.netty.channel" level="WARN"/>
<logger name="io.netty.buffer" level="WARN"/>-->
<!-- <logger name="org.bitcoinj.core.BitcoinSerializer" level="WARN"/> <!-- <logger name="org.bitcoinj.core.BitcoinSerializer" level="WARN"/>
<logger name="org.bitcoinj.core.AbstractBlockChain" level="WARN"/> <logger name="org.bitcoinj.core.AbstractBlockChain" level="WARN"/>
<logger name="org.bitcoinj.wallet.DeterministicKeyChain" level="WARN"/>--> <logger name="org.bitcoinj.wallet.DeterministicKeyChain" level="WARN"/>-->

52
doc/tomp2pReadme.md
View file

@ -1,52 +0,0 @@
How to publish custom TomP2P snapshots:
Update this fork to upstream/master:
git clone https://github.com/bitsquare/TomP2P.git
cd TomP2P
git remote add upstream https://github.com/tomp2p/TomP2P.git
git checkout master
git pull upstream master
git push origin master
Create a custom qualified snapshot version:
git checkout published (if the first time: git checkout -b published)
git reset --hard master
export COMMITHASH=$(git log --oneline -1 | cut -d" " -f1)
git grep -l 5.0-Alpha | xargs perl -p -i -e "s/5.0-Alpha(..?)-SNAPSHOT/5.0-Alpha$1.$COMMITHASH-SNAPSHOT/"
git diff # review changes to poms
git commit -am"Qualify pom version for publication"
Build artifacts:
mvn clean package -DskipTests
Upload artifacts to Artifactory:
Log in at https://partnerdemo.artifactoryonline.com/partnerdemo/webapp/login.html (@ManfredKarrer and @cbeams know the creds)
Go to https://partnerdemo.artifactoryonline.com/partnerdemo/webapp/deployartifact.html
Upload each of the tomp2p jar files, and accept all defaults in the form that follows
Finally, upload the top-level pom.xml (this is the tomp2p-parent pom).
Re-apply this README:
git show bitsquare-published-91276e8:README > README
git add README
git commit -m"Add README with publication instructions"
Tag the published branch:
git tag bitsquare-published-$COMMITHASH published
(Force) push the published branch:
git push -f origin published
Push tags:
git push --tags
Note this would all be a lot easier if TomP2P published its own snapshots :)

29
doc/update_fork_tomp2p.sh
View file

@ -1,29 +0,0 @@
#!/bin/bash
cd /Users/mk/Documents/_intellij/TomP2P-master_fork/TomP2P
git reset --hard
git remote add upstream https://github.com/tomp2p/TomP2P.git
git checkout master
git pull upstream master
git push origin master
git checkout published
git reset --hard master
export COMMITHASH=$(git log --oneline -1 | cut -d" " -f1)
git grep -l 5.0-Alpha | xargs perl -p -i -e "s/5.0-Alpha(..?)-SNAPSHOT/5.0-Alpha$1.$COMMITHASH-SNAPSHOT/"
git commit -am"Qualify pom version for publication"
echo $COMMITHASH
mvn clean install -DskipTests
git show bitsquare-published-91276e8:README > README
git add README
git commit -m"Add README with publication instructions"
git tag bitsquare-published-$COMMITHASH published
git push -f origin published
git push --tags
echo $COMMITHASH

5
gui/src/main/java/io/bitsquare/gui/Navigation.java
View file

@ -29,8 +29,7 @@ import org.slf4j.LoggerFactory;
import java.io.Serializable; import java.io.Serializable;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CopyOnWriteArrayList;
public class Navigation implements Serializable { public class Navigation implements Serializable {
// That object is saved to disc. We need to take care of changes to not break deserialization. // That object is saved to disc. We need to take care of changes to not break deserialization.
@ -46,7 +45,7 @@ public class Navigation implements Serializable {
// New listeners can be added during iteration so we use CopyOnWriteArrayList to // New listeners can be added during iteration so we use CopyOnWriteArrayList to
// prevent invalid array modification // prevent invalid array modification
transient private final List<Listener> listeners = new CopyOnWriteArrayList<>(); transient private final CopyOnWriteArraySet<Listener> listeners = new CopyOnWriteArraySet<>();
transient private final Storage<Navigation> remoteStorage; transient private final Storage<Navigation> remoteStorage;
transient private ViewPath currentPath; transient private ViewPath currentPath;
// Used for returning to the last important view. After setup is done we want to // Used for returning to the last important view. After setup is done we want to

4
gui/src/main/java/io/bitsquare/gui/main/offer/offerbook/OfferBook.java
View file

@ -30,7 +30,7 @@ import org.slf4j.LoggerFactory;
import javax.inject.Inject; import javax.inject.Inject;
import java.io.Serializable; import java.io.Serializable;
import java.util.List; import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet;
/** /**
* Holds and manages the unsorted and unfiltered offerbook list of both buy and sell offers. * Holds and manages the unsorted and unfiltered offerbook list of both buy and sell offers.
@ -91,7 +91,7 @@ public class OfferBook {
public void fillOfferBookListItems() { public void fillOfferBookListItems() {
log.debug("fillOfferBookListItems"); log.debug("fillOfferBookListItems");
List<Offer> offers = offerBookService.getOffers(); List<Offer> offers = offerBookService.getOffers();
CopyOnWriteArrayList<OfferBookListItem> list = new CopyOnWriteArrayList<>(); CopyOnWriteArraySet<OfferBookListItem> list = new CopyOnWriteArraySet<>();
offers.stream().forEach(e -> list.add(new OfferBookListItem(e))); offers.stream().forEach(e -> list.add(new OfferBookListItem(e)));
offerBookListItems.clear(); offerBookListItems.clear();
offerBookListItems.addAll(list); offerBookListItems.addAll(list);

6
gui/src/main/java/io/bitsquare/gui/util/FormBuilder.java
View file

@ -30,8 +30,8 @@ import javafx.scene.layout.*;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.List; import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet;
public class FormBuilder { public class FormBuilder {
private static final Logger log = LoggerFactory.getLogger(FormBuilder.class); private static final Logger log = LoggerFactory.getLogger(FormBuilder.class);
@ -633,7 +633,7 @@ public class FormBuilder {
} }
public static void removeRowsFromGridPane(GridPane gridPane, int fromGridRow, int toGridRow) { public static void removeRowsFromGridPane(GridPane gridPane, int fromGridRow, int toGridRow) {
List<Node> nodes = new CopyOnWriteArrayList<>(gridPane.getChildren()); Set<Node> nodes = new CopyOnWriteArraySet<>(gridPane.getChildren());
nodes.stream() nodes.stream()
.filter(e -> GridPane.getRowIndex(e) >= fromGridRow && GridPane.getRowIndex(e) <= toGridRow) .filter(e -> GridPane.getRowIndex(e) >= fromGridRow && GridPane.getRowIndex(e) <= toGridRow)
.forEach(e -> gridPane.getChildren().remove(e)); .forEach(e -> gridPane.getChildren().remove(e));

5
gui/src/main/resources/logback.xml
View file

@ -24,7 +24,6 @@
<logger name="io.bitsquare.storage.FileManager" level="WARN"/> <logger name="io.bitsquare.storage.FileManager" level="WARN"/>
<logger name="org.bitcoinj" level="WARN"/> <logger name="org.bitcoinj" level="WARN"/>
<logger name="net.tomp2p" level="WARN"/>
<logger name="org.bitcoinj.core.BitcoinSerializer" level="WARN"/> <logger name="org.bitcoinj.core.BitcoinSerializer" level="WARN"/>
<logger name="org.bitcoinj.core.Peer" level="WARN"/> <logger name="org.bitcoinj.core.Peer" level="WARN"/>
@ -36,10 +35,6 @@
<logger name="org.bitcoinj.core.BitcoinSerializer" level="ERROR"/> <logger name="org.bitcoinj.core.BitcoinSerializer" level="ERROR"/>
<logger name="org.bitcoinj.core.Peer" level="ERROR"/>--> <logger name="org.bitcoinj.core.Peer" level="ERROR"/>-->
<!-- <logger name="net.tomp2p.message.Encoder" level="WARN"/>
<logger name="net.tomp2p.message.Decoder" level="WARN"/>
<logger name="net.tomp2p.message.MessageHeaderCodec" level="WARN"/>
<logger name="io.netty.util" level="WARN"/> <logger name="io.netty.util" level="WARN"/>
<logger name="io.netty.channel" level="WARN"/> <logger name="io.netty.channel" level="WARN"/>

23
network/src/main/java/io/bitsquare/p2p/P2PService.java
View file

@ -1,5 +1,6 @@
package io.bitsquare.p2p; package io.bitsquare.p2p;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture; import com.google.common.util.concurrent.SettableFuture;
@ -37,7 +38,7 @@ import java.math.BigInteger;
import java.security.PublicKey; import java.security.PublicKey;
import java.util.*; import java.util.*;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet;
import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkNotNull;
@ -62,20 +63,20 @@ public class P2PService {
private NetworkNode networkNode; private NetworkNode networkNode;
private PeerGroup peerGroup; private PeerGroup peerGroup;
private ProtectedExpirableDataStorage dataStorage; private ProtectedExpirableDataStorage dataStorage;
private final List<DecryptedMailListener> decryptedMailListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<DecryptedMailListener> decryptedMailListeners = new CopyOnWriteArraySet<>();
private final List<DecryptedMailboxListener> decryptedMailboxListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<DecryptedMailboxListener> decryptedMailboxListeners = new CopyOnWriteArraySet<>();
private final List<P2PServiceListener> p2pServiceListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<P2PServiceListener> p2pServiceListeners = new CopyOnWriteArraySet<>();
private final Map<DecryptedMsgWithPubKey, ProtectedMailboxData> mailboxMap = new ConcurrentHashMap<>(); private final Map<DecryptedMsgWithPubKey, ProtectedMailboxData> mailboxMap = new ConcurrentHashMap<>();
private volatile boolean shutDownInProgress; private volatile boolean shutDownInProgress;
private List<Address> seedNodeAddresses; private Set<Address> seedNodeAddresses;
private Set<Address> connectedSeedNodes = new HashSet<>(); private Set<Address> connectedSeedNodes = new HashSet<>();
private Set<Address> authenticatedPeerAddresses = new HashSet<>(); private Set<Address> authenticatedPeerAddresses = new HashSet<>();
private boolean authenticatedToFirstPeer; private boolean authenticatedToFirstPeer;
private boolean allDataReceived; private boolean allDataReceived;
public boolean authenticated; public boolean authenticated;
private boolean shutDownComplete; private boolean shutDownComplete;
private List<Runnable> shutDownResultHandlers = new CopyOnWriteArrayList<>(); private CopyOnWriteArraySet<Runnable> shutDownResultHandlers = new CopyOnWriteArraySet<>();
private final List<Long> getDataSetMessageNonceList = new ArrayList<>(); private final CopyOnWriteArraySet<Long> getDataSetMessageNonceList = new CopyOnWriteArraySet<>();
private boolean allSeedNodesRequested; private boolean allSeedNodesRequested;
private Timer sendGetAllDataMessageTimer; private Timer sendGetAllDataMessageTimer;
private volatile boolean hiddenServiceReady; private volatile boolean hiddenServiceReady;
@ -565,11 +566,11 @@ public class P2PService {
// Private // Private
/////////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////////
private void sendGetAllDataMessage(List<Address> seedNodeAddresses) { private void sendGetAllDataMessage(Set<Address> seedNodeAddresses) {
Address networkNodeAddress = networkNode.getAddress(); Address networkNodeAddress = networkNode.getAddress();
if (networkNodeAddress != null) if (networkNodeAddress != null)
seedNodeAddresses.remove(networkNodeAddress); seedNodeAddresses.remove(networkNodeAddress);
List<Address> remainingSeedNodeAddresses = new CopyOnWriteArrayList<>(seedNodeAddresses); List<Address> remainingSeedNodeAddresses = new ArrayList<>(seedNodeAddresses);
if (!seedNodeAddresses.isEmpty()) { if (!seedNodeAddresses.isEmpty()) {
Collections.shuffle(remainingSeedNodeAddresses); Collections.shuffle(remainingSeedNodeAddresses);
@ -593,7 +594,7 @@ public class P2PService {
sendGetAllDataMessageTimer = Utilities.runTimerTaskWithRandomDelay(() -> { sendGetAllDataMessageTimer = Utilities.runTimerTaskWithRandomDelay(() -> {
Thread.currentThread().setName("SendGetAllDataMessageTimer-" + new Random().nextInt(1000)); Thread.currentThread().setName("SendGetAllDataMessageTimer-" + new Random().nextInt(1000));
try { try {
UserThread.execute(() -> sendGetAllDataMessage(remainingSeedNodeAddresses)); UserThread.execute(() -> sendGetAllDataMessage(Sets.newHashSet(remainingSeedNodeAddresses)));
} catch (Throwable t) { } catch (Throwable t) {
t.printStackTrace(); t.printStackTrace();
log.error("Executing task failed. " + t.getMessage()); log.error("Executing task failed. " + t.getMessage());
@ -608,7 +609,7 @@ public class P2PService {
public void onFailure(Throwable throwable) { public void onFailure(Throwable throwable) {
log.info("Send GetAllDataMessage to " + candidate + " failed. Exception:" + throwable.getMessage()); log.info("Send GetAllDataMessage to " + candidate + " failed. Exception:" + throwable.getMessage());
log.trace("We try to connect another random seed node. " + remainingSeedNodeAddresses); log.trace("We try to connect another random seed node. " + remainingSeedNodeAddresses);
UserThread.execute(() -> sendGetAllDataMessage(remainingSeedNodeAddresses)); UserThread.execute(() -> sendGetAllDataMessage(Sets.newHashSet(remainingSeedNodeAddresses)));
} }
}); });
} else { } else {

26
network/src/main/java/io/bitsquare/p2p/network/Connection.java
View file

@ -18,7 +18,6 @@ import java.net.Socket;
import java.net.SocketException; import java.net.SocketException;
import java.net.SocketTimeoutException; import java.net.SocketTimeoutException;
import java.util.Date; import java.util.Date;
import java.util.Map;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
@ -34,7 +33,7 @@ public class Connection {
private static final Logger log = LoggerFactory.getLogger(Connection.class); 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_MSG_SIZE = 5 * 1024 * 1024; // 5 MB of compressed data
private static final int MAX_ILLEGAL_REQUESTS = 5; private static final int MAX_ILLEGAL_REQUESTS = 5;
private static final int SOCKET_TIMEOUT = 60 * 1000; // 1 min. private static final int SOCKET_TIMEOUT = 10 * 60 * 1000; // 10 min. //TODO set shorter
private InputHandler inputHandler; private InputHandler inputHandler;
private boolean isAuthenticated; private boolean isAuthenticated;
@ -44,7 +43,7 @@ public class Connection {
private final String portInfo; private final String portInfo;
private final String uid; private final String uid;
private final ExecutorService executorService = Executors.newSingleThreadExecutor(); private final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// set in init // set in init
private ObjectOutputStream objectOutputStream; private ObjectOutputStream objectOutputStream;
@ -89,7 +88,7 @@ public class Connection {
// We create a thread for handling inputStream data // We create a thread for handling inputStream data
inputHandler = new InputHandler(sharedSpace, objectInputStream, portInfo); inputHandler = new InputHandler(sharedSpace, objectInputStream, portInfo);
executorService.submit(inputHandler); singleThreadExecutor.submit(inputHandler);
} catch (IOException e) { } catch (IOException e) {
sharedSpace.handleConnectionException(e); sharedSpace.handleConnectionException(e);
} }
@ -208,18 +207,19 @@ public class Connection {
UserThread.execute(() -> sharedSpace.getConnectionListener().onDisconnect(ConnectionListener.Reason.SHUT_DOWN, this)); UserThread.execute(() -> sharedSpace.getConnectionListener().onDisconnect(ConnectionListener.Reason.SHUT_DOWN, this));
if (sendCloseConnectionMessage) { if (sendCloseConnectionMessage) {
executorService.submit(() -> { new Thread(() -> {
Thread.currentThread().setName("Connection:Send-CloseConnectionMessage-" + this.getObjectId()); Thread.currentThread().setName("Connection:SendCloseConnectionMessage-" + this.getObjectId());
try { try {
sendMessage(new CloseConnectionMessage()); sendMessage(new CloseConnectionMessage());
// give a bit of time for closing gracefully // give a bit of time for closing gracefully
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
UserThread.execute(() -> continueShutDown(shutDownCompleteHandler));
} catch (Throwable t) { } catch (Throwable t) {
throw t; t.printStackTrace();
log.error(t.getMessage());
} finally {
UserThread.execute(() -> continueShutDown(shutDownCompleteHandler));
} }
}); }).start();
} else { } else {
continueShutDown(shutDownCompleteHandler); continueShutDown(shutDownCompleteHandler);
} }
@ -235,7 +235,7 @@ public class Connection {
} catch (IOException e) { } catch (IOException e) {
e.printStackTrace(); e.printStackTrace();
} finally { } finally {
MoreExecutors.shutdownAndAwaitTermination(executorService, 500, TimeUnit.MILLISECONDS); MoreExecutors.shutdownAndAwaitTermination(singleThreadExecutor, 500, TimeUnit.MILLISECONDS);
log.debug("Connection shutdown complete " + this.toString()); log.debug("Connection shutdown complete " + this.toString());
// dont use executorService as its shut down but call handler on own thread // dont use executorService as its shut down but call handler on own thread
@ -302,7 +302,7 @@ public class Connection {
private final MessageListener messageListener; private final MessageListener messageListener;
private final ConnectionListener connectionListener; private final ConnectionListener connectionListener;
private final boolean useCompression; private final boolean useCompression;
private final Map<IllegalRequest, Integer> illegalRequests = new ConcurrentHashMap<>(); private final ConcurrentHashMap<IllegalRequest, Integer> illegalRequests = new ConcurrentHashMap<>();
// mutable // mutable
private Date lastActivityDate; private Date lastActivityDate;

4
network/src/main/java/io/bitsquare/p2p/network/LocalhostNetworkNode.java
View file

@ -25,8 +25,8 @@ import java.util.function.Consumer;
public class LocalhostNetworkNode extends NetworkNode { public class LocalhostNetworkNode extends NetworkNode {
private static final Logger log = LoggerFactory.getLogger(LocalhostNetworkNode.class); private static final Logger log = LoggerFactory.getLogger(LocalhostNetworkNode.class);
private static int simulateTorDelayTorNode = 2 * 1000; private static int simulateTorDelayTorNode = 2 * 100;
private static int simulateTorDelayHiddenService = 2 * 1000; private static int simulateTorDelayHiddenService = 2 * 100;
private Address address; private Address address;
public static void setSimulateTorDelayTorNode(int simulateTorDelayTorNode) { public static void setSimulateTorDelayTorNode(int simulateTorDelayTorNode) {

12
network/src/main/java/io/bitsquare/p2p/network/NetworkNode.java
View file

@ -15,10 +15,8 @@ import java.net.ConnectException;
import java.net.ServerSocket; import java.net.ServerSocket;
import java.net.Socket; import java.net.Socket;
import java.util.HashSet; import java.util.HashSet;
import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.CopyOnWriteArraySet;
import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkNotNull;
@ -27,11 +25,11 @@ public abstract class NetworkNode implements MessageListener, ConnectionListener
private static final Logger log = LoggerFactory.getLogger(NetworkNode.class); private static final Logger log = LoggerFactory.getLogger(NetworkNode.class);
protected final int port; protected final int port;
private final Set<Connection> outBoundConnections = new CopyOnWriteArraySet<>(); private final CopyOnWriteArraySet<Connection> outBoundConnections = new CopyOnWriteArraySet<>();
private final Set<Connection> inBoundConnections = new CopyOnWriteArraySet<>(); private final CopyOnWriteArraySet<Connection> inBoundConnections = new CopyOnWriteArraySet<>();
private final List<MessageListener> messageListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<MessageListener> messageListeners = new CopyOnWriteArraySet<>();
private final List<ConnectionListener> connectionListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<ConnectionListener> connectionListeners = new CopyOnWriteArraySet<>();
protected final List<SetupListener> setupListeners = new CopyOnWriteArrayList<>(); protected final CopyOnWriteArraySet<SetupListener> setupListeners = new CopyOnWriteArraySet<>();
protected ListeningExecutorService executorService; protected ListeningExecutorService executorService;
private Server server; private Server server;
private volatile boolean shutDownInProgress; private volatile boolean shutDownInProgress;

52
network/src/main/java/io/bitsquare/p2p/peer/PeerGroup.java
View file

@ -1,5 +1,6 @@
package io.bitsquare.p2p.peer; package io.bitsquare.p2p.peer;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture; import com.google.common.util.concurrent.SettableFuture;
@ -16,7 +17,6 @@ import org.slf4j.LoggerFactory;
import java.util.*; import java.util.*;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -40,12 +40,12 @@ public class PeerGroup {
} }
private final NetworkNode networkNode; private final NetworkNode networkNode;
private final List<Address> seedNodes; private final CopyOnWriteArraySet<Address> seedNodes;
private final Map<Address, Long> nonceMap = new ConcurrentHashMap<>(); private final ConcurrentHashMap<Address, Long> nonceMap = new ConcurrentHashMap<>();
private final List<PeerListener> peerListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<PeerListener> peerListeners = new CopyOnWriteArraySet<>();
private final Map<Address, Peer> authenticatedPeers = new ConcurrentHashMap<>(); private final ConcurrentHashMap<Address, Peer> authenticatedPeers = new ConcurrentHashMap<>();
private final Set<Address> reportedPeerAddresses = new CopyOnWriteArraySet<>(); private final CopyOnWriteArraySet<Address> reportedPeerAddresses = new CopyOnWriteArraySet<>();
private final Map<Address, Runnable> authenticationCompleteHandlers = new ConcurrentHashMap<>(); private final ConcurrentHashMap<Address, Runnable> authenticationCompleteHandlers = new ConcurrentHashMap<>();
private final Timer maintenanceTimer = new Timer(); private final Timer maintenanceTimer = new Timer();
private volatile boolean shutDownInProgress; private volatile boolean shutDownInProgress;
@ -54,11 +54,11 @@ public class PeerGroup {
// Constructor // Constructor
/////////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////////
public PeerGroup(final NetworkNode networkNode, List<Address> seeds) { public PeerGroup(final NetworkNode networkNode, Set<Address> seeds) {
this.networkNode = networkNode; this.networkNode = networkNode;
// We copy it as we remove ourselves later from the list if we are a seed node // We copy it as we remove ourselves later from the list if we are a seed node
this.seedNodes = new CopyOnWriteArrayList<>(seeds); this.seedNodes = new CopyOnWriteArraySet<>(seeds);
init(networkNode); init(networkNode);
} }
@ -146,7 +146,7 @@ public class PeerGroup {
private void pingPeers() { private void pingPeers() {
log.trace("pingPeers"); log.trace("pingPeers");
List<Peer> connectedPeersList = new ArrayList<>(authenticatedPeers.values()); Set<Peer> connectedPeersList = new HashSet<>(authenticatedPeers.values());
connectedPeersList.stream() connectedPeersList.stream()
.filter(e -> (new Date().getTime() - e.connection.getLastActivityDate().getTime()) > PING_AFTER_CONNECTION_INACTIVITY) .filter(e -> (new Date().getTime() - e.connection.getLastActivityDate().getTime()) > PING_AFTER_CONNECTION_INACTIVITY)
.forEach(e -> Utilities.runTimerTaskWithRandomDelay(() -> { .forEach(e -> Utilities.runTimerTaskWithRandomDelay(() -> {
@ -225,9 +225,8 @@ public class PeerGroup {
return authenticatedPeers; return authenticatedPeers;
} }
// Use ArrayList not List as we need it serializable public Set<Address> getAllPeerAddresses() {
public ArrayList<Address> getAllPeerAddresses() { CopyOnWriteArraySet<Address> allPeerAddresses = new CopyOnWriteArraySet<>(reportedPeerAddresses);
ArrayList<Address> allPeerAddresses = new ArrayList<>(reportedPeerAddresses);
allPeerAddresses.addAll(authenticatedPeers.values().stream() allPeerAddresses.addAll(authenticatedPeers.values().stream()
.map(e -> e.address).collect(Collectors.toList())); .map(e -> e.address).collect(Collectors.toList()));
// remove own address and seed nodes // remove own address and seed nodes
@ -255,7 +254,7 @@ public class PeerGroup {
}); });
} }
private void sendRequestAuthenticationMessage(final List<Address> remainingSeedNodes, final Address address) { private void sendRequestAuthenticationMessage(Set<Address> remainingSeedNodes, final Address address) {
log.info("We try to authenticate to a random seed node. " + address); log.info("We try to authenticate to a random seed node. " + address);
startAuthTs = System.currentTimeMillis(); startAuthTs = System.currentTimeMillis();
final boolean[] alreadyConnected = {false}; final boolean[] alreadyConnected = {false};
@ -285,8 +284,8 @@ public class PeerGroup {
} }
} }
private void getNextSeedNode(List<Address> remainingSeedNodes) { private void getNextSeedNode(Set<Address> remainingSeedNodes) {
List<Address> remainingSeedNodeAddresses = new CopyOnWriteArrayList<>(remainingSeedNodes); List<Address> remainingSeedNodeAddresses = new ArrayList<>(remainingSeedNodes);
Address myAddress = getAddress(); Address myAddress = getAddress();
if (myAddress != null) if (myAddress != null)
@ -295,7 +294,7 @@ public class PeerGroup {
if (!remainingSeedNodeAddresses.isEmpty()) { if (!remainingSeedNodeAddresses.isEmpty()) {
Collections.shuffle(remainingSeedNodeAddresses); Collections.shuffle(remainingSeedNodeAddresses);
Address address = remainingSeedNodeAddresses.remove(0); Address address = remainingSeedNodeAddresses.remove(0);
sendRequestAuthenticationMessage(remainingSeedNodeAddresses, address); sendRequestAuthenticationMessage(Sets.newHashSet(remainingSeedNodeAddresses), address);
} else { } else {
log.info("No other seed node found. That is expected for the first seed node."); log.info("No other seed node found. That is expected for the first seed node.");
} }
@ -346,12 +345,14 @@ public class PeerGroup {
ChallengeMessage challengeMessage = (ChallengeMessage) message; ChallengeMessage challengeMessage = (ChallengeMessage) message;
Address peerAddress = challengeMessage.address; Address peerAddress = challengeMessage.address;
log.trace("ChallengeMessage from " + peerAddress + " at " + getAddress()); log.trace("ChallengeMessage from " + peerAddress + " at " + getAddress());
log.trace("nonceMap" + nonceMap);
log.trace("challengeMessage" + challengeMessage);
HashMap<Address, Long> tempNonceMap = new HashMap<>(nonceMap); HashMap<Address, Long> tempNonceMap = new HashMap<>(nonceMap);
boolean verified = verifyNonceAndAuthenticatePeerAddress(challengeMessage.requesterNonce, peerAddress); boolean verified = verifyNonceAndAuthenticatePeerAddress(challengeMessage.requesterNonce, peerAddress);
if (verified) { if (verified) {
connection.setPeerAddress(peerAddress); connection.setPeerAddress(peerAddress);
SettableFuture<Connection> future = networkNode.sendMessage(peerAddress, SettableFuture<Connection> future = networkNode.sendMessage(peerAddress,
new GetPeersMessage(getAddress(), challengeMessage.challengerNonce, getAllPeerAddresses())); new GetPeersMessage(getAddress(), challengeMessage.challengerNonce, new ArrayList<Address>(getAllPeerAddresses())));
Futures.addCallback(future, new FutureCallback<Connection>() { Futures.addCallback(future, new FutureCallback<Connection>() {
@Override @Override
public void onSuccess(Connection connection) { public void onSuccess(Connection connection) {
@ -376,7 +377,7 @@ public class PeerGroup {
setAuthenticated(connection, peerAddress); setAuthenticated(connection, peerAddress);
purgeReportedPeers(); purgeReportedPeers();
SettableFuture<Connection> future = networkNode.sendMessage(peerAddress, SettableFuture<Connection> future = networkNode.sendMessage(peerAddress,
new PeersMessage(getAddress(), getAllPeerAddresses())); new PeersMessage(getAddress(), new ArrayList(getAllPeerAddresses())));
log.trace("sent PeersMessage to " + peerAddress + " from " + getAddress() log.trace("sent PeersMessage to " + peerAddress + " from " + getAddress()
+ " with allPeers=" + getAllPeerAddresses()); + " with allPeers=" + getAllPeerAddresses());
Futures.addCallback(future, new FutureCallback<Connection>() { Futures.addCallback(future, new FutureCallback<Connection>() {
@ -443,7 +444,7 @@ public class PeerGroup {
int all = getAllPeerAddresses().size(); int all = getAllPeerAddresses().size();
if (all > 1000) { if (all > 1000) {
int diff = all - 100; int diff = all - 100;
List<Address> list = getNotConnectedPeerAddresses(); List<Address> list = new ArrayList<>(getNotConnectedPeerAddresses());
for (int i = 0; i < diff; i++) { for (int i = 0; i < diff; i++) {
Address toRemove = list.remove(new Random().nextInt(list.size())); Address toRemove = list.remove(new Random().nextInt(list.size()));
reportedPeerAddresses.remove(toRemove); reportedPeerAddresses.remove(toRemove);
@ -451,13 +452,9 @@ public class PeerGroup {
} }
} }
private List<Address> getNotConnectedPeerAddresses() { private synchronized CopyOnWriteArraySet<Address> getNotConnectedPeerAddresses() {
ArrayList<Address> list = new ArrayList<>(getAllPeerAddresses()); CopyOnWriteArraySet<Address> list = new CopyOnWriteArraySet<>(getAllPeerAddresses());
log.debug("## getNotConnectedPeerAddresses ");
log.debug("## reportedPeersList=" + list);
authenticatedPeers.values().stream().forEach(e -> list.remove(e.address)); authenticatedPeers.values().stream().forEach(e -> list.remove(e.address));
log.debug("## connectedPeers=" + authenticatedPeers);
log.debug("## reportedPeersList=" + list);
return list; return list;
} }
@ -536,7 +533,7 @@ public class PeerGroup {
} }
private Address getRandomNotConnectedPeerAddress() { private Address getRandomNotConnectedPeerAddress() {
List<Address> list = getNotConnectedPeerAddresses(); List<Address> list = new ArrayList<>(getNotConnectedPeerAddresses());
if (list.size() > 0) { if (list.size() > 0) {
Collections.shuffle(list); Collections.shuffle(list);
return list.get(0); return list.get(0);
@ -591,7 +588,6 @@ public class PeerGroup {
if (disconnectedPeer != null) if (disconnectedPeer != null)
UserThread.execute(() -> peerListeners.stream().forEach(e -> e.onPeerRemoved(peerAddress))); UserThread.execute(() -> peerListeners.stream().forEach(e -> e.onPeerRemoved(peerAddress)));
log.trace("removePeer [post]");
printConnectedPeersMap(); printConnectedPeersMap();
printReportedPeersMap(); printReportedPeersMap();

650
network/src/main/java/io/bitsquare/p2p/routing/Routing.java
View file

@ -1,650 +0,0 @@
package io.bitsquare.p2p.routing;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
import io.bitsquare.common.UserThread;
import io.bitsquare.common.util.Utilities;
import io.bitsquare.p2p.Address;
import io.bitsquare.p2p.network.*;
import io.bitsquare.p2p.routing.messages.*;
import io.bitsquare.p2p.storage.messages.BroadcastMessage;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
public class Routing {
private static final Logger log = LoggerFactory.getLogger(Routing.class);
private static int simulateAuthTorNode = 0;
public static void setSimulateAuthTorNode(int simulateAuthTorNode) {
Routing.simulateAuthTorNode = simulateAuthTorNode;
}
private static int MAX_CONNECTIONS = 8;
private static int MAINTENANCE_INTERVAL = new Random().nextInt(15 * 60 * 1000) + 15 * 60 * 1000; // 15-30 min.
private static int PING_AFTER_CONNECTION_INACTIVITY = 5 * 60 * 1000; // 5 min
private long startAuthTs;
public static void setMaxConnections(int maxConnections) {
MAX_CONNECTIONS = maxConnections;
}
private final NetworkNode networkNode;
private final List<Address> seedNodes;
private final Map<Address, Long> nonceMap = new ConcurrentHashMap<>();
private final List<RoutingListener> routingListeners = new CopyOnWriteArrayList<>();
private final Map<Address, Peer> authenticatedPeers = new ConcurrentHashMap<>();
private final Set<Address> reportedPeerAddresses = new CopyOnWriteArraySet<>();
private final Map<Address, Runnable> authenticationCompleteHandlers = new ConcurrentHashMap<>();
private final Timer maintenanceTimer = new Timer();
private volatile boolean shutDownInProgress;
///////////////////////////////////////////////////////////////////////////////////////////
// Constructor
///////////////////////////////////////////////////////////////////////////////////////////
public Routing(final NetworkNode networkNode, List<Address> seeds) {
this.networkNode = networkNode;
// We copy it as we remove ourselves later from the list if we are a seed node
this.seedNodes = new CopyOnWriteArrayList<>(seeds);
init(networkNode);
}
private void init(NetworkNode networkNode) {
networkNode.addMessageListener((message, connection) -> {
if (message instanceof AuthenticationMessage)
processAuthenticationMessage((AuthenticationMessage) message, connection);
else if (message instanceof MaintenanceMessage)
processMaintenanceMessage((MaintenanceMessage) message, connection);
});
networkNode.addConnectionListener(new ConnectionListener() {
@Override
public void onConnection(Connection connection) {
}
@Override
public void onPeerAddressAuthenticated(Address peerAddress, Connection connection) {
}
@Override
public void onDisconnect(Reason reason, Connection connection) {
// only removes authenticated nodes
if (connection.isAuthenticated())
removePeer(connection.getPeerAddress());
}
@Override
public void onError(Throwable throwable) {
}
});
networkNode.addSetupListener(new SetupListener() {
@Override
public void onTorNodeReady() {
}
@Override
public void onHiddenServiceReady() {
// remove ourselves in case we are a seed node
Address myAddress = getAddress();
if (myAddress != null)
seedNodes.remove(myAddress);
}
@Override
public void onSetupFailed(Throwable throwable) {
}
});
maintenanceTimer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
Thread.currentThread().setName("MaintenanceTimer-" + new Random().nextInt(1000));
try {
UserThread.execute(() -> {
disconnectOldConnections();
pingPeers();
});
} catch (Throwable t) {
t.printStackTrace();
log.error("Executing task failed. " + t.getMessage());
}
}
}, MAINTENANCE_INTERVAL, MAINTENANCE_INTERVAL);
}
private void disconnectOldConnections() {
List<Connection> authenticatedConnections = networkNode.getAllConnections().stream()
.filter(e -> e.isAuthenticated())
.collect(Collectors.toList());
if (authenticatedConnections.size() > MAX_CONNECTIONS) {
authenticatedConnections.sort((o1, o2) -> o1.getLastActivityDate().compareTo(o2.getLastActivityDate()));
log.info("Number of connections exceeds MAX_CONNECTIONS. Current size=" + authenticatedConnections.size());
Connection connection = authenticatedConnections.remove(0);
log.info("Shutdown oldest connection with last activity date=" + connection.getLastActivityDate() + " / connection=" + connection);
connection.shutDown(() -> Utilities.runTimerTask(() -> {
Thread.currentThread().setName("DelayDisconnectOldConnectionsTimer-" + new Random().nextInt(1000));
disconnectOldConnections();
}, 1, TimeUnit.MILLISECONDS));
}
}
private void pingPeers() {
log.trace("pingPeers");
List<Peer> connectedPeersList = new ArrayList<>(authenticatedPeers.values());
connectedPeersList.stream()
.filter(e -> (new Date().getTime() - e.connection.getLastActivityDate().getTime()) > PING_AFTER_CONNECTION_INACTIVITY)
.forEach(e -> Utilities.runTimerTaskWithRandomDelay(() -> {
Thread.currentThread().setName("DelayPingPeersTimer-" + new Random().nextInt(1000));
SettableFuture<Connection> future = networkNode.sendMessage(e.connection, new PingMessage(e.getPingNonce()));
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("PingMessage sent successfully");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.info("PingMessage sending failed " + throwable.getMessage());
removePeer(e.address);
}
});
}, 5, 10));
}
///////////////////////////////////////////////////////////////////////////////////////////
// API
///////////////////////////////////////////////////////////////////////////////////////////
public void shutDown() {
if (!shutDownInProgress) {
shutDownInProgress = true;
if (maintenanceTimer != null)
maintenanceTimer.cancel();
}
}
public void broadcast(BroadcastMessage message, @Nullable Address sender) {
log.trace("Broadcast message to " + authenticatedPeers.values().size() + " peers.");
log.trace("message = " + message);
printConnectedPeersMap();
authenticatedPeers.values().stream()
.filter(e -> !e.address.equals(sender))
.forEach(peer -> {
log.trace("Broadcast message from " + getAddress() + " to " + peer.address + ".");
SettableFuture<Connection> future = networkNode.sendMessage(peer.address, message);
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("Broadcast from " + getAddress() + " to " + peer.address + " succeeded.");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.info("Broadcast failed. " + throwable.getMessage());
removePeer(peer.address);
}
});
});
}
public void addMessageListener(MessageListener messageListener) {
networkNode.addMessageListener(messageListener);
}
public void removeMessageListener(MessageListener messageListener) {
networkNode.removeMessageListener(messageListener);
}
public void addRoutingListener(RoutingListener routingListener) {
routingListeners.add(routingListener);
}
public void removeRoutingListener(RoutingListener routingListener) {
routingListeners.remove(routingListener);
}
public Map<Address, Peer> getAuthenticatedPeers() {
return authenticatedPeers;
}
// Use ArrayList not List as we need it serializable
public ArrayList<Address> getAllPeerAddresses() {
ArrayList<Address> allPeerAddresses = new ArrayList<>(reportedPeerAddresses);
allPeerAddresses.addAll(authenticatedPeers.values().stream()
.map(e -> e.address).collect(Collectors.toList()));
// remove own address and seed nodes
allPeerAddresses.remove(getAddress());
return allPeerAddresses;
}
///////////////////////////////////////////////////////////////////////////////////////////
// Authentication
///////////////////////////////////////////////////////////////////////////////////////////
// authentication example:
// node2 -> node1 RequestAuthenticationMessage
// node1: close connection
// node1 -> node2 ChallengeMessage on new connection
// node2: authentication to node1 done if nonce ok
// node2 -> node1 GetPeersMessage
// node1: authentication to node2 done if nonce ok
// node1 -> node2 PeersMessage
public void startAuthentication(Set<Address> connectedSeedNodes) {
connectedSeedNodes.forEach(connectedSeedNode -> {
sendRequestAuthenticationMessage(seedNodes, connectedSeedNode);
});
}
private void sendRequestAuthenticationMessage(final List<Address> remainingSeedNodes, final Address address) {
log.info("We try to authenticate to a random seed node. " + address);
startAuthTs = System.currentTimeMillis();
final boolean[] alreadyConnected = {false};
authenticatedPeers.values().stream().forEach(e -> {
remainingSeedNodes.remove(e.address);
if (address.equals(e.address))
alreadyConnected[0] = true;
});
if (!alreadyConnected[0]) {
long nonce = addToMapAndGetNonce(address);
SettableFuture<Connection> future = networkNode.sendMessage(address, new RequestAuthenticationMessage(getAddress(), nonce));
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(@Nullable Connection connection) {
log.info("send RequestAuthenticationMessage to " + address + " succeeded.");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.info("Send RequestAuthenticationMessage to " + address + " failed. Exception:" + throwable.getMessage());
log.trace("We try to authenticate to another random seed nodes of that list: " + remainingSeedNodes);
getNextSeedNode(remainingSeedNodes);
}
});
} else {
getNextSeedNode(remainingSeedNodes);
}
}
private void getNextSeedNode(List<Address> remainingSeedNodes) {
List<Address> remainingSeedNodeAddresses = new CopyOnWriteArrayList<>(remainingSeedNodes);
Address myAddress = getAddress();
if (myAddress != null)
remainingSeedNodeAddresses.remove(myAddress);
if (!remainingSeedNodeAddresses.isEmpty()) {
Collections.shuffle(remainingSeedNodeAddresses);
Address address = remainingSeedNodeAddresses.remove(0);
sendRequestAuthenticationMessage(remainingSeedNodeAddresses, address);
} else {
log.info("No other seed node found. That is expected for the first seed node.");
}
}
private void processAuthenticationMessage(AuthenticationMessage message, Connection connection) {
log.trace("processAuthenticationMessage " + message + " from " + connection.getPeerAddress() + " at " + getAddress());
if (message instanceof RequestAuthenticationMessage) {
RequestAuthenticationMessage requestAuthenticationMessage = (RequestAuthenticationMessage) message;
Address peerAddress = requestAuthenticationMessage.address;
log.trace("RequestAuthenticationMessage from " + peerAddress + " at " + getAddress());
connection.shutDown(() -> Utilities.runTimerTask(() -> {
Thread.currentThread().setName("DelaySendChallengeMessageTimer-" + new Random().nextInt(1000));
// we delay a bit as listeners for connection.onDisconnect are on other threads and might lead to
// inconsistent state (removal of connection from NetworkNode.authenticatedConnections)
log.trace("processAuthenticationMessage: connection.shutDown complete. RequestAuthenticationMessage from " + peerAddress + " at " + getAddress());
long nonce = addToMapAndGetNonce(peerAddress);
SettableFuture<Connection> future = networkNode.sendMessage(peerAddress, new ChallengeMessage(getAddress(), requestAuthenticationMessage.nonce, nonce));
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.debug("onSuccess sending ChallengeMessage");
}
@Override
public void onFailure(Throwable throwable) {
log.warn("onFailure sending ChallengeMessage. We try again.");
SettableFuture<Connection> future = networkNode.sendMessage(peerAddress, new ChallengeMessage(getAddress(), requestAuthenticationMessage.nonce, nonce));
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.debug("onSuccess sending 2. ChallengeMessage");
}
@Override
public void onFailure(Throwable throwable) {
log.warn("onFailure sending ChallengeMessage. We give up.");
}
});
}
});
},
100 + simulateAuthTorNode,
TimeUnit.MILLISECONDS));
} else if (message instanceof ChallengeMessage) {
ChallengeMessage challengeMessage = (ChallengeMessage) message;
Address peerAddress = challengeMessage.address;
log.trace("ChallengeMessage from " + peerAddress + " at " + getAddress());
HashMap<Address, Long> tempNonceMap = new HashMap<>(nonceMap);
boolean verified = verifyNonceAndAuthenticatePeerAddress(challengeMessage.requesterNonce, peerAddress);
if (verified) {
connection.setPeerAddress(peerAddress);
SettableFuture<Connection> future = networkNode.sendMessage(peerAddress,
new GetPeersMessage(getAddress(), challengeMessage.challengerNonce, getAllPeerAddresses()));
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("GetPeersMessage sent successfully from " + getAddress() + " to " + peerAddress);
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.info("GetPeersMessage sending failed " + throwable.getMessage());
removePeer(peerAddress);
}
});
} else {
log.warn("verifyNonceAndAuthenticatePeerAddress failed. challengeMessage=" + challengeMessage + " / nonceMap=" + tempNonceMap);
}
} else if (message instanceof GetPeersMessage) {
GetPeersMessage getPeersMessage = (GetPeersMessage) message;
Address peerAddress = getPeersMessage.address;
log.trace("GetPeersMessage from " + peerAddress + " at " + getAddress());
boolean verified = verifyNonceAndAuthenticatePeerAddress(getPeersMessage.challengerNonce, peerAddress);
if (verified) {
setAuthenticated(connection, peerAddress);
purgeReportedPeers();
SettableFuture<Connection> future = networkNode.sendMessage(peerAddress,
new PeersMessage(getAddress(), getAllPeerAddresses()));
log.trace("sent PeersMessage to " + peerAddress + " from " + getAddress()
+ " with allPeers=" + getAllPeerAddresses());
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("PeersMessage sent successfully from " + getAddress() + " to " + peerAddress);
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.info("PeersMessage sending failed " + throwable.getMessage());
removePeer(peerAddress);
}
});
// now we add the reported peers to our own set
ArrayList<Address> peerAddresses = ((GetPeersMessage) message).peerAddresses;
log.trace("Received peers: " + peerAddresses);
// remove ourselves
addToReportedPeers(peerAddresses, connection);
}
} else if (message instanceof PeersMessage) {
PeersMessage peersMessage = (PeersMessage) message;
Address peerAddress = peersMessage.address;
log.trace("PeersMessage from " + peerAddress + " at " + getAddress());
ArrayList<Address> peerAddresses = peersMessage.peerAddresses;
log.trace("Received peers: " + peerAddresses);
// remove ourselves
addToReportedPeers(peerAddresses, connection);
// we wait until the handshake is completed before setting the authenticate flag
// authentication at both sides of the connection
log.info("\n\nAuthenticationComplete\nPeer with address " + peerAddress
+ " authenticated (" + connection.getObjectId() + "). Took "
+ (System.currentTimeMillis() - startAuthTs) + " ms. \n\n");
setAuthenticated(connection, peerAddress);
Runnable authenticationCompleteHandler = authenticationCompleteHandlers.remove(connection.getPeerAddress());
if (authenticationCompleteHandler != null)
authenticationCompleteHandler.run();
authenticateToNextRandomPeer();
}
}
private void addToReportedPeers(ArrayList<Address> peerAddresses, Connection connection) {
log.trace("addToReportedPeers");
// we disconnect misbehaving nodes trying to send too many peers
// reported peers include the peers connected peers which is normally max. 8 but we give some headroom
// for safety
if (peerAddresses.size() > 1100) {
connection.shutDown();
} else {
peerAddresses.remove(getAddress());
reportedPeerAddresses.addAll(peerAddresses);
purgeReportedPeers();
}
}
private void purgeReportedPeers() {
log.trace("purgeReportedPeers");
int all = getAllPeerAddresses().size();
if (all > 1000) {
int diff = all - 100;
List<Address> list = getNotConnectedPeerAddresses();
for (int i = 0; i < diff; i++) {
Address toRemove = list.remove(new Random().nextInt(list.size()));
reportedPeerAddresses.remove(toRemove);
}
}
}
private List<Address> getNotConnectedPeerAddresses() {
ArrayList<Address> list = new ArrayList<>(getAllPeerAddresses());
log.debug("## getNotConnectedPeerAddresses ");
log.debug("## reportedPeersList=" + list);
authenticatedPeers.values().stream().forEach(e -> list.remove(e.address));
log.debug("## connectedPeers=" + authenticatedPeers);
log.debug("## reportedPeersList=" + list);
return list;
}
private void authenticateToNextRandomPeer() {
Utilities.runTimerTaskWithRandomDelay(() -> {
Thread.currentThread().setName("DelayAuthenticateToNextRandomPeerTimer-" + new Random().nextInt(1000));
if (getAuthenticatedPeers().size() <= MAX_CONNECTIONS) {
Address randomNotConnectedPeerAddress = getRandomNotConnectedPeerAddress();
if (randomNotConnectedPeerAddress != null) {
log.info("We try to build an authenticated connection to a random peer. " + randomNotConnectedPeerAddress);
authenticateToPeer(randomNotConnectedPeerAddress, null, () -> authenticateToNextRandomPeer());
} else {
log.info("No more peers available for connecting.");
}
} else {
log.info("We have already enough connections.");
}
}, 200, 400, TimeUnit.MILLISECONDS);
}
public void authenticateToPeer(Address address, @Nullable Runnable authenticationCompleteHandler, @Nullable Runnable faultHandler) {
startAuthTs = System.currentTimeMillis();
if (authenticationCompleteHandler != null)
authenticationCompleteHandlers.put(address, authenticationCompleteHandler);
long nonce = addToMapAndGetNonce(address);
SettableFuture<Connection> future = networkNode.sendMessage(address, new RequestAuthenticationMessage(getAddress(), nonce));
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(@Nullable Connection connection) {
log.debug("send RequestAuthenticationMessage succeeded");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.info("send IdMessage failed. " + throwable.getMessage());
removePeer(address);
if (faultHandler != null) faultHandler.run();
}
});
}
private long addToMapAndGetNonce(Address peerAddress) {
long nonce = new Random().nextLong();
while (nonce == 0) {
nonce = new Random().nextLong();
}
log.trace("addToMapAndGetNonce nonceMap=" + nonceMap + " / peerAddress=" + peerAddress);
nonceMap.put(peerAddress, nonce);
return nonce;
}
private boolean verifyNonceAndAuthenticatePeerAddress(long peersNonce, Address peerAddress) {
log.trace("verifyNonceAndAuthenticatePeerAddress nonceMap=" + nonceMap + " / peerAddress=" + peerAddress);
Long nonce = nonceMap.remove(peerAddress);
return nonce != null && nonce == peersNonce;
}
private void setAuthenticated(Connection connection, Address peerAddress) {
log.info("\n\n############################################################\n" +
"We are authenticated to:" +
"\nconnection=" + connection
+ "\nmyAddress=" + getAddress()
+ "\npeerAddress= " + peerAddress
+ "\n############################################################\n");
connection.setAuthenticated(peerAddress, connection);
Peer peer = new Peer(connection);
addAuthenticatedPeer(peerAddress, peer);
routingListeners.stream().forEach(e -> e.onConnectionAuthenticated(connection));
log.debug("\n### setAuthenticated post connection " + connection);
}
private Address getRandomNotConnectedPeerAddress() {
List<Address> list = getNotConnectedPeerAddresses();
if (list.size() > 0) {
Collections.shuffle(list);
return list.get(0);
} else {
return null;
}
}
///////////////////////////////////////////////////////////////////////////////////////////
// Maintenance
///////////////////////////////////////////////////////////////////////////////////////////
private void processMaintenanceMessage(MaintenanceMessage message, Connection connection) {
log.debug("Received routing message " + message + " at " + getAddress() + " from " + connection.getPeerAddress());
if (message instanceof PingMessage) {
SettableFuture<Connection> future = networkNode.sendMessage(connection, new PongMessage(((PingMessage) message).nonce));
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("PongMessage sent successfully");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
log.info("PongMessage sending failed " + throwable.getMessage());
removePeer(connection.getPeerAddress());
}
});
} else if (message instanceof PongMessage) {
Peer peer = authenticatedPeers.get(connection.getPeerAddress());
if (peer != null) {
if (((PongMessage) message).nonce != peer.getPingNonce()) {
removePeer(peer.address);
log.warn("PongMessage invalid: self/peer " + getAddress() + "/" + connection.getPeerAddress());
}
}
}
}
///////////////////////////////////////////////////////////////////////////////////////////
// Peers
///////////////////////////////////////////////////////////////////////////////////////////
private void removePeer(@Nullable Address peerAddress) {
reportedPeerAddresses.remove(peerAddress);
Peer disconnectedPeer;
disconnectedPeer = authenticatedPeers.remove(peerAddress);
if (disconnectedPeer != null)
UserThread.execute(() -> routingListeners.stream().forEach(e -> e.onPeerRemoved(peerAddress)));
log.trace("removePeer [post]");
printConnectedPeersMap();
printReportedPeersMap();
log.trace("removePeer nonceMap=" + nonceMap + " / peerAddress=" + peerAddress);
nonceMap.remove(peerAddress);
}
private void addAuthenticatedPeer(Address address, Peer peer) {
boolean firstPeerAdded;
authenticatedPeers.put(address, peer);
firstPeerAdded = authenticatedPeers.size() == 1;
UserThread.execute(() -> routingListeners.stream().forEach(e -> e.onPeerAdded(peer)));
if (firstPeerAdded)
UserThread.execute(() -> routingListeners.stream().forEach(e -> e.onFirstPeerAdded(peer)));
if (authenticatedPeers.size() > MAX_CONNECTIONS)
disconnectOldConnections();
log.trace("addConnectedPeer [post]");
printConnectedPeersMap();
}
private Address getAddress() {
return networkNode.getAddress();
}
///////////////////////////////////////////////////////////////////////////////////////////
// Utils
///////////////////////////////////////////////////////////////////////////////////////////
public void printConnectedPeersMap() {
StringBuilder result = new StringBuilder("\nConnected peers for node " + getAddress() + ":");
authenticatedPeers.values().stream().forEach(e -> {
result.append("\n\t" + e.address);
});
result.append("\n");
log.info(result.toString());
}
public void printReportedPeersMap() {
StringBuilder result = new StringBuilder("\nReported peerAddresses for node " + getAddress() + ":");
reportedPeerAddresses.stream().forEach(e -> {
result.append("\n\t" + e);
});
result.append("\n");
log.info(result.toString());
}
private String getObjectId() {
return super.toString().split("@")[1].toString();
}
}

10
network/src/main/java/io/bitsquare/p2p/seed/SeedNode.java
View file

@ -11,15 +11,15 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.io.File; import java.io.File;
import java.util.ArrayList; import java.util.HashSet;
import java.util.List; import java.util.Set;
public class SeedNode { public class SeedNode {
private static final Logger log = LoggerFactory.getLogger(SeedNode.class); private static final Logger log = LoggerFactory.getLogger(SeedNode.class);
private int port = 8001; private int port = 8001;
private boolean useLocalhost = true; private boolean useLocalhost = true;
private List<Address> seedNodes; private Set<Address> seedNodes;
private P2PService p2PService; private P2PService p2PService;
protected boolean stopped; protected boolean stopped;
@ -41,7 +41,7 @@ public class SeedNode {
useLocalhost = ("true").equals(args[1]); useLocalhost = ("true").equals(args[1]);
if (args.length > 2) { if (args.length > 2) {
seedNodes = new ArrayList<>(); seedNodes = new HashSet<>();
for (int i = 2; i < args.length; i++) { for (int i = 2; i < args.length; i++) {
seedNodes.add(new Address(args[i])); seedNodes.add(new Address(args[i]));
} }
@ -54,7 +54,7 @@ public class SeedNode {
createAndStartP2PService(null, null, port, useLocalhost, seedNodes, null); createAndStartP2PService(null, null, port, useLocalhost, seedNodes, null);
} }
public void createAndStartP2PService(EncryptionService encryptionService, KeyRing keyRing, int port, boolean useLocalhost, @Nullable List<Address> seedNodes, @Nullable P2PServiceListener listener) { public void createAndStartP2PService(EncryptionService encryptionService, KeyRing keyRing, int port, boolean useLocalhost, @Nullable Set<Address> seedNodes, @Nullable P2PServiceListener listener) {
SeedNodesRepository seedNodesRepository = new SeedNodesRepository(); SeedNodesRepository seedNodesRepository = new SeedNodesRepository();
if (seedNodes != null && !seedNodes.isEmpty()) { if (seedNodes != null && !seedNodes.isEmpty()) {
if (useLocalhost) if (useLocalhost)

18
network/src/main/java/io/bitsquare/p2p/seed/SeedNodesRepository.java
View file

@ -1,41 +1,41 @@
package io.bitsquare.p2p.seed; package io.bitsquare.p2p.seed;
import com.google.common.collect.Sets;
import io.bitsquare.p2p.Address; import io.bitsquare.p2p.Address;
import java.util.Arrays; import java.util.Set;
import java.util.List;
public class SeedNodesRepository { public class SeedNodesRepository {
protected List<Address> torSeedNodeAddresses = Arrays.asList( protected Set<Address> torSeedNodeAddresses = Sets.newHashSet(
new Address("lmvdenjkyvx2ovga.onion:8001") new Address("lmvdenjkyvx2ovga.onion:8001")
); );
protected List<Address> localhostSeedNodeAddresses = Arrays.asList( protected Set<Address> localhostSeedNodeAddresses = Sets.newHashSet(
new Address("localhost:8001"), new Address("localhost:8001"),
new Address("localhost:8002"), new Address("localhost:8002"),
new Address("localhost:8003") new Address("localhost:8003")
); );
public List<Address> getTorSeedNodeAddresses() { public Set<Address> getTorSeedNodeAddresses() {
return torSeedNodeAddresses; return torSeedNodeAddresses;
} }
public List<Address> geSeedNodeAddresses(boolean useLocalhost) { public Set<Address> geSeedNodeAddresses(boolean useLocalhost) {
return useLocalhost ? localhostSeedNodeAddresses : torSeedNodeAddresses; return useLocalhost ? localhostSeedNodeAddresses : torSeedNodeAddresses;
} }
public List<Address> getLocalhostSeedNodeAddresses() { public Set<Address> getLocalhostSeedNodeAddresses() {
return localhostSeedNodeAddresses; return localhostSeedNodeAddresses;
} }
public void setTorSeedNodeAddresses(List<Address> torSeedNodeAddresses) { public void setTorSeedNodeAddresses(Set<Address> torSeedNodeAddresses) {
this.torSeedNodeAddresses = torSeedNodeAddresses; this.torSeedNodeAddresses = torSeedNodeAddresses;
} }
public void setLocalhostSeedNodeAddresses(List<Address> localhostSeedNodeAddresses) { public void setLocalhostSeedNodeAddresses(Set<Address> localhostSeedNodeAddresses) {
this.localhostSeedNodeAddresses = localhostSeedNodeAddresses; this.localhostSeedNodeAddresses = localhostSeedNodeAddresses;
} }
} }

9
network/src/main/java/io/bitsquare/p2p/storage/ProtectedExpirableDataStorage.java
View file

@ -20,9 +20,12 @@ import java.io.File;
import java.math.BigInteger; import java.math.BigInteger;
import java.security.KeyPair; import java.security.KeyPair;
import java.security.PublicKey; import java.security.PublicKey;
import java.util.*; import java.util.Map;
import java.util.Random;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet;
public class ProtectedExpirableDataStorage { public class ProtectedExpirableDataStorage {
private static final Logger log = LoggerFactory.getLogger(ProtectedExpirableDataStorage.class); private static final Logger log = LoggerFactory.getLogger(ProtectedExpirableDataStorage.class);
@ -32,7 +35,7 @@ public class ProtectedExpirableDataStorage {
private final PeerGroup peerGroup; private final PeerGroup peerGroup;
private final Map<BigInteger, ProtectedData> map = new ConcurrentHashMap<>(); private final Map<BigInteger, ProtectedData> map = new ConcurrentHashMap<>();
private final List<HashMapChangedListener> hashMapChangedListeners = new CopyOnWriteArrayList<>(); private final CopyOnWriteArraySet<HashMapChangedListener> hashMapChangedListeners = new CopyOnWriteArraySet<>();
private ConcurrentHashMap<BigInteger, Integer> sequenceNumberMap = new ConcurrentHashMap<>(); private ConcurrentHashMap<BigInteger, Integer> sequenceNumberMap = new ConcurrentHashMap<>();
private final Storage<ConcurrentHashMap> storage; private final Storage<ConcurrentHashMap> storage;
private boolean authenticated; private boolean authenticated;

8
network/src/test/java/io/bitsquare/p2p/P2PServiceTest.java
View file

@ -22,7 +22,8 @@ import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.security.*; import java.security.*;
import java.security.cert.CertificateException; import java.security.cert.CertificateException;
import java.util.ArrayList; import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
// TorNode created. Took 6 sec. // TorNode created. Took 6 sec.
@ -30,12 +31,13 @@ import java.util.concurrent.CountDownLatch;
// Connection establishment takes about 4 sec. // Connection establishment takes about 4 sec.
// need to define seed node addresses first before using tor version // need to define seed node addresses first before using tor version
// Ignored for automated tests
@Ignore @Ignore
public class P2PServiceTest { public class P2PServiceTest {
private static final Logger log = LoggerFactory.getLogger(P2PServiceTest.class); private static final Logger log = LoggerFactory.getLogger(P2PServiceTest.class);
boolean useLocalhost = true; boolean useLocalhost = true;
private ArrayList<Address> seedNodes; private Set<Address> seedNodes;
private int sleepTime; private int sleepTime;
private KeyRing keyRing1, keyRing2, keyRing3; private KeyRing keyRing1, keyRing2, keyRing3;
private EncryptionService encryptionService1, encryptionService2, encryptionService3; private EncryptionService encryptionService1, encryptionService2, encryptionService3;
@ -66,7 +68,7 @@ public class P2PServiceTest {
encryptionService2 = new EncryptionService(keyRing2); encryptionService2 = new EncryptionService(keyRing2);
encryptionService3 = new EncryptionService(keyRing3); encryptionService3 = new EncryptionService(keyRing3);
seedNodes = new ArrayList<>(); seedNodes = new HashSet<>();
if (useLocalhost) { if (useLocalhost) {
seedNodes.add(new Address("localhost:8001")); seedNodes.add(new Address("localhost:8001"));
seedNodes.add(new Address("localhost:8002")); seedNodes.add(new Address("localhost:8002"));

6
network/src/test/java/io/bitsquare/p2p/TestUtils.java
View file

@ -9,7 +9,7 @@ import org.slf4j.LoggerFactory;
import java.io.*; import java.io.*;
import java.security.*; import java.security.*;
import java.util.ArrayList; import java.util.Set;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
public class TestUtils { public class TestUtils {
@ -62,7 +62,7 @@ public class TestUtils {
return result; return result;
} }
public static SeedNode getAndStartSeedNode(int port, EncryptionService encryptionService, KeyRing keyRing, boolean useLocalhost, ArrayList<Address> seedNodes) throws InterruptedException { public static SeedNode getAndStartSeedNode(int port, EncryptionService encryptionService, KeyRing keyRing, boolean useLocalhost, Set<Address> seedNodes) throws InterruptedException {
SeedNode seedNode; SeedNode seedNode;
if (useLocalhost) { if (useLocalhost) {
@ -107,7 +107,7 @@ public class TestUtils {
return seedNode; return seedNode;
} }
public static P2PService getAndAuthenticateP2PService(int port, EncryptionService encryptionService, KeyRing keyRing, boolean useLocalhost, ArrayList<Address> seedNodes) throws InterruptedException { public static P2PService getAndAuthenticateP2PService(int port, EncryptionService encryptionService, KeyRing keyRing, boolean useLocalhost, Set<Address> seedNodes) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
SeedNodesRepository seedNodesRepository = new SeedNodesRepository(); SeedNodesRepository seedNodesRepository = new SeedNodesRepository();
if (seedNodes != null && !seedNodes.isEmpty()) { if (seedNodes != null && !seedNodes.isEmpty()) {

11
network/src/test/java/io/bitsquare/p2p/routing/PeerGroupTest.java
View file

@ -13,7 +13,8 @@ import org.junit.*;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.ArrayList; import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
// TorNode created. Took 6 sec. // TorNode created. Took 6 sec.
@ -27,7 +28,7 @@ public class PeerGroupTest {
boolean useLocalhost = true; boolean useLocalhost = true;
private CountDownLatch latch; private CountDownLatch latch;
private ArrayList<Address> seedNodes; private Set<Address> seedNodes;
private int sleepTime; private int sleepTime;
private SeedNode seedNode1, seedNode2, seedNode3; private SeedNode seedNode1, seedNode2, seedNode3;
@ -37,7 +38,7 @@ public class PeerGroupTest {
LocalhostNetworkNode.setSimulateTorDelayHiddenService(8); LocalhostNetworkNode.setSimulateTorDelayHiddenService(8);
PeerGroup.setMaxConnections(100); PeerGroup.setMaxConnections(100);
seedNodes = new ArrayList<>(); seedNodes = new HashSet<>();
if (useLocalhost) { if (useLocalhost) {
//seedNodes.add(new Address("localhost:8001")); //seedNodes.add(new Address("localhost:8001"));
// seedNodes.add(new Address("localhost:8002")); // seedNodes.add(new Address("localhost:8002"));
@ -77,7 +78,7 @@ public class PeerGroupTest {
public void testSingleSeedNode() throws InterruptedException { public void testSingleSeedNode() throws InterruptedException {
LocalhostNetworkNode.setSimulateTorDelayTorNode(0); LocalhostNetworkNode.setSimulateTorDelayTorNode(0);
LocalhostNetworkNode.setSimulateTorDelayHiddenService(0); LocalhostNetworkNode.setSimulateTorDelayHiddenService(0);
seedNodes = new ArrayList<>(); seedNodes = new HashSet<>();
seedNodes.add(new Address("localhost:8001")); seedNodes.add(new Address("localhost:8001"));
seedNode1 = new SeedNode(); seedNode1 = new SeedNode();
latch = new CountDownLatch(2); latch = new CountDownLatch(2);
@ -116,7 +117,7 @@ public class PeerGroupTest {
public void test2SeedNodes() throws InterruptedException { public void test2SeedNodes() throws InterruptedException {
LocalhostNetworkNode.setSimulateTorDelayTorNode(0); LocalhostNetworkNode.setSimulateTorDelayTorNode(0);
LocalhostNetworkNode.setSimulateTorDelayHiddenService(0); LocalhostNetworkNode.setSimulateTorDelayHiddenService(0);
seedNodes = new ArrayList<>(); seedNodes = new HashSet<>();
seedNodes.add(new Address("localhost:8001")); seedNodes.add(new Address("localhost:8001"));
seedNodes.add(new Address("localhost:8002")); seedNodes.add(new Address("localhost:8002"));

5
network/src/test/java/io/bitsquare/p2p/storage/ProtectedDataStorageTest.java
View file

@ -24,8 +24,9 @@ import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.security.*; import java.security.*;
import java.security.cert.CertificateException; import java.security.cert.CertificateException;
import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
@ -34,7 +35,7 @@ public class ProtectedDataStorageTest {
private static final Logger log = LoggerFactory.getLogger(ProtectedDataStorageTest.class); private static final Logger log = LoggerFactory.getLogger(ProtectedDataStorageTest.class);
boolean useClearNet = true; boolean useClearNet = true;
private ArrayList<Address> seedNodes = new ArrayList<>(); private Set<Address> seedNodes = new HashSet<>();
private NetworkNode networkNode1; private NetworkNode networkNode1;
private PeerGroup peerGroup1; private PeerGroup peerGroup1;
private EncryptionService encryptionService1, encryptionService2; private EncryptionService encryptionService1, encryptionService2;