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

Fix bug with missing broadcast

Browse source
This commit is contained in:
Manfred Karrer 2016-02-27 00:44:33 +01:00
parent 891799b09c
commit b511267340
14 changed files with 44 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -556,7 +556,7 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
public void onBroadcastFailed(String errorMessage) {
}
};
boolean result = p2PDataStorage.add(protectedMailboxStorageEntry, networkNode.getNodeAddress(), listener, true, true);
boolean result = p2PDataStorage.add(protectedMailboxStorageEntry, networkNode.getNodeAddress(), listener, true);
if (!result) {
//TODO remove and add again with a delay to ensure the data will be broadcasted
sendMailboxMessageListener.onFault("Data already exists in our local database");
@ -616,13 +616,13 @@ public class P2PService implements SetupListener, MessageListener, ConnectionLis
// Data storage
///////////////////////////////////////////////////////////////////////////////////////////
public boolean addData(StoragePayload storagePayload, boolean forceBroadcast, boolean isDataOwner) {
public boolean addData(StoragePayload storagePayload, boolean isDataOwner) {
Log.traceCall();
checkArgument(optionalKeyRing.isPresent(), "keyRing not set. Seems that is called on a seed node which must not happen.");
if (isBootstrapped()) {
try {
ProtectedStorageEntry protectedStorageEntry = p2PDataStorage.getProtectedData(storagePayload, optionalKeyRing.get().getSignatureKeyPair());
return p2PDataStorage.add(protectedStorageEntry, networkNode.getNodeAddress(), null, forceBroadcast, isDataOwner);
return p2PDataStorage.add(protectedStorageEntry, networkNode.getNodeAddress(), null, isDataOwner);
} catch (CryptoException e) {
log.error("Signing at getDataWithSignedSeqNr failed. That should never happen.");
return false;

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

@ -2,19 +2,19 @@ package io.bitsquare.p2p.network;
public enum CloseConnectionReason {
// First block are from different exceptions
SOCKET_CLOSED(false),
RESET(false),
SOCKET_TIMEOUT(false),
TERMINATED(false), // EOFException
UNKNOWN_EXCEPTION(false),
SOCKET_CLOSED(false, false),
RESET(false, false),
SOCKET_TIMEOUT(false, false),
TERMINATED(false, false), // EOFException
UNKNOWN_EXCEPTION(false, false),
// Planned
APP_SHUT_DOWN(true, true),
CLOSE_REQUESTED_BY_PEER(false, true),
// send msg
SEND_MSG_FAILURE(false),
SEND_MSG_TIMEOUT(false),
SEND_MSG_FAILURE(false, false),
SEND_MSG_TIMEOUT(false, false),
// maintenance
TOO_MANY_CONNECTIONS_OPEN(true, true),
@ -27,10 +27,6 @@ public enum CloseConnectionReason {
public final boolean sendCloseMessage;
public boolean isIntended;
CloseConnectionReason(boolean sendCloseMessage) {
this(sendCloseMessage, true);
}
CloseConnectionReason(boolean sendCloseMessage, boolean isIntended) {
this.sendCloseMessage = sendCloseMessage;
this.isIntended = isIntended;

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

@ -1,5 +1,6 @@
package io.bitsquare.p2p.network;
import com.google.common.util.concurrent.CycleDetectingLockFactory;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.Uninterruptibles;
import io.bitsquare.app.Log;
@ -68,6 +69,8 @@ public class Connection implements MessageListener {
return MAX_MSG_SIZE;
}
private static final CycleDetectingLockFactory cycleDetectingLockFactory = CycleDetectingLockFactory.newInstance(CycleDetectingLockFactory.Policies.THROW);
///////////////////////////////////////////////////////////////////////////////////////////
// Class fields
@ -79,7 +82,7 @@ public class Connection implements MessageListener {
private final String portInfo;
private final String uid;
private final ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
private final ReentrantLock objectOutputStreamLock = new ReentrantLock(true);
private final ReentrantLock objectOutputStreamLock = cycleDetectingLockFactory.newReentrantLock("objectOutputStreamLock");
// holder of state shared between InputHandler and Connection
private final SharedModel sharedModel;
private final Statistic statistic;
@ -104,7 +107,6 @@ public class Connection implements MessageListener {
Connection(Socket socket, MessageListener messageListener, ConnectionListener connectionListener,
@Nullable NodeAddress peersNodeAddress) {
this.socket = socket;
//this.messageListener = messageListener;
this.connectionListener = connectionListener;
uid = UUID.randomUUID().toString();
statistic = new Statistic();

2
network/src/main/java/io/bitsquare/p2p/peers/PeerManager.java
View file

@ -47,7 +47,7 @@ public class PeerManager implements ConnectionListener {
}
static {
setMaxConnections(10);
setMaxConnections(12);
}
private static final int MAX_REPORTED_PEERS = 1000;

2
network/src/main/java/io/bitsquare/p2p/peers/getdata/RequestDataHandler.java
View file

@ -157,7 +157,7 @@ public class RequestDataHandler implements MessageListener {
"at that moment");
((GetDataResponse) message).dataSet.stream()
.forEach(protectedData -> dataStorage.add(protectedData,
connection.getPeersNodeAddressOptional().get(), null, false, false));
connection.getPeersNodeAddressOptional().get(), null, false));
cleanup();
listener.onComplete();

22
network/src/main/java/io/bitsquare/p2p/storage/P2PDataStorage.java
View file

@ -113,7 +113,7 @@ public class P2PDataStorage implements MessageListener, ConnectionListener {
Log.traceCall(StringUtils.abbreviate(message.toString(), 100) + "\n\tconnection=" + connection);
connection.getPeersNodeAddressOptional().ifPresent(peersNodeAddress -> {
if (message instanceof AddDataMessage) {
add(((AddDataMessage) message).protectedStorageEntry, peersNodeAddress, null, false, false);
add(((AddDataMessage) message).protectedStorageEntry, peersNodeAddress, null, false);
} else if (message instanceof RemoveDataMessage) {
remove(((RemoveDataMessage) message).protectedStorageEntry, peersNodeAddress, false);
} else if (message instanceof RemoveMailboxDataMessage) {
@ -136,7 +136,7 @@ public class P2PDataStorage implements MessageListener, ConnectionListener {
@Override
public void onDisconnect(CloseConnectionReason closeConnectionReason, Connection connection) {
if (connection.getPeersNodeAddressOptional().isPresent() && !closeConnectionReason.isIntended) {
if (connection.hasPeersNodeAddress() && !closeConnectionReason.isIntended) {
map.values().stream()
.forEach(protectedData -> {
ExpirablePayload expirablePayload = protectedData.getStoragePayload();
@ -151,6 +151,8 @@ public class P2PDataStorage implements MessageListener, ConnectionListener {
ByteArray hashOfPayload = getHashAsByteArray(expirablePayload);
boolean containsKey = map.containsKey(hashOfPayload);
if (containsKey) {
log.info("We remove the data as the data owner got disconnected with " +
"closeConnectionReason=" + closeConnectionReason);
doRemoveProtectedExpirableData(protectedData, hashOfPayload);
} else {
log.debug("Remove data ignored as we don't have an entry for that data.");
@ -172,13 +174,14 @@ public class P2PDataStorage implements MessageListener, ConnectionListener {
///////////////////////////////////////////////////////////////////////////////////////////
public boolean add(ProtectedStorageEntry protectedStorageEntry, @Nullable NodeAddress sender,
@Nullable BroadcastHandler.Listener listener, boolean forceBroadcast, boolean isDataOwner) {
@Nullable BroadcastHandler.Listener listener, boolean isDataOwner) {
Log.traceCall();
ByteArray hashOfPayload = getHashAsByteArray(protectedStorageEntry.getStoragePayload());
boolean sequenceNrValid = isSequenceNrValid(protectedStorageEntry.sequenceNumber, hashOfPayload);
boolean result = checkPublicKeys(protectedStorageEntry, true)
&& checkSignature(protectedStorageEntry)
&& isSequenceNrValid(protectedStorageEntry.sequenceNumber, hashOfPayload);
&& sequenceNrValid;
boolean containsKey = map.containsKey(hashOfPayload);
if (containsKey)
@ -197,10 +200,7 @@ public class P2PDataStorage implements MessageListener, ConnectionListener {
log.trace(sb.toString());
log.info("Data set after doAdd: size=" + map.values().size());
if (!containsKey || forceBroadcast)
broadcast(new AddDataMessage(protectedStorageEntry), sender, listener, isDataOwner);
else
log.trace("Not broadcasting data as we had it already in our map.");
broadcast(new AddDataMessage(protectedStorageEntry), sender, listener, isDataOwner);
hashMapChangedListeners.stream().forEach(e -> e.onAdded(protectedStorageEntry));
} else {
@ -383,13 +383,13 @@ public class P2PDataStorage implements MessageListener, ConnectionListener {
private boolean isSequenceNrValid(int newSequenceNumber, ByteArray hashOfData) {
if (sequenceNumberMap.containsKey(hashOfData)) {
Integer storedSequenceNumber = sequenceNumberMap.get(hashOfData).sequenceNr;
if (newSequenceNumber < storedSequenceNumber) {
if (newSequenceNumber > storedSequenceNumber) {
return true;
} else {
log.info("Sequence number is invalid. sequenceNumber = "
+ newSequenceNumber + " / storedSequenceNumber=" + storedSequenceNumber + "\n" +
"That can happen if the data owner gets an old delayed data storage message.");
return false;
} else {
return true;
}
} else {
return true;

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

@ -99,7 +99,7 @@ public class ProtectedDataStorageTest {
//@Test
public void testAddAndRemove() throws InterruptedException, NoSuchAlgorithmException, CertificateException, KeyStoreException, IOException, CryptoException, SignatureException, InvalidKeyException, NoSuchProviderException {
ProtectedStorageEntry data = dataStorage1.getProtectedData(mockData, storageSignatureKeyPair1);
Assert.assertTrue(dataStorage1.add(data, null, null, true, true));
Assert.assertTrue(dataStorage1.add(data, null, null, true));
Assert.assertEquals(1, dataStorage1.getMap().size());
int newSequenceNumber = data.sequenceNumber + 1;
@ -115,7 +115,7 @@ public class ProtectedDataStorageTest {
mockData.ttl = (int) (P2PDataStorage.CHECK_TTL_INTERVAL_SEC * 1.5);
ProtectedStorageEntry data = dataStorage1.getProtectedData(mockData, storageSignatureKeyPair1);
log.debug("data.date " + data.timeStamp);
Assert.assertTrue(dataStorage1.add(data, null, null, true, true));
Assert.assertTrue(dataStorage1.add(data, null, null, true));
log.debug("test 1");
Assert.assertEquals(1, dataStorage1.getMap().size());
@ -163,7 +163,7 @@ public class ProtectedDataStorageTest {
public void testRefreshTTL() throws InterruptedException, NoSuchAlgorithmException, CertificateException, KeyStoreException, IOException, CryptoException, SignatureException, InvalidKeyException, NoSuchProviderException {
mockData.ttl = (int) (P2PDataStorage.CHECK_TTL_INTERVAL_SEC * 1.5);
ProtectedStorageEntry data = dataStorage1.getProtectedData(mockData, storageSignatureKeyPair1);
Assert.assertTrue(dataStorage1.add(data, null, null, true, true));
Assert.assertTrue(dataStorage1.add(data, null, null, true));
Assert.assertEquals(1, dataStorage1.getMap().size());
Thread.sleep(P2PDataStorage.CHECK_TTL_INTERVAL_SEC);
log.debug("test 1");