Git-Mirrors/haveno
1
0
Fork 0
mirror of https://github.com/haveno-dex/haveno.git synced 2025-04-20 23:56:30 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add KeepAlive handling, add statistics, add table in network view for statistics, reduce connection timeout

Browse Source
This commit is contained in:
Manfred Karrer 2016-02-15 18:28:29 +01:00
parent 54a0fe9cc0
commit 540f49f1e1
31 changed files with 748 additions and 711 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
gui/src/main/java/io/bitsquare/gui/main/MainViewModel.java
View File

@ -196,11 +196,11 @@ public class MainViewModel implements ViewModel {
onAllServicesInitialized();
});
startupTimeout = FxTimer.runLater(Duration.ofMillis(60000), () -> {
startupTimeout = FxTimer.runLater(Duration.ofMinutes(3), () -> {
log.warn("startupTimeout called");
MainView.blur();
new Popup().warning("The application could not startup after 60 seconds.\n" +
"There might be some network connection problems.\n\n" +
new Popup().warning("The application could not startup after 3 minutes.\n" +
"There might be some network connection problems or a unstable Tor path.\n\n" +
"Please restart and try again.")
.closeButtonText("Shut down")
.onClose(BitsquareApp.shutDownHandler::run)

47
gui/src/main/java/io/bitsquare/gui/main/settings/network/NetworkSettingsView.fxml
View File

@ -19,6 +19,7 @@
<?import io.bitsquare.gui.components.TitledGroupBg?>
<?import javafx.geometry.*?>
<?import javafx.scene.control.cell.*?>
<?import javafx.scene.control.*?>
<?import javafx.scene.layout.*?>
<GridPane fx:id="root" fx:controller="io.bitsquare.gui.main.settings.network.NetworkSettingsView"
@ -65,12 +66,50 @@
</TextField>
<Label fx:id="p2PPeersLabel" text="Connected peers:" GridPane.rowIndex="4"/>
<TextArea fx:id="p2PPeersTextArea" GridPane.rowIndex="4" GridPane.columnIndex="1" GridPane.hgrow="ALWAYS"
GridPane.vgrow="ALWAYS" editable="false" focusTraversable="false"/>
<TableView fx:id="p2PPeerTable" GridPane.rowIndex="4" GridPane.columnIndex="1" VBox.vgrow="ALWAYS"
GridPane.hgrow="ALWAYS">
<columns>
<TableColumn text="Onion address" fx:id="onionAddressColumn" minWidth="220">
<cellValueFactory>
<PropertyValueFactory property="onionAddress"/>
</cellValueFactory>
</TableColumn>
<TableColumn text="Type" fx:id="connectionTypeColumn" minWidth="70" maxWidth="80">
<cellValueFactory>
<PropertyValueFactory property="connectionType"/>
</cellValueFactory>
</TableColumn>
<TableColumn text="Established" fx:id="creationDateColumn" minWidth="180" maxWidth="180">
<cellValueFactory>
<PropertyValueFactory property="creationDate"/>
</cellValueFactory>
</TableColumn>
<TableColumn text="Last activity" fx:id="lastActivityColumn" minWidth="100" maxWidth="120">
<cellValueFactory>
<PropertyValueFactory property="lastActivity"/>
</cellValueFactory>
</TableColumn>
<TableColumn text="Sent" fx:id="sentBytesColumn" minWidth="100" maxWidth="120">
<cellValueFactory>
<PropertyValueFactory property="sentBytes"/>
</cellValueFactory>
</TableColumn>
<TableColumn text="Received" fx:id="receivedBytesColumn" minWidth="100" maxWidth="120">
<cellValueFactory>
<PropertyValueFactory property="receivedBytes"/>
</cellValueFactory>
</TableColumn>
<TableColumn text="Identity" fx:id="peerTypeColumn" minWidth="100" maxWidth="100">
<cellValueFactory>
<PropertyValueFactory property="peerType"/>
</cellValueFactory>
</TableColumn>
</columns>
</TableView>
<columnConstraints>
<ColumnConstraints hgrow="SOMETIMES" halignment="RIGHT" minWidth="200.0"/>
<ColumnConstraints hgrow="ALWAYS" minWidth="300.0"/>
<ColumnConstraints hgrow="NEVER" halignment="RIGHT"/>
<ColumnConstraints hgrow="ALWAYS"/>
</columnConstraints>
</GridPane>

61
gui/src/main/java/io/bitsquare/gui/main/settings/network/NetworkSettingsView.java
View File

@ -29,8 +29,6 @@ import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.P2PService;
import io.bitsquare.p2p.P2PServiceListener;
import io.bitsquare.p2p.network.LocalhostNetworkNode;
import io.bitsquare.p2p.network.OutboundConnection;
import io.bitsquare.p2p.seed.SeedNodesRepository;
import io.bitsquare.user.Preferences;
import javafx.beans.value.ChangeListener;
import javafx.collections.FXCollections;
@ -46,7 +44,6 @@ import org.reactfx.util.FxTimer;
import javax.inject.Inject;
import java.time.Duration;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
@FxmlView
@ -63,20 +60,26 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
@FXML
ComboBox<BitcoinNetwork> netWorkComboBox;
@FXML
TextArea bitcoinPeersTextArea, p2PPeersTextArea;
TextArea bitcoinPeersTextArea;
@FXML
Label bitcoinPeersLabel, p2PPeersLabel;
@FXML
CheckBox useTorCheckBox;
@FXML
TableView<NetworkStatisticListItem> p2PPeerTable;
@FXML
TableColumn<NetworkStatisticListItem, String> onionAddressColumn, connectionTypeColumn, creationDateColumn,
lastActivityColumn, sentBytesColumn, receivedBytesColumn, peerTypeColumn;
/* TableColumn<NetworkStatisticListItem, NetworkStatisticListItem> onionAddressColumn, connectionTypeColumn, creationDateColumn,
lastActivityColumn, sentBytesColumn, receivedBytesColumn, peerTypeColumn;
*/
private P2PServiceListener p2PServiceListener;
private ChangeListener<Number> numP2PPeersChangeListener;
private ChangeListener<List<Peer>> bitcoinPeersChangeListener;
private final Set<NodeAddress> seedNodeAddresses;
@Inject
public NetworkSettingsView(WalletService walletService, P2PService p2PService, Preferences preferences,
SeedNodesRepository seedNodesRepository, BSFormatter formatter) {
BSFormatter formatter) {
this.walletService = walletService;
this.p2PService = p2PService;
this.preferences = preferences;
@ -84,7 +87,6 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
BitcoinNetwork bitcoinNetwork = preferences.getBitcoinNetwork();
boolean useLocalhost = p2PService.getNetworkNode() instanceof LocalhostNetworkNode;
this.seedNodeAddresses = seedNodesRepository.getSeedNodeAddresses(useLocalhost, bitcoinNetwork.ordinal());
}
public void initialize() {
@ -138,6 +140,19 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
public void onSetupFailed(Throwable throwable) {
}
};
p2PPeerTable.setColumnResizePolicy(TableView.CONSTRAINED_RESIZE_POLICY);
p2PPeerTable.setPlaceholder(new Label("No connections are available"));
p2PPeerTable.getSortOrder().add(creationDateColumn);
creationDateColumn.setSortType(TableColumn.SortType.ASCENDING);
//TODO sorting needs other NetworkStatisticListItem as columns type
/* creationDateColumn.setComparator((o1, o2) ->
o1.statistic.getCreationDate().compareTo(o2.statistic.getCreationDate()));
sentBytesColumn.setComparator((o1, o2) ->
((Integer) o1.statistic.getSentBytes()).compareTo(((Integer) o2.statistic.getSentBytes())));
receivedBytesColumn.setComparator((o1, o2) ->
((Integer) o1.statistic.getReceivedBytes()).compareTo(((Integer) o2.statistic.getReceivedBytes())));*/
}
@Override
@ -171,9 +186,9 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
walletService.connectedPeersProperty().addListener(bitcoinPeersChangeListener);
updateBitcoinPeersTextArea();
numP2PPeersChangeListener = (observable, oldValue, newValue) -> updateP2PPeersTextArea();
numP2PPeersChangeListener = (observable, oldValue, newValue) -> updateP2PStatistics();
p2PService.getNumConnectedPeers().addListener(numP2PPeersChangeListener);
updateP2PPeersTextArea();
updateP2PStatistics();
}
@Override
@ -188,24 +203,18 @@ public class NetworkSettingsView extends ActivatableViewAndModel<GridPane, Activ
if (numP2PPeersChangeListener != null)
p2PService.getNumConnectedPeers().removeListener(numP2PPeersChangeListener);
p2PPeerTable.getItems().forEach(NetworkStatisticListItem::cleanup);
}
private void updateP2PPeersTextArea() {
p2PPeersTextArea.clear();
p2PPeersTextArea.setText(p2PService.getNetworkNode().getConfirmedConnections()
.stream()
.map(connection -> {
if (connection.getPeersNodeAddressOptional().isPresent()) {
NodeAddress nodeAddress = connection.getPeersNodeAddressOptional().get();
return nodeAddress.getFullAddress() + " (" +
(connection instanceof OutboundConnection ? "outbound" : "inbound") +
(seedNodeAddresses.contains(nodeAddress) ? " / seed node)" : ")");
} else {
// Should never be the case
return "";
}
})
.collect(Collectors.joining("\n")));
private void updateP2PStatistics() {
p2PPeerTable.getItems().forEach(NetworkStatisticListItem::cleanup);
List<NetworkStatisticListItem> list = p2PService.getNetworkNode().getConfirmedConnections().stream()
.map(connection -> new NetworkStatisticListItem(connection, formatter))
.collect(Collectors.toList());
p2PPeerTable.setItems(FXCollections.observableArrayList(list));
p2PPeerTable.sort();
}
private void updateBitcoinPeersTextArea() {

123
gui/src/main/java/io/bitsquare/gui/main/settings/network/NetworkStatisticListItem.java Normal file
View File

@ -0,0 +1,123 @@
/*
* This file is part of Bitsquare.
*
* Bitsquare is free software: you can redistribute it and/or modify it
* under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or (at
* your option) any later version.
*
* Bitsquare is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
* License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with Bitsquare. If not, see <http://www.gnu.org/licenses/>.
*/
package io.bitsquare.gui.main.settings.network;
import io.bitsquare.common.UserThread;
import io.bitsquare.gui.util.BSFormatter;
import io.bitsquare.p2p.network.Connection;
import io.bitsquare.p2p.network.OutboundConnection;
import io.bitsquare.p2p.network.Statistic;
import javafx.beans.property.SimpleStringProperty;
import javafx.beans.property.StringProperty;
import org.apache.commons.lang3.time.DurationFormatUtils;
import org.fxmisc.easybind.EasyBind;
import org.fxmisc.easybind.Subscription;
import org.reactfx.util.FxTimer;
import org.reactfx.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
public class NetworkStatisticListItem {
private static final Logger log = LoggerFactory.getLogger(NetworkStatisticListItem.class);
final Statistic statistic;
private final Connection connection;
private final Subscription sentBytesSubscription, receivedBytesSubscription;
private final Timer timer;
private BSFormatter formatter;
private StringProperty lastActivity = new SimpleStringProperty();
private StringProperty sentBytes = new SimpleStringProperty();
private StringProperty receivedBytes = new SimpleStringProperty();
public NetworkStatisticListItem(Connection connection, BSFormatter formatter) {
this.connection = connection;
this.formatter = formatter;
this.statistic = connection.getStatistic();
sentBytesSubscription = EasyBind.subscribe(statistic.sentBytesProperty,
e -> sentBytes.set(formatter.formatBytes((int) e)));
receivedBytesSubscription = EasyBind.subscribe(statistic.receivedBytesProperty,
e -> receivedBytes.set(formatter.formatBytes((int) e)));
timer = FxTimer.runPeriodically(Duration.ofMillis(1000),
() -> UserThread.execute(() -> onLastActivityChanged(statistic.lastActivityTimestampProperty.get())));
onLastActivityChanged(statistic.lastActivityTimestampProperty.get());
}
private void onLastActivityChanged(long timeStamp) {
lastActivity.set(DurationFormatUtils.formatDuration(System.currentTimeMillis() - timeStamp, "mm:ss.SSS"));
}
public void cleanup() {
sentBytesSubscription.unsubscribe();
receivedBytesSubscription.unsubscribe();
timer.stop();
}
public String getOnionAddress() {
if (connection.getPeersNodeAddressOptional().isPresent())
return connection.getPeersNodeAddressOptional().get().getFullAddress();
else
return "";
}
public String getConnectionType() {
return connection instanceof OutboundConnection ? "outbound" : "inbound";
}
public String getCreationDate() {
return formatter.formatDateTime(statistic.getCreationDate());
}
public String getPeerType() {
if (connection.getPeerType() == Connection.PeerType.SEED_NODE)
return "Seed node";
else if (connection.getPeerType() == Connection.PeerType.DIRECT_MSG_PEER)
return "Peer (direct)";
else
return "Peer";
}
public String getLastActivity() {
return lastActivity.get();
}
public StringProperty lastActivityProperty() {
return lastActivity;
}
public String getSentBytes() {
return sentBytes.get();
}
public StringProperty sentBytesProperty() {
return sentBytes;
}
public String getReceivedBytes() {
return receivedBytes.get();
}
public StringProperty receivedBytesProperty() {
return receivedBytes;
}
}

75
gui/src/main/java/io/bitsquare/gui/util/BSFormatter.java
View File

@ -23,6 +23,7 @@ import io.bitsquare.locale.LanguageUtil;
import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.trade.offer.Offer;
import io.bitsquare.user.Preferences;
import org.apache.commons.lang3.time.DurationFormatUtils;
import org.bitcoinj.core.Coin;
import org.bitcoinj.utils.Fiat;
import org.bitcoinj.utils.MonetaryFormat;
@ -374,8 +375,8 @@ public class BSFormatter {
}
public String getDaysHoursMinutes(Date startDate, Date endDate) {
long different = endDate.getTime() - startDate.getTime();
long secondsInMilli = 1000;
return DurationFormatUtils.formatDurationWords(endDate.getTime() - startDate.getTime(), true, true);
/* long secondsInMilli = 1000;
long minutesInMilli = secondsInMilli * 60;
long hoursInMilli = minutesInMilli * 60;
long daysInMilli = hoursInMilli * 24;
@ -394,9 +395,64 @@ public class BSFormatter {
else if (elapsedMinutes > 0)
return elapsedMinutes + " " + minuteString;
else
return null;
return null;*/
}
/* public List<String> tokenizePeriod(long time, boolean showSeconds, boolean showMillis) {
log.error("formatPastTimestamp " + time + "");
long secondsInMilli = 1000;
long minutesInMilli = secondsInMilli * 60;
long hoursInMilli = minutesInMilli * 60;
long daysInMilli = hoursInMilli * 24;
long elapsedDays = time / daysInMilli;
time = time % daysInMilli;
long elapsedHours = time / hoursInMilli;
time = time % hoursInMilli;
long elapsedMinutes = time / minutesInMilli;
time = time % minutesInMilli;
long elapsedSeconds = time / secondsInMilli;
time = time % secondsInMilli;
long elapsedMillis = time;
String dayString = elapsedDays == 1 ? elapsedDays + " day" : elapsedDays + " days";
String hourString = elapsedHours == 1 ? elapsedHours + " hour" : elapsedHours + " hours";
String minuteString = elapsedMinutes == 1 ? elapsedMinutes + " minute" : elapsedMinutes + " minutes";
String secondsString = elapsedSeconds == 1 ? elapsedSeconds + " second" : elapsedSeconds + " seconds";
String millisString = elapsedMillis + " ms";
List<String> tokens = new ArrayList<>();
if (elapsedDays > 0)
tokens.add(dayString);
if (elapsedHours > 0)
tokens.add(hourString);
if (elapsedMinutes > 0)
tokens.add(minuteString);
if (showSeconds && elapsedSeconds > 0)
tokens.add(secondsString);
if (showMillis && elapsedMillis > 0)
tokens.add(millisString);
return tokens;
}
private String formatTimeTokens(List<String> tokens) {
if (tokens.size() > 1) {
String last = tokens.remove(tokens.size() - 1);
String result = tokens.stream().collect(Collectors.joining(", "));
return result + " and " + last;
} else if (tokens.size() == 1) {
return tokens.get(0);
} else {
return "";
}
}*/
public String booleanToYesNo(boolean value) {
return value ? "Yes" : "No";
@ -429,4 +485,17 @@ public class BSFormatter {
else
return isOfferer ? "Seller (offerer)" : "Buyer (taker)";
}
public String formatBytes(int bytes) {
double kb = 1024;
double mb = kb * kb;
DecimalFormat decimalFormat = new DecimalFormat("#.##");
if (bytes < kb)
return bytes + " bytes";
else if (bytes < mb)
return decimalFormat.format(bytes / kb) + " KB";
else
return decimalFormat.format(bytes / mb) + " MB";
}
}

17
network/src/main/java/io/bitsquare/p2p/P2PService.java
View File

@ -16,7 +16,11 @@ import io.bitsquare.crypto.EncryptionService;
import io.bitsquare.crypto.PrefixedSealedAndSignedMessage;
import io.bitsquare.p2p.messaging.*;
import io.bitsquare.p2p.network.*;
import io.bitsquare.p2p.peers.*;
import io.bitsquare.p2p.peers.Broadcaster;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.getdata.RequestDataManager;
import io.bitsquare.p2p.peers.keepalive.KeepAliveManager;
import io.bitsquare.p2p.peers.peerexchange.PeerExchangeManager;
import io.bitsquare.p2p.seed.SeedNodesRepository;
import io.bitsquare.p2p.storage.HashMapChangedListener;
import io.bitsquare.p2p.storage.P2PDataStorage;
@ -78,7 +82,7 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
private Subscription networkReadySubscription;
private boolean isBootstrapped;
private ChangeListener<Number> numOfBroadcastsChangeListener;
private MaintenanceManager maintenanceManager;
private KeepAliveManager keepAliveManager;
///////////////////////////////////////////////////////////////////////////////////////////
@ -126,7 +130,7 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
peerExchangeManager = new PeerExchangeManager(networkNode, peerManager, seedNodeAddresses);
maintenanceManager = new MaintenanceManager(networkNode, peerManager, seedNodeAddresses);
keepAliveManager = new KeepAliveManager(networkNode, peerManager);
// We need to have both the initial data delivered and the hidden service published
@ -171,8 +175,8 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
if (peerExchangeManager != null)
peerExchangeManager.shutDown();
if (maintenanceManager != null)
maintenanceManager.shutDown();
if (keepAliveManager != null)
keepAliveManager.shutDown();
if (networkNode != null)
networkNode.shutDown(() -> {
@ -214,6 +218,7 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
Log.traceCall();
requestDataManager.requestPreliminaryData();
keepAliveManager.start();
p2pServiceListeners.stream().forEach(SetupListener::onTorNodeReady);
}
@ -226,8 +231,6 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
hiddenServicePublished.set(true);
p2pServiceListeners.stream().forEach(SetupListener::onHiddenServicePublished);
maintenanceManager.start();
}
@Override

60
network/src/main/java/io/bitsquare/p2p/network/Connection.java
View File

@ -24,7 +24,10 @@ import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.*;
import static com.google.common.base.Preconditions.checkArgument;
@ -36,15 +39,6 @@ import static com.google.common.base.Preconditions.checkNotNull;
*/
public class Connection implements MessageListener {
private static final Logger log = LoggerFactory.getLogger(Connection.class);
private static final int MAX_MSG_SIZE = 100 * 1024; // 100 kb of compressed data
private static final int MSG_THROTTLE_PER_SEC = 10; // With MAX_MSG_SIZE of 100kb results in bandwidth of 10 mbit/sec
private static final int MSG_THROTTLE_PER_10SEC = 50; // With MAX_MSG_SIZE of 100kb results in bandwidth of 5 mbit/sec for 10 sec
//timeout on blocking Socket operations like ServerSocket.accept() or SocketInputStream.read()
private static final int SOCKET_TIMEOUT = (int) TimeUnit.MINUTES.toMillis(30);
public static int getMaxMsgSize() {
return MAX_MSG_SIZE;
}
///////////////////////////////////////////////////////////////////////////////////////////
// Enums
@ -53,7 +47,21 @@ public class Connection implements MessageListener {
public enum PeerType {
SEED_NODE,
PEER,
DIRECT_MSG_PEER
DIRECT_MSG_PEER;
}
///////////////////////////////////////////////////////////////////////////////////////////
// Static
///////////////////////////////////////////////////////////////////////////////////////////
private static final int MAX_MSG_SIZE = 100 * 1024; // 100 kb of compressed data
private static final int MSG_THROTTLE_PER_SEC = 10; // With MAX_MSG_SIZE of 100kb results in bandwidth of 10 mbit/sec
private static final int MSG_THROTTLE_PER_10SEC = 50; // With MAX_MSG_SIZE of 100kb results in bandwidth of 5 mbit/sec for 10 sec
private static final int SOCKET_TIMEOUT = (int) TimeUnit.SECONDS.toMillis(60);
public static int getMaxMsgSize() {
return MAX_MSG_SIZE;
}
@ -69,6 +77,7 @@ public class Connection implements MessageListener {
private final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// holder of state shared between InputHandler and Connection
private final SharedModel sharedModel;
private final Statistic statistic;
// set in init
private InputHandler inputHandler;
@ -97,6 +106,7 @@ public class Connection implements MessageListener {
this.socket = socket;
//this.messageListener = messageListener;
this.connectionListener = connectionListener;
statistic = new Statistic();
addMessageListener(messageListener);
@ -129,8 +139,6 @@ public class Connection implements MessageListener {
sharedModel.handleConnectionException(e);
}
sharedModel.updateLastActivityDate();
// Use Peer as default, in case of other types they will set it as soon as possible.
peerType = PeerType.PEER;
@ -182,7 +190,9 @@ public class Connection implements MessageListener {
if (!stopped) {
objectOutputStream.writeObject(objectToWrite);
objectOutputStream.flush();
sharedModel.updateLastActivityDate();
statistic.addSentBytes(ByteArrayUtils.objectToByteArray(objectToWrite).length);
statistic.updateLastActivityTimestamp();
}
} catch (IOException e) {
// an exception lead to a shutdown
@ -282,10 +292,6 @@ public class Connection implements MessageListener {
return peersNodeAddressOptional;
}
public Date getLastActivityDate() {
return sharedModel.getLastActivityDate();
}
public String getUid() {
return uid;
}
@ -310,6 +316,9 @@ public class Connection implements MessageListener {
return sharedModel.getRuleViolation();
}
public Statistic getStatistic() {
return statistic;
}
///////////////////////////////////////////////////////////////////////////////////////////
// ShutDown
@ -410,7 +419,6 @@ public class Connection implements MessageListener {
"peerAddress=" + peersNodeAddressOptional +
", peerType=" + peerType +
", uid='" + uid + '\'' +
", lastActivityDate=" + getLastActivityDate() +
'}';
}
@ -444,7 +452,6 @@ public class Connection implements MessageListener {
private final ConcurrentHashMap<RuleViolation, Integer> ruleViolations = new ConcurrentHashMap<>();
// mutable
private Date lastActivityDate;
private volatile boolean stopped;
private CloseConnectionReason closeConnectionReason;
private RuleViolation ruleViolation;
@ -454,14 +461,6 @@ public class Connection implements MessageListener {
this.socket = socket;
}
public void updateLastActivityDate() {
lastActivityDate = new Date();
}
public Date getLastActivityDate() {
return lastActivityDate;
}
public void reportInvalidRequest(RuleViolation ruleViolation) {
log.warn("We got reported an corrupt request " + ruleViolation + "\n\tconnection=" + this);
int numRuleViolations;
@ -539,7 +538,6 @@ public class Connection implements MessageListener {
return "SharedSpace{" +
", socket=" + socket +
", ruleViolations=" + ruleViolations +
", lastActivityDate=" + lastActivityDate +
'}';
}
}
@ -648,7 +646,6 @@ public class Connection implements MessageListener {
}
Connection connection = sharedModel.connection;
sharedModel.updateLastActivityDate();
if (message instanceof CloseConnectionMessage) {
CloseConnectionReason[] values = CloseConnectionReason.values();
log.info("CloseConnectionMessage received. Reason={}\n\t" +
@ -656,6 +653,9 @@ public class Connection implements MessageListener {
stop();
sharedModel.shutDown(CloseConnectionReason.CLOSE_REQUESTED_BY_PEER);
} else if (!stopped) {
connection.statistic.updateLastActivityTimestamp();
connection.statistic.addReceivedBytes(size);
// First a seed node gets a message form a peer (PreliminaryDataRequest using
// AnonymousMessage interface) which does not has its hidden service
// published, so does not know its address. As the IncomingConnection does not has the

59
network/src/main/java/io/bitsquare/p2p/network/Statistic.java Normal file
View File

@ -0,0 +1,59 @@
package io.bitsquare.p2p.network;
import javafx.beans.property.IntegerProperty;
import javafx.beans.property.LongProperty;
import javafx.beans.property.SimpleIntegerProperty;
import javafx.beans.property.SimpleLongProperty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Date;
public class Statistic {
private static final Logger log = LoggerFactory.getLogger(Statistic.class);
private final Date creationDate;
private long lastActivityTimestamp;
private int sentBytes = 0;
private int receivedBytes = 0;
public LongProperty lastActivityTimestampProperty = new SimpleLongProperty(System.currentTimeMillis());
public IntegerProperty sentBytesProperty = new SimpleIntegerProperty(0);
public IntegerProperty receivedBytesProperty = new SimpleIntegerProperty(0);
public Statistic() {
creationDate = new Date();
updateLastActivityTimestamp();
}
public Date getCreationDate() {
return creationDate;
}
public void updateLastActivityTimestamp() {
lastActivityTimestamp = System.currentTimeMillis();
lastActivityTimestampProperty.set(lastActivityTimestamp);
}
public long getLastActivityTimestamp() {
return lastActivityTimestamp;
}
public int getSentBytes() {
return sentBytes;
}
public void addSentBytes(int sentBytes) {
this.sentBytes += sentBytes;
sentBytesProperty.set(this.sentBytes);
}
public int getReceivedBytes() {
return receivedBytes;
}
public void addReceivedBytes(int receivedBytes) {
this.receivedBytes += receivedBytes;
receivedBytesProperty.set(this.receivedBytes);
}
}

217
network/src/main/java/io/bitsquare/p2p/peers/MaintenanceHandshake.java
View File

@ -1,217 +0,0 @@
package io.bitsquare.p2p.peers;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
import io.bitsquare.app.Log;
import io.bitsquare.common.UserThread;
import io.bitsquare.p2p.Message;
import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.network.CloseConnectionReason;
import io.bitsquare.p2p.network.Connection;
import io.bitsquare.p2p.network.MessageListener;
import io.bitsquare.p2p.network.NetworkNode;
import io.bitsquare.p2p.peers.messages.peers.GetPeersRequest;
import io.bitsquare.p2p.peers.messages.peers.GetPeersResponse;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.util.HashSet;
import java.util.Random;
import java.util.Timer;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
public class MaintenanceHandshake implements MessageListener {
private static final Logger log = LoggerFactory.getLogger(MaintenanceHandshake.class);
///////////////////////////////////////////////////////////////////////////////////////////
// Listener
///////////////////////////////////////////////////////////////////////////////////////////
public interface Listener {
void onComplete();
void onFault(String errorMessage, @Nullable Connection connection);
}
///////////////////////////////////////////////////////////////////////////////////////////
// Class fields
///////////////////////////////////////////////////////////////////////////////////////////
private final NetworkNode networkNode;
private final PeerManager peerManager;
private final Listener listener;
private final long nonce = new Random().nextLong();
private Timer timeoutTimer;
///////////////////////////////////////////////////////////////////////////////////////////
// Constructor
///////////////////////////////////////////////////////////////////////////////////////////
public MaintenanceHandshake(NetworkNode networkNode, PeerManager peerManager, Listener listener) {
this.networkNode = networkNode;
this.peerManager = peerManager;
this.listener = listener;
networkNode.addMessageListener(this);
}
public void shutDown() {
networkNode.removeMessageListener(this);
stopTimeoutTimer();
}
///////////////////////////////////////////////////////////////////////////////////////////
// API
///////////////////////////////////////////////////////////////////////////////////////////
public void requestReportedPeers(NodeAddress nodeAddress) {
Log.traceCall("nodeAddress=" + nodeAddress + " / this=" + this);
checkNotNull(networkNode.getNodeAddress(), "PeerExchangeHandshake.requestReportedPeers: My node address must " +
"not be null at requestReportedPeers");
GetPeersRequest getPeersRequest = new GetPeersRequest(networkNode.getNodeAddress(), nonce, getConnectedPeers(nodeAddress));
SettableFuture<Connection> future = networkNode.sendMessage(nodeAddress, getPeersRequest);
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("Send " + getPeersRequest + " to " + nodeAddress + " succeeded.");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
String errorMessage = "Sending getPeersRequest to " + nodeAddress +
" failed. That is expected if the peer is offline.\n\tgetPeersRequest=" + getPeersRequest +
".\n\tException=" + throwable.getMessage();
log.info(errorMessage);
peerManager.shutDownConnection(nodeAddress, CloseConnectionReason.SEND_MSG_FAILURE);
shutDown();
listener.onFault(errorMessage, null);
}
});
checkArgument(timeoutTimer == null, "requestReportedPeers must not be called twice.");
timeoutTimer = UserThread.runAfter(() -> {
String errorMessage = "A timeout occurred at sending getPeersRequest:" + getPeersRequest + " for nodeAddress:" + nodeAddress;
log.info(errorMessage + " / PeerExchangeHandshake=" +
MaintenanceHandshake.this);
log.info("timeoutTimer called on " + this);
peerManager.shutDownConnection(nodeAddress, CloseConnectionReason.SEND_MSG_TIMEOUT);
shutDown();
listener.onFault(errorMessage, null);
},
20, TimeUnit.SECONDS);
}
public void onGetPeersRequest(GetPeersRequest getPeersRequest, final Connection connection) {
Log.traceCall("getPeersRequest=" + getPeersRequest + "\n\tconnection=" + connection + "\n\tthis=" + this);
HashSet<ReportedPeer> reportedPeers = getPeersRequest.reportedPeers;
peerManager.printReportedPeers(reportedPeers);
checkArgument(connection.getPeersNodeAddressOptional().isPresent(),
"The peers address must have been already set at the moment");
GetPeersResponse getPeersResponse = new GetPeersResponse(getPeersRequest.nonce,
getConnectedPeers(connection.getPeersNodeAddressOptional().get()));
SettableFuture<Connection> future = networkNode.sendMessage(connection,
getPeersResponse);
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("GetPeersResponse sent successfully");
shutDown();
listener.onComplete();
}
@Override
public void onFailure(@NotNull Throwable throwable) {
String errorMessage = "Sending getPeersRequest to " + connection +
" failed. That is expected if the peer is offline. getPeersRequest=" + getPeersRequest + "." +
"Exception: " + throwable.getMessage();
log.info(errorMessage);
peerManager.shutDownConnection(connection, CloseConnectionReason.SEND_MSG_FAILURE);
shutDown();
listener.onFault(errorMessage, connection);
}
});
checkArgument(timeoutTimer == null, "onGetPeersRequest must not be called twice.");
timeoutTimer = UserThread.runAfter(() -> {
String errorMessage = "A timeout occurred at sending getPeersResponse:" + getPeersResponse + " on connection:" + connection;
log.info(errorMessage + " / PeerExchangeHandshake=" +
MaintenanceHandshake.this);
log.info("timeoutTimer called. this=" + this);
peerManager.shutDownConnection(connection, CloseConnectionReason.SEND_MSG_TIMEOUT);
shutDown();
listener.onFault(errorMessage, connection);
},
20, TimeUnit.SECONDS);
peerManager.addToReportedPeers(reportedPeers, connection);
}
///////////////////////////////////////////////////////////////////////////////////////////
// MessageListener implementation
///////////////////////////////////////////////////////////////////////////////////////////
@Override
public void onMessage(Message message, Connection connection) {
if (message instanceof GetPeersResponse) {
Log.traceCall(message.toString() + "\n\tconnection=" + connection);
Log.traceCall("this=" + this);
GetPeersResponse getPeersResponse = (GetPeersResponse) message;
if (getPeersResponse.requestNonce == nonce) {
stopTimeoutTimer();
HashSet<ReportedPeer> reportedPeers = getPeersResponse.reportedPeers;
StringBuilder result = new StringBuilder("Received peers:");
reportedPeers.stream().forEach(e -> result.append("\n\t").append(e));
log.trace(result.toString());
peerManager.addToReportedPeers(reportedPeers, connection);
shutDown();
listener.onComplete();
} else {
log.debug("Nonce not matching. That can happen rarely if we get a response after a canceled handshake " +
"(timeout causes connection close but peer might have sent a msg before connection " +
"was closed).\n\tWe drop that message. nonce={} / requestNonce={}",
nonce, getPeersResponse.requestNonce);
}
}
}
///////////////////////////////////////////////////////////////////////////////////////////
// Private
///////////////////////////////////////////////////////////////////////////////////////////
private HashSet<ReportedPeer> getConnectedPeers(NodeAddress receiverNodeAddress) {
return new HashSet<>(peerManager.getConnectedPeers().stream()
.filter(e -> !peerManager.isSeedNode(e) &&
!peerManager.isSelf(e) &&
!e.nodeAddress.equals(receiverNodeAddress)
)
.collect(Collectors.toSet()));
}
private void stopTimeoutTimer() {
if (timeoutTimer != null) {
timeoutTimer.cancel();
timeoutTimer = null;
}
}
}

283
network/src/main/java/io/bitsquare/p2p/peers/MaintenanceManager.java
View File

@ -1,283 +0,0 @@
package io.bitsquare.p2p.peers;
import com.google.common.util.concurrent.MoreExecutors;
import io.bitsquare.app.Log;
import io.bitsquare.common.UserThread;
import io.bitsquare.common.util.Utilities;
import io.bitsquare.p2p.Message;
import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.network.*;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
public class MaintenanceManager implements MessageListener, ConnectionListener {
private static final Logger log = LoggerFactory.getLogger(MaintenanceManager.class);
private static final int MAINTENANCE_DELAY_SEC = 5 * 60;
private final NetworkNode networkNode;
private final PeerManager peerManager;
private final Set<NodeAddress> seedNodeAddresses;
private ScheduledThreadPoolExecutor executor;
private final Map<NodeAddress, PeerExchangeHandler> peerExchangeHandshakeMap = new HashMap<>();
private Timer connectToMorePeersTimer, maintainConnectionsTimer;
private boolean shutDownInProgress;
///////////////////////////////////////////////////////////////////////////////////////////
// Constructor
///////////////////////////////////////////////////////////////////////////////////////////
public MaintenanceManager(NetworkNode networkNode, PeerManager peerManager, Set<NodeAddress> seedNodeAddresses) {
this.networkNode = networkNode;
this.peerManager = peerManager;
checkArgument(!seedNodeAddresses.isEmpty(), "seedNodeAddresses must not be empty");
this.seedNodeAddresses = new HashSet<>(seedNodeAddresses);
networkNode.addMessageListener(this);
}
public void shutDown() {
Log.traceCall();
shutDownInProgress = true;
networkNode.removeMessageListener(this);
stopConnectToMorePeersTimer();
stopMaintainConnectionsTimer();
peerExchangeHandshakeMap.values().stream().forEach(PeerExchangeHandler::cleanup);
if (executor != null)
MoreExecutors.shutdownAndAwaitTermination(executor, 100, TimeUnit.MILLISECONDS);
}
///////////////////////////////////////////////////////////////////////////////////////////
// API
///////////////////////////////////////////////////////////////////////////////////////////
public void start() {
if (executor == null) {
executor = Utilities.getScheduledThreadPoolExecutor("MaintenanceManager", 1, 2, 5);
int delay = new Random().nextInt(120) + MAINTENANCE_DELAY_SEC; // add 1-2 min. randomness
executor.scheduleAtFixedRate(() -> UserThread.execute(this::maintainConnections),
delay, delay, TimeUnit.SECONDS);
}
}
///////////////////////////////////////////////////////////////////////////////////////////
// ConnectionListener implementation
///////////////////////////////////////////////////////////////////////////////////////////
@Override
public void onConnection(Connection connection) {
}
@Override
public void onDisconnect(CloseConnectionReason closeConnectionReason, Connection connection) {
/* // We use a timer to throttle if we get a series of disconnects
// The more connections we have the more relaxed we are with a checkConnections
stopMaintainConnectionsTimer();
int size = networkNode.getAllConnections().size();
int delay = 10 + 2 * size * size; // 12 sec - 210 sec (3.5 min)
maintainConnectionsTimer = UserThread.runAfter(this::maintainConnections,
delay, TimeUnit.SECONDS);*/
}
@Override
public void onError(Throwable throwable) {
}
///////////////////////////////////////////////////////////////////////////////////////////
// MessageListener implementation
///////////////////////////////////////////////////////////////////////////////////////////
@Override
public void onMessage(Message message, Connection connection) {
/* if (message instanceof GetPeersRequest) {
Log.traceCall(message.toString() + "\n\tconnection=" + connection);
PeerExchangeHandshake peerExchangeHandshake = new PeerExchangeHandshake(networkNode,
peerManager,
new PeerExchangeHandshake.Listener() {
@Override
public void onComplete() {
log.trace("PeerExchangeHandshake of inbound connection complete.\n\tConnection={}", connection);
}
@Override
public void onFault(String errorMessage, @Nullable Connection connection) {
log.trace("PeerExchangeHandshake of outbound connection failed.\n\terrorMessage={}\n\t" +
"connection={}", errorMessage, connection);
peerManager.handleConnectionFault(connection);
}
});
peerExchangeHandshake.onGetPeersRequest((GetPeersRequest) message, connection);
}*/
}
///////////////////////////////////////////////////////////////////////////////////////////
// Private
///////////////////////////////////////////////////////////////////////////////////////////
private void requestReportedPeers(NodeAddress nodeAddress, List<NodeAddress> remainingNodeAddresses) {
Log.traceCall("nodeAddress=" + nodeAddress);
if (!peerExchangeHandshakeMap.containsKey(nodeAddress)) {
PeerExchangeHandler peerExchangeHandler = new PeerExchangeHandler(networkNode,
peerManager,
new PeerExchangeHandler.Listener() {
@Override
public void onComplete() {
log.trace("PeerExchangeHandshake of outbound connection complete. nodeAddress={}", nodeAddress);
peerExchangeHandshakeMap.remove(nodeAddress);
connectToMorePeers();
}
@Override
public void onFault(String errorMessage, @Nullable Connection connection) {
log.trace("PeerExchangeHandshake of outbound connection failed.\n\terrorMessage={}\n\t" +
"nodeAddress={}", errorMessage, nodeAddress);
peerExchangeHandshakeMap.remove(nodeAddress);
peerManager.handleConnectionFault(nodeAddress, connection);
if (!shutDownInProgress) {
if (!remainingNodeAddresses.isEmpty()) {
log.info("There are remaining nodes available for requesting peers. " +
"We will try getReportedPeers again.");
requestReportedPeersFromRandomPeer(remainingNodeAddresses);
} else {
log.info("There is no remaining node available for requesting peers. " +
"That is expected if no other node is online.\n\t" +
"We will try again after a random pause.");
if (connectToMorePeersTimer == null)
connectToMorePeersTimer = UserThread.runAfterRandomDelay(
MaintenanceManager.this::connectToMorePeers, 20, 30);
}
}
}
});
peerExchangeHandshakeMap.put(nodeAddress, peerExchangeHandler);
peerExchangeHandler.requestConnectedPeers(nodeAddress);
} else {
//TODO check when that happens
log.warn("We have started already a peerExchangeHandshake. " +
"We ignore that call. " +
"nodeAddress=" + nodeAddress);
}
}
// we check if we have at least one seed node connected
private void maintainConnections() {
Log.traceCall();
stopMaintainConnectionsTimer();
// we want at least 1 seed node connected
Set<Connection> confirmedConnections = networkNode.getConfirmedConnections();
long numberOfConnectedSeedNodes = confirmedConnections.stream()
.filter(peerManager::isSeedNode)
.count();
if (numberOfConnectedSeedNodes == 0) {
ArrayList<NodeAddress> nodeAddresses = new ArrayList<>(seedNodeAddresses);
Collections.shuffle(nodeAddresses);
requestReportedPeersFromRandomPeer(nodeAddresses);
}
// We try to get sufficient connections by connecting to reported and persisted peers
if (numberOfConnectedSeedNodes == 0) {
// If we requested a seed node we delay a bit to not have too many requests simultaneously
if (connectToMorePeersTimer == null)
connectToMorePeersTimer = UserThread.runAfter(this::connectToMorePeers, 10);
} else {
connectToMorePeers();
}
// Use all outbound connections older than 10 min. for updating reported peers and make sure we keep the connection alive
// Inbound connections should be maintained be the requesting peer
confirmedConnections.stream()
.filter(c -> c.getPeersNodeAddressOptional().isPresent() &&
c instanceof OutboundConnection &&
new Date().getTime() - c.getLastActivityDate().getTime() > TimeUnit.MINUTES.toMillis(10))
.forEach(c -> {
log.trace("Call requestReportedPeers on a confirmedConnection by the maintainConnections call");
requestReportedPeers(c.getPeersNodeAddressOptional().get(), new ArrayList<>());
});
}
private void connectToMorePeers() {
Log.traceCall();
stopConnectToMorePeersTimer();
if (!peerManager.hasSufficientConnections()) {
// We create a new list of not connected candidates
// 1. reported sorted by most recent lastActivityDate
// 2. persisted sorted by most recent lastActivityDate
// 3. seenNodes
List<NodeAddress> list = new ArrayList<>(getFilteredAndSortedList(peerManager.getReportedPeers(), new ArrayList<>()));
list.addAll(getFilteredAndSortedList(peerManager.getPersistedPeers(), list));
ArrayList<NodeAddress> seedNodeAddresses = new ArrayList<>(this.seedNodeAddresses);
Collections.shuffle(seedNodeAddresses);
list.addAll(seedNodeAddresses.stream()
.filter(e -> !list.contains(e) &&
!peerManager.isSelf(e) &&
!peerManager.isConfirmed(e))
.collect(Collectors.toSet()));
log.info("Sorted and filtered list: list.size()=" + list.size());
log.trace("Sorted and filtered list: list=" + list);
if (!list.isEmpty()) {
NodeAddress nextCandidate = list.get(0);
list.remove(nextCandidate);
requestReportedPeers(nextCandidate, list);
} else {
log.info("No more peers are available for requestReportedPeers.");
}
} else {
log.info("We have already sufficient connections.");
}
}
// sorted by most recent lastActivityDate
private List<NodeAddress> getFilteredAndSortedList(Set<ReportedPeer> set, List<NodeAddress> list) {
return set.stream()
.filter(e -> !list.contains(e.nodeAddress) &&
!peerManager.isSeedNode(e) &&
!peerManager.isSelf(e) &&
!peerManager.isConfirmed(e))
.collect(Collectors.toList())
.stream()
.filter(e -> e.lastActivityDate != null)
.sorted((o1, o2) -> o2.lastActivityDate.compareTo(o1.lastActivityDate))
.map(e -> e.nodeAddress)
.collect(Collectors.toList());
}
private void requestReportedPeersFromRandomPeer(List<NodeAddress> remainingNodeAddresses) {
NodeAddress nextCandidate = remainingNodeAddresses.get(new Random().nextInt(remainingNodeAddresses.size()));
remainingNodeAddresses.remove(nextCandidate);
requestReportedPeers(nextCandidate, remainingNodeAddresses);
}
private void stopConnectToMorePeersTimer() {
if (connectToMorePeersTimer != null) {
connectToMorePeersTimer.cancel();
connectToMorePeersTimer = null;
}
}
private void stopMaintainConnectionsTimer() {
if (maintainConnectionsTimer != null) {
maintainConnectionsTimer.cancel();
maintainConnectionsTimer = null;
}
}
}

23
network/src/main/java/io/bitsquare/p2p/peers/PeerManager.java
View File

@ -5,7 +5,8 @@ import io.bitsquare.common.UserThread;
import io.bitsquare.p2p.Message;
import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.network.*;
import io.bitsquare.p2p.peers.messages.data.GetUpdatedDataRequest;
import io.bitsquare.p2p.peers.getdata.messages.GetUpdatedDataRequest;
import io.bitsquare.p2p.peers.peerexchange.ReportedPeer;
import io.bitsquare.storage.Storage;
import javafx.beans.value.ChangeListener;
import org.slf4j.Logger;
@ -30,6 +31,8 @@ public class PeerManager implements ConnectionListener, MessageListener {
private static int MIN_CONNECTIONS;
private static int MAX_CONNECTIONS_EXTENDED_1;
private static int MAX_CONNECTIONS_EXTENDED_2;
private static int MAX_CONNECTIONS_EXTENDED_3;
private boolean printReportedPeersDetails = true;
@ -94,6 +97,9 @@ public class PeerManager implements ConnectionListener, MessageListener {
stopCheckMaxConnectionsTimer();
}
public int getMaxConnections() {
return MAX_CONNECTIONS_EXTENDED_3;
}
///////////////////////////////////////////////////////////////////////////////////////////
// ConnectionListener implementation
@ -198,7 +204,7 @@ public class PeerManager implements ConnectionListener, MessageListener {
}
if (candidates.size() > 0) {
candidates.sort((o1, o2) -> o1.getLastActivityDate().compareTo(o2.getLastActivityDate()));
candidates.sort((o1, o2) -> ((Long) o1.getStatistic().getLastActivityTimestamp()).compareTo(((Long) o2.getStatistic().getLastActivityTimestamp())));
log.info("Candidates.size() for shut down=" + candidates.size());
Connection connection = candidates.remove(0);
log.info("We are going to shut down the oldest connection.\n\tconnection=" + connection.toString());
@ -225,7 +231,7 @@ public class PeerManager implements ConnectionListener, MessageListener {
.collect(Collectors.toList());
if (candidates.size() > 1) {
candidates.sort((o1, o2) -> o1.getLastActivityDate().compareTo(o2.getLastActivityDate()));
candidates.sort((o1, o2) -> ((Long) o1.getStatistic().getLastActivityTimestamp()).compareTo(((Long) o2.getStatistic().getLastActivityTimestamp())));
log.info("Number of connections exceeding MAX_CONNECTIONS_EXTENDED_1. Current size=" + candidates.size());
Connection connection = candidates.remove(0);
log.info("We are going to shut down the oldest connection.\n\tconnection=" + connection.toString());
@ -260,8 +266,8 @@ public class PeerManager implements ConnectionListener, MessageListener {
private void removeTooOldReportedPeers() {
Log.traceCall();
Set<ReportedPeer> reportedPeersToRemove = reportedPeers.stream()
.filter(reportedPeer -> reportedPeer.lastActivityDate != null &&
new Date().getTime() - reportedPeer.lastActivityDate.getTime() > MAX_AGE)
.filter(reportedPeer -> reportedPeer.date != null &&
new Date().getTime() - reportedPeer.date.getTime() > MAX_AGE)
.collect(Collectors.toSet());
reportedPeersToRemove.forEach(this::removeReportedPeer);
}
@ -345,8 +351,8 @@ public class PeerManager implements ConnectionListener, MessageListener {
private void removeTooOldPersistedPeers() {
Log.traceCall();
Set<ReportedPeer> persistedPeersToRemove = persistedPeers.stream()
.filter(reportedPeer -> reportedPeer.lastActivityDate != null &&
new Date().getTime() - reportedPeer.lastActivityDate.getTime() > MAX_AGE)
.filter(reportedPeer -> reportedPeer.date != null &&
new Date().getTime() - reportedPeer.date.getTime() > MAX_AGE)
.collect(Collectors.toSet());
persistedPeersToRemove.forEach(this::removePersistedPeer);
}
@ -377,7 +383,6 @@ public class PeerManager implements ConnectionListener, MessageListener {
} else {
if (reportedPeer != null) {
removePersistedPeer(nodeAddress);
reportedPeer.penalizeLastActivityDate();
persistedPeers.add(reportedPeer);
dbStorage.queueUpForSave(persistedPeers, 5000);
@ -486,7 +491,7 @@ public class PeerManager implements ConnectionListener, MessageListener {
// networkNode.getConfirmedConnections includes:
// filter(connection -> connection.getPeersNodeAddressOptional().isPresent())
return networkNode.getConfirmedConnections().stream()
.map(c -> new ReportedPeer(c.getPeersNodeAddressOptional().get(), c.getLastActivityDate()))
.map(c -> new ReportedPeer(c.getPeersNodeAddressOptional().get()))
.collect(Collectors.toSet());
}

11
network/src/main/java/io/bitsquare/p2p/peers/RequestDataHandshake.java → network/src/main/java/io/bitsquare/p2p/peers/getdata/RequestDataHandshake.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers;
package io.bitsquare.p2p.peers.getdata;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
@ -11,10 +11,11 @@ import io.bitsquare.p2p.network.CloseConnectionReason;
import io.bitsquare.p2p.network.Connection;
import io.bitsquare.p2p.network.MessageListener;
import io.bitsquare.p2p.network.NetworkNode;
import io.bitsquare.p2p.peers.messages.data.GetDataRequest;
import io.bitsquare.p2p.peers.messages.data.GetDataResponse;
import io.bitsquare.p2p.peers.messages.data.GetUpdatedDataRequest;
import io.bitsquare.p2p.peers.messages.data.PreliminaryGetDataRequest;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.getdata.messages.GetDataRequest;
import io.bitsquare.p2p.peers.getdata.messages.GetDataResponse;
import io.bitsquare.p2p.peers.getdata.messages.GetUpdatedDataRequest;
import io.bitsquare.p2p.peers.getdata.messages.PreliminaryGetDataRequest;
import io.bitsquare.p2p.storage.P2PDataStorage;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

10
network/src/main/java/io/bitsquare/p2p/peers/RequestDataManager.java → network/src/main/java/io/bitsquare/p2p/peers/getdata/RequestDataManager.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers;
package io.bitsquare.p2p.peers.getdata;
import io.bitsquare.app.Log;
import io.bitsquare.common.UserThread;
@ -7,7 +7,9 @@ import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.network.Connection;
import io.bitsquare.p2p.network.MessageListener;
import io.bitsquare.p2p.network.NetworkNode;
import io.bitsquare.p2p.peers.messages.data.GetDataRequest;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.getdata.messages.GetDataRequest;
import io.bitsquare.p2p.peers.peerexchange.ReportedPeer;
import io.bitsquare.p2p.storage.P2PDataStorage;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
@ -241,8 +243,8 @@ public class RequestDataManager implements MessageListener {
!peerManager.isSelf(e))
.collect(Collectors.toList())
.stream()
.filter(e -> e.lastActivityDate != null)
.sorted((o1, o2) -> o2.lastActivityDate.compareTo(o1.lastActivityDate))
.filter(e -> e.date != null)
.sorted((o1, o2) -> o2.date.compareTo(o1.date))
.map(e -> e.nodeAddress)
.collect(Collectors.toList());
}

2
network/src/main/java/io/bitsquare/p2p/peers/messages/data/GetDataRequest.java → network/src/main/java/io/bitsquare/p2p/peers/getdata/messages/GetDataRequest.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers.messages.data;
package io.bitsquare.p2p.peers.getdata.messages;
import io.bitsquare.p2p.Message;

2
network/src/main/java/io/bitsquare/p2p/peers/messages/data/GetDataResponse.java → network/src/main/java/io/bitsquare/p2p/peers/getdata/messages/GetDataResponse.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers.messages.data;
package io.bitsquare.p2p.peers.getdata.messages;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.Message;

2
network/src/main/java/io/bitsquare/p2p/peers/messages/data/GetUpdatedDataRequest.java → network/src/main/java/io/bitsquare/p2p/peers/getdata/messages/GetUpdatedDataRequest.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers.messages.data;
package io.bitsquare.p2p.peers.getdata.messages;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.NodeAddress;

2
network/src/main/java/io/bitsquare/p2p/peers/messages/data/PreliminaryGetDataRequest.java → network/src/main/java/io/bitsquare/p2p/peers/getdata/messages/PreliminaryGetDataRequest.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers.messages.data;
package io.bitsquare.p2p.peers.getdata.messages;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.network.messages.AnonymousMessage;

115
network/src/main/java/io/bitsquare/p2p/peers/keepalive/KeepAliveHandler.java Normal file
View File

@ -0,0 +1,115 @@
package io.bitsquare.p2p.peers.keepalive;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
import io.bitsquare.app.Log;
import io.bitsquare.p2p.Message;
import io.bitsquare.p2p.network.CloseConnectionReason;
import io.bitsquare.p2p.network.Connection;
import io.bitsquare.p2p.network.MessageListener;
import io.bitsquare.p2p.network.NetworkNode;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.keepalive.messages.Ping;
import io.bitsquare.p2p.peers.keepalive.messages.Pong;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.util.Random;
public class KeepAliveHandler implements MessageListener {
private static final Logger log = LoggerFactory.getLogger(KeepAliveHandler.class);
private Connection connection;
///////////////////////////////////////////////////////////////////////////////////////////
// Listener
///////////////////////////////////////////////////////////////////////////////////////////
public interface Listener {
void onComplete();
void onFault(String errorMessage, @Nullable Connection connection);
}
///////////////////////////////////////////////////////////////////////////////////////////
// Class fields
///////////////////////////////////////////////////////////////////////////////////////////
private final NetworkNode networkNode;
private final PeerManager peerManager;
private final Listener listener;
private final int nonce = new Random().nextInt();
///////////////////////////////////////////////////////////////////////////////////////////
// Constructor
///////////////////////////////////////////////////////////////////////////////////////////
public KeepAliveHandler(NetworkNode networkNode, PeerManager peerManager, Listener listener) {
this.networkNode = networkNode;
this.peerManager = peerManager;
this.listener = listener;
}
public void cleanup() {
if (connection != null)
connection.removeMessageListener(this);
}
///////////////////////////////////////////////////////////////////////////////////////////
// API
///////////////////////////////////////////////////////////////////////////////////////////
public void sendPing(Connection connection) {
Log.traceCall("connection=" + connection + " / this=" + this);
this.connection = connection;
connection.addMessageListener(this);
Ping ping = new Ping(nonce);
SettableFuture<Connection> future = networkNode.sendMessage(connection, ping);
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("Send " + ping + " to " + connection + " succeeded.");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
String errorMessage = "Sending ping to " + connection +
" failed. That is expected if the peer is offline.\n\tping=" + ping +
".\n\tException=" + throwable.getMessage();
log.info(errorMessage);
cleanup();
peerManager.shutDownConnection(connection, CloseConnectionReason.SEND_MSG_FAILURE);
listener.onFault(errorMessage, connection);
}
});
}
///////////////////////////////////////////////////////////////////////////////////////////
// MessageListener implementation
///////////////////////////////////////////////////////////////////////////////////////////
@Override
public void onMessage(Message message, Connection connection) {
if (message instanceof Pong) {
Log.traceCall(message.toString() + "\n\tconnection=" + connection);
Pong pong = (Pong) message;
if (pong.requestNonce == nonce) {
cleanup();
listener.onComplete();
} else {
log.warn("Nonce not matching. That should never happen.\n\t" +
"We drop that message. nonce={} / requestNonce={}",
nonce, pong.requestNonce);
}
}
}
}

144
network/src/main/java/io/bitsquare/p2p/peers/keepalive/KeepAliveManager.java Normal file
View File

@ -0,0 +1,144 @@
package io.bitsquare.p2p.peers.keepalive;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import io.bitsquare.app.Log;
import io.bitsquare.common.UserThread;
import io.bitsquare.common.util.Utilities;
import io.bitsquare.p2p.Message;
import io.bitsquare.p2p.network.*;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.keepalive.messages.Ping;
import io.bitsquare.p2p.peers.keepalive.messages.Pong;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class KeepAliveManager implements MessageListener {
private static final Logger log = LoggerFactory.getLogger(KeepAliveManager.class);
private static final int INTERVAL_SEC = new Random().nextInt(10) + 10;
private final NetworkNode networkNode;
private final PeerManager peerManager;
private ScheduledThreadPoolExecutor executor;
private final Map<String, KeepAliveHandler> maintenanceHandlerMap = new HashMap<>();
private boolean shutDownInProgress;
///////////////////////////////////////////////////////////////////////////////////////////
// Constructor
///////////////////////////////////////////////////////////////////////////////////////////
public KeepAliveManager(NetworkNode networkNode, PeerManager peerManager) {
this.networkNode = networkNode;
this.peerManager = peerManager;
networkNode.addMessageListener(this);
}
public void shutDown() {
Log.traceCall();
shutDownInProgress = true;
networkNode.removeMessageListener(this);
maintenanceHandlerMap.values().stream().forEach(KeepAliveHandler::cleanup);
if (executor != null)
MoreExecutors.shutdownAndAwaitTermination(executor, 100, TimeUnit.MILLISECONDS);
}
///////////////////////////////////////////////////////////////////////////////////////////
// API
///////////////////////////////////////////////////////////////////////////////////////////
public void start() {
if (executor == null) {
executor = Utilities.getScheduledThreadPoolExecutor("KeepAliveManager", 1, 2, 5);
executor.scheduleAtFixedRate(() -> UserThread.execute(this::keepAlive),
INTERVAL_SEC, INTERVAL_SEC, TimeUnit.SECONDS);
}
}
///////////////////////////////////////////////////////////////////////////////////////////
// MessageListener implementation
///////////////////////////////////////////////////////////////////////////////////////////
@Override
public void onMessage(Message message, Connection connection) {
if (message instanceof Ping) {
Log.traceCall(message.toString() + "\n\tconnection=" + connection);
Ping ping = (Ping) message;
Pong pong = new Pong(ping.nonce);
SettableFuture<Connection> future = networkNode.sendMessage(connection, pong);
Futures.addCallback(future, new FutureCallback<Connection>() {
@Override
public void onSuccess(Connection connection) {
log.trace("Pong sent successfully");
}
@Override
public void onFailure(@NotNull Throwable throwable) {
String errorMessage = "Sending pong to " + connection +
" failed. That is expected if the peer is offline. pong=" + pong + "." +
"Exception: " + throwable.getMessage();
log.info(errorMessage);
peerManager.handleConnectionFault(connection);
peerManager.shutDownConnection(connection, CloseConnectionReason.SEND_MSG_FAILURE);
}
});
}
}
///////////////////////////////////////////////////////////////////////////////////////////
// Private
///////////////////////////////////////////////////////////////////////////////////////////
private void keepAlive() {
Log.traceCall();
if (!shutDownInProgress) {
networkNode.getConfirmedConnections().stream()
.filter(connection -> connection instanceof OutboundConnection)
.forEach(connection -> {
if (!maintenanceHandlerMap.containsKey(connection.getUid())) {
KeepAliveHandler keepAliveHandler = new KeepAliveHandler(networkNode, peerManager, new KeepAliveHandler.Listener() {
@Override
public void onComplete() {
maintenanceHandlerMap.remove(connection.getUid());
}
@Override
public void onFault(String errorMessage, Connection connection) {
maintenanceHandlerMap.remove(connection.getUid());
}
});
maintenanceHandlerMap.put(connection.getUid(), keepAliveHandler);
keepAliveHandler.sendPing(connection);
} else {
log.warn("Connection with id {} has not completed and is still in our map. " +
"We will try to ping that peer at the next schedule.", connection.getUid());
}
});
int size = maintenanceHandlerMap.size();
log.info("maintenanceHandlerMap size=" + size);
if (size > peerManager.getMaxConnections())
log.warn("Seems we don't clean up out map correctly.\n" +
"maintenanceHandlerMap size={}, peerManager.getMaxConnections()={}", size, peerManager.getMaxConnections());
}
}
}

6
network/src/main/java/io/bitsquare/p2p/peers/messages/maintenance/MaintenanceMessage.java → network/src/main/java/io/bitsquare/p2p/peers/keepalive/messages/KeepAliveMessage.java
View File

@ -1,9 +1,9 @@
package io.bitsquare.p2p.peers.messages.maintenance;
package io.bitsquare.p2p.peers.keepalive.messages;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.Message;
public abstract class MaintenanceMessage implements Message {
public abstract class KeepAliveMessage implements Message {
private final int messageVersion = Version.getP2PMessageVersion();
@Override
@ -13,7 +13,7 @@ public abstract class MaintenanceMessage implements Message {
@Override
public String toString() {
return "MaintenanceMessage{" +
return "KeepAliveMessage{" +
"messageVersion=" + messageVersion +
'}';
}

21
network/src/main/java/io/bitsquare/p2p/peers/keepalive/messages/Ping.java Normal file
View File

@ -0,0 +1,21 @@
package io.bitsquare.p2p.peers.keepalive.messages;
import io.bitsquare.app.Version;
public final class Ping extends KeepAliveMessage {
// That object is sent over the wire, so we need to take care of version compatibility.
private static final long serialVersionUID = Version.P2P_NETWORK_VERSION;
public int nonce;
public Ping(int nonce) {
this.nonce = nonce;
}
@Override
public String toString() {
return "PingRequest{" +
", nonce=" + nonce +
"} " + super.toString();
}
}

21
network/src/main/java/io/bitsquare/p2p/peers/keepalive/messages/Pong.java Normal file
View File

@ -0,0 +1,21 @@
package io.bitsquare.p2p.peers.keepalive.messages;
import io.bitsquare.app.Version;
public final class Pong extends KeepAliveMessage {
// That object is sent over the wire, so we need to take care of version compatibility.
private static final long serialVersionUID = Version.P2P_NETWORK_VERSION;
public final int requestNonce;
public Pong(int requestNonce) {
this.requestNonce = requestNonce;
}
@Override
public String toString() {
return "PongResponse{" +
"requestNonce=" + requestNonce +
"} " + super.toString();
}
}

37
network/src/main/java/io/bitsquare/p2p/peers/messages/maintenance/PingRequest.java
View File

@ -1,37 +0,0 @@
package io.bitsquare.p2p.peers.messages.maintenance;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.network.messages.SendersNodeAddressMessage;
import io.bitsquare.p2p.peers.ReportedPeer;
import java.util.HashSet;
public final class PingRequest extends MaintenanceMessage implements SendersNodeAddressMessage {
// That object is sent over the wire, so we need to take care of version compatibility.
private static final long serialVersionUID = Version.P2P_NETWORK_VERSION;
private final NodeAddress senderNodeAddress;
public long nonce;
public final HashSet<ReportedPeer> reportedPeers;
public PingRequest(NodeAddress senderNodeAddress, long nonce, HashSet<ReportedPeer> reportedPeers) {
this.senderNodeAddress = senderNodeAddress;
this.nonce = nonce;
this.reportedPeers = reportedPeers;
}
@Override
public NodeAddress getSenderNodeAddress() {
return senderNodeAddress;
}
@Override
public String toString() {
return "GetPeersRequest{" +
"senderNodeAddress=" + senderNodeAddress +
", nonce=" + nonce +
", reportedPeers.size()=" + reportedPeers.size() +
"} " + super.toString();
}
}

27
network/src/main/java/io/bitsquare/p2p/peers/messages/maintenance/PongResponse.java
View File

@ -1,27 +0,0 @@
package io.bitsquare.p2p.peers.messages.maintenance;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.peers.ReportedPeer;
import java.util.HashSet;
public final class PongResponse extends MaintenanceMessage {
// That object is sent over the wire, so we need to take care of version compatibility.
private static final long serialVersionUID = Version.P2P_NETWORK_VERSION;
public final long requestNonce;
public final HashSet<ReportedPeer> reportedPeers;
public PongResponse(long requestNonce, HashSet<ReportedPeer> reportedPeers) {
this.requestNonce = requestNonce;
this.reportedPeers = reportedPeers;
}
@Override
public String toString() {
return "GetPeersResponse{" +
"requestNonce=" + requestNonce +
", reportedPeers.size()=" + reportedPeers.size() +
"} " + super.toString();
}
}

13
network/src/main/java/io/bitsquare/p2p/peers/GetPeersRequestHandler.java → network/src/main/java/io/bitsquare/p2p/peers/peerexchange/GetPeersRequestHandler.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers;
package io.bitsquare.p2p.peers.peerexchange;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
@ -8,8 +8,9 @@ import io.bitsquare.common.UserThread;
import io.bitsquare.p2p.network.CloseConnectionReason;
import io.bitsquare.p2p.network.Connection;
import io.bitsquare.p2p.network.NetworkNode;
import io.bitsquare.p2p.peers.messages.peers.GetPeersRequest;
import io.bitsquare.p2p.peers.messages.peers.GetPeersResponse;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.peerexchange.messages.GetPeersRequest;
import io.bitsquare.p2p.peers.peerexchange.messages.GetPeersResponse;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -61,7 +62,7 @@ public class GetPeersRequestHandler {
// API
///////////////////////////////////////////////////////////////////////////////////////////
public void process(GetPeersRequest getPeersRequest, final Connection connection) {
public void handle(GetPeersRequest getPeersRequest, final Connection connection) {
Log.traceCall("getPeersRequest=" + getPeersRequest + "\n\tconnection=" + connection + "\n\tthis=" + this);
checkArgument(connection.getPeersNodeAddressOptional().isPresent(),
@ -80,8 +81,8 @@ public class GetPeersRequestHandler {
@Override
public void onFailure(@NotNull Throwable throwable) {
String errorMessage = "Sending getPeersRequest to " + connection +
" failed. That is expected if the peer is offline. getPeersRequest=" + getPeersRequest + "." +
String errorMessage = "Sending getPeersResponse to " + connection +
" failed. That is expected if the peer is offline. getPeersResponse=" + getPeersResponse + "." +
"Exception: " + throwable.getMessage();
log.info(errorMessage);
handleFault(errorMessage, CloseConnectionReason.SEND_MSG_FAILURE, connection);

19
network/src/main/java/io/bitsquare/p2p/peers/PeerExchangeHandler.java → network/src/main/java/io/bitsquare/p2p/peers/peerexchange/PeerExchangeHandler.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers;
package io.bitsquare.p2p.peers.peerexchange;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
@ -11,8 +11,9 @@ import io.bitsquare.p2p.network.CloseConnectionReason;
import io.bitsquare.p2p.network.Connection;
import io.bitsquare.p2p.network.MessageListener;
import io.bitsquare.p2p.network.NetworkNode;
import io.bitsquare.p2p.peers.messages.peers.GetPeersRequest;
import io.bitsquare.p2p.peers.messages.peers.GetPeersResponse;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.peerexchange.messages.GetPeersRequest;
import io.bitsquare.p2p.peers.peerexchange.messages.GetPeersResponse;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -49,7 +50,7 @@ public class PeerExchangeHandler implements MessageListener {
private final NetworkNode networkNode;
private final PeerManager peerManager;
private final Listener listener;
private final long nonce = new Random().nextLong();
private final int nonce = new Random().nextInt();
private Timer timeoutTimer;
public Connection connection;
@ -62,8 +63,6 @@ public class PeerExchangeHandler implements MessageListener {
this.networkNode = networkNode;
this.peerManager = peerManager;
this.listener = listener;
//networkNode.addMessageListener(this);
}
public void cleanup() {
@ -81,9 +80,9 @@ public class PeerExchangeHandler implements MessageListener {
// API
///////////////////////////////////////////////////////////////////////////////////////////
public void requestConnectedPeers(NodeAddress nodeAddress) {
public void sendGetPeersRequest(NodeAddress nodeAddress) {
Log.traceCall("nodeAddress=" + nodeAddress + " / this=" + this);
checkNotNull(networkNode.getNodeAddress(), "PeerExchangeHandshake.requestReportedPeers: My node address must " +
checkNotNull(networkNode.getNodeAddress(), "PeerExchangeHandler.requestReportedPeers: My node address must " +
"not be null at requestReportedPeers");
GetPeersRequest getPeersRequest = new GetPeersRequest(networkNode.getNodeAddress(), nonce, peerManager.getConnectedPeersNonSeedNodes(nodeAddress));
SettableFuture<Connection> future = networkNode.sendMessage(nodeAddress, getPeersRequest);
@ -108,7 +107,7 @@ public class PeerExchangeHandler implements MessageListener {
checkArgument(timeoutTimer == null, "requestReportedPeers must not be called twice.");
timeoutTimer = UserThread.runAfter(() -> {
String errorMessage = "A timeout occurred at sending getPeersRequest:" + getPeersRequest + " for nodeAddress:" + nodeAddress;
log.info(errorMessage + " / PeerExchangeHandshake=" +
log.info(errorMessage + " / PeerExchangeHandler=" +
PeerExchangeHandler.this);
log.info("timeoutTimer called on " + this);
handleFault(errorMessage, CloseConnectionReason.SEND_MSG_TIMEOUT, nodeAddress);
@ -133,7 +132,7 @@ public class PeerExchangeHandler implements MessageListener {
cleanup();
listener.onComplete();
} else {
log.trace("Nonce not matching. That message is not intended for us.\n\t" +
log.warn("Nonce not matching. That should never happen.\n\t" +
"We drop that message. nonce={} / requestNonce={}",
nonce, getPeersResponse.requestNonce);
}

17
network/src/main/java/io/bitsquare/p2p/peers/PeerExchangeManager.java → network/src/main/java/io/bitsquare/p2p/peers/peerexchange/PeerExchangeManager.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers;
package io.bitsquare.p2p.peers.peerexchange;
import com.google.common.util.concurrent.MoreExecutors;
import io.bitsquare.app.Log;
@ -7,7 +7,8 @@ import io.bitsquare.common.util.Utilities;
import io.bitsquare.p2p.Message;
import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.network.*;
import io.bitsquare.p2p.peers.messages.peers.GetPeersRequest;
import io.bitsquare.p2p.peers.PeerManager;
import io.bitsquare.p2p.peers.peerexchange.messages.GetPeersRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -24,7 +25,7 @@ public class PeerExchangeManager implements MessageListener, ConnectionListener
private static final Logger log = LoggerFactory.getLogger(PeerExchangeManager.class);
private static final long RETRY_DELAY_SEC = 60;
private static final long MAINTENANCE_DELAY_SEC = 5;
private static final long REQUEST_PERIODICALLY_INTERVAL_MINUTES = 10;
private final NetworkNode networkNode;
private final PeerManager peerManager;
@ -33,7 +34,7 @@ public class PeerExchangeManager implements MessageListener, ConnectionListener
private Timer connectToMorePeersTimer;
private boolean shutDownInProgress;
private ScheduledThreadPoolExecutor executor;
///////////////////////////////////////////////////////////////////////////////////////////
// Constructor
@ -76,7 +77,7 @@ public class PeerExchangeManager implements MessageListener, ConnectionListener
if (executor == null) {
executor = Utilities.getScheduledThreadPoolExecutor("PeerExchangeManager", 1, 2, 5);
executor.scheduleAtFixedRate(() -> UserThread.execute(this::requestAgain),
MAINTENANCE_DELAY_SEC, MAINTENANCE_DELAY_SEC, TimeUnit.SECONDS);
REQUEST_PERIODICALLY_INTERVAL_MINUTES, REQUEST_PERIODICALLY_INTERVAL_MINUTES, TimeUnit.MINUTES);
}
}
@ -126,7 +127,7 @@ public class PeerExchangeManager implements MessageListener, ConnectionListener
peerManager.handleConnectionFault(connection);
}
});
getPeersRequestHandler.process((GetPeersRequest) message, connection);
getPeersRequestHandler.handle((GetPeersRequest) message, connection);
}
}
@ -182,7 +183,7 @@ public class PeerExchangeManager implements MessageListener, ConnectionListener
}
});
peerExchangeHandlerMap.put(nodeAddress, peerExchangeHandler);
peerExchangeHandler.requestConnectedPeers(nodeAddress);
peerExchangeHandler.sendGetPeersRequest(nodeAddress);
} else {
//TODO check when that happens
log.warn("We have started already a peerExchangeHandshake. " +
@ -237,7 +238,7 @@ public class PeerExchangeManager implements MessageListener, ConnectionListener
///////////////////////////////////////////////////////////////////////////////////////////
private void requestAgain() {
checkNotNull(networkNode.getNodeAddress(), "My node address must not be null at sendUpdateRequest");
checkNotNull(networkNode.getNodeAddress(), "My node address must not be null at requestAgain");
Set<NodeAddress> candidates = new HashSet<>(getNodeAddresses(peerManager.getReportedPeers()));
candidates.addAll(getNodeAddresses(peerManager.getPersistedPeers()));
candidates.addAll(seedNodeAddresses);

21
network/src/main/java/io/bitsquare/p2p/peers/ReportedPeer.java → network/src/main/java/io/bitsquare/p2p/peers/peerexchange/ReportedPeer.java
View File

@ -1,34 +1,23 @@
package io.bitsquare.p2p.peers;
package io.bitsquare.p2p.peers.peerexchange;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.NodeAddress;
import java.io.Serializable;
import java.util.Date;
import java.util.concurrent.TimeUnit;
public class ReportedPeer implements Serializable {
// That object is sent over the wire, so we need to take care of version compatibility.
private static final long serialVersionUID = Version.P2P_NETWORK_VERSION;
public final NodeAddress nodeAddress;
public Date lastActivityDate;
public Date date;
public ReportedPeer(NodeAddress nodeAddress, Date lastActivityDate) {
public ReportedPeer(NodeAddress nodeAddress) {
this.nodeAddress = nodeAddress;
this.lastActivityDate = lastActivityDate;
this.date = new Date();
}
public void penalizeLastActivityDate() {
if (lastActivityDate != null) {
long now = new Date().getTime();
long diff = Math.max(TimeUnit.DAYS.toMillis(1), now - lastActivityDate.getTime());
long reduced = now - diff * 2;
lastActivityDate = new Date(reduced);
}
}
// We don't use the lastActivityDate for identity
@Override
public boolean equals(Object o) {
if (this == o) return true;
@ -50,7 +39,7 @@ public class ReportedPeer implements Serializable {
public String toString() {
return "ReportedPeer{" +
"address=" + nodeAddress +
", lastActivityDate=" + lastActivityDate +
", date=" + date +
'}';
}
}

8
network/src/main/java/io/bitsquare/p2p/peers/messages/peers/GetPeersRequest.java → network/src/main/java/io/bitsquare/p2p/peers/peerexchange/messages/GetPeersRequest.java
View File

@ -1,9 +1,9 @@
package io.bitsquare.p2p.peers.messages.peers;
package io.bitsquare.p2p.peers.peerexchange.messages;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.NodeAddress;
import io.bitsquare.p2p.network.messages.SendersNodeAddressMessage;
import io.bitsquare.p2p.peers.ReportedPeer;
import io.bitsquare.p2p.peers.peerexchange.ReportedPeer;
import java.util.HashSet;
@ -12,10 +12,10 @@ public final class GetPeersRequest extends PeerExchangeMessage implements Sender
private static final long serialVersionUID = Version.P2P_NETWORK_VERSION;
private final NodeAddress senderNodeAddress;
public long nonce;
public int nonce;
public final HashSet<ReportedPeer> reportedPeers;
public GetPeersRequest(NodeAddress senderNodeAddress, long nonce, HashSet<ReportedPeer> reportedPeers) {
public GetPeersRequest(NodeAddress senderNodeAddress, int nonce, HashSet<ReportedPeer> reportedPeers) {
this.senderNodeAddress = senderNodeAddress;
this.nonce = nonce;
this.reportedPeers = reportedPeers;

8
network/src/main/java/io/bitsquare/p2p/peers/messages/peers/GetPeersResponse.java → network/src/main/java/io/bitsquare/p2p/peers/peerexchange/messages/GetPeersResponse.java
View File

@ -1,7 +1,7 @@
package io.bitsquare.p2p.peers.messages.peers;
package io.bitsquare.p2p.peers.peerexchange.messages;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.peers.ReportedPeer;
import io.bitsquare.p2p.peers.peerexchange.ReportedPeer;
import java.util.HashSet;
@ -9,10 +9,10 @@ public final class GetPeersResponse extends PeerExchangeMessage {
// That object is sent over the wire, so we need to take care of version compatibility.
private static final long serialVersionUID = Version.P2P_NETWORK_VERSION;
public final long requestNonce;
public final int requestNonce;
public final HashSet<ReportedPeer> reportedPeers;
public GetPeersResponse(long requestNonce, HashSet<ReportedPeer> reportedPeers) {
public GetPeersResponse(int requestNonce, HashSet<ReportedPeer> reportedPeers) {
this.requestNonce = requestNonce;
this.reportedPeers = reportedPeers;
}

2
network/src/main/java/io/bitsquare/p2p/peers/messages/peers/PeerExchangeMessage.java → network/src/main/java/io/bitsquare/p2p/peers/peerexchange/messages/PeerExchangeMessage.java
View File

@ -1,4 +1,4 @@
package io.bitsquare.p2p.peers.messages.peers;
package io.bitsquare.p2p.peers.peerexchange.messages;
import io.bitsquare.app.Version;
import io.bitsquare.p2p.Message;