From fe1df6dafad2e50a0e8eb4621e8549308e72b314 Mon Sep 17 00:00:00 2001 From: akwizgran Date: Thu, 6 Jun 2019 16:39:54 +0100 Subject: [PATCH] Move pending contact events to rendezvous poller. --- .../api/rendezvous/RendezvousPoller.java | 12 + .../event/RendezvousFailedEvent.java | 25 -- .../rendezvous/event/RendezvousPollEvent.java | 36 +++ .../bramble/contact/ContactManagerImpl.java | 61 +---- .../bramble/rendezvous/RendezvousModule.java | 1 + .../bramble/rendezvous/RendezvousPoller.java | 7 - .../rendezvous/RendezvousPollerImpl.java | 72 ++++- .../contact/ContactManagerImplTest.java | 150 +---------- .../rendezvous/RendezvousPollerImplTest.java | 250 ++++++++++++++++-- 9 files changed, 348 insertions(+), 266 deletions(-) create mode 100644 bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/RendezvousPoller.java delete mode 100644 bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousFailedEvent.java create mode 100644 bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousPollEvent.java delete mode 100644 bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPoller.java diff --git a/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/RendezvousPoller.java b/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/RendezvousPoller.java new file mode 100644 index 000000000..e6f473fdf --- /dev/null +++ b/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/RendezvousPoller.java @@ -0,0 +1,12 @@ +package org.briarproject.bramble.api.rendezvous; + +import org.briarproject.bramble.api.plugin.TransportId; + +/** + * Interface for the poller that makes rendezvous connections to pending + * contacts. + */ +public interface RendezvousPoller { + + long getLastPollTime(TransportId t); +} diff --git a/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousFailedEvent.java b/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousFailedEvent.java deleted file mode 100644 index 0530db1e2..000000000 --- a/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousFailedEvent.java +++ /dev/null @@ -1,25 +0,0 @@ -package org.briarproject.bramble.api.rendezvous.event; - -import org.briarproject.bramble.api.contact.PendingContactId; -import org.briarproject.bramble.api.event.Event; -import org.briarproject.bramble.api.nullsafety.NotNullByDefault; - -import javax.annotation.concurrent.Immutable; - -/** - * An event that is broadcast when a rendezvous with a pending contact fails. - */ -@Immutable -@NotNullByDefault -public class RendezvousFailedEvent extends Event { - - private final PendingContactId pendingContactId; - - public RendezvousFailedEvent(PendingContactId pendingContactId) { - this.pendingContactId = pendingContactId; - } - - public PendingContactId getPendingContactId() { - return pendingContactId; - } -} diff --git a/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousPollEvent.java b/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousPollEvent.java new file mode 100644 index 000000000..0281a9017 --- /dev/null +++ b/bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousPollEvent.java @@ -0,0 +1,36 @@ +package org.briarproject.bramble.api.rendezvous.event; + +import org.briarproject.bramble.api.contact.PendingContactId; +import org.briarproject.bramble.api.event.Event; +import org.briarproject.bramble.api.nullsafety.NotNullByDefault; +import org.briarproject.bramble.api.plugin.TransportId; + +import java.util.Collection; + +import javax.annotation.concurrent.Immutable; + +/** + * An event that is broadcast when a transport plugin is polled for connections + * to one or more pending contacts. + */ +@Immutable +@NotNullByDefault +public class RendezvousPollEvent extends Event { + + private final TransportId transportId; + private final Collection pendingContacts; + + public RendezvousPollEvent(TransportId transportId, + Collection pendingContacts) { + this.transportId = transportId; + this.pendingContacts = pendingContacts; + } + + public TransportId getTransportId() { + return transportId; + } + + public Collection getPendingContacts() { + return pendingContacts; + } +} diff --git a/bramble-core/src/main/java/org/briarproject/bramble/contact/ContactManagerImpl.java b/bramble-core/src/main/java/org/briarproject/bramble/contact/ContactManagerImpl.java index 5c9ec6fa5..f3138fd75 100644 --- a/bramble-core/src/main/java/org/briarproject/bramble/contact/ContactManagerImpl.java +++ b/bramble-core/src/main/java/org/briarproject/bramble/contact/ContactManagerImpl.java @@ -17,7 +17,6 @@ import org.briarproject.bramble.api.db.DbException; import org.briarproject.bramble.api.db.NoSuchContactException; import org.briarproject.bramble.api.db.Transaction; import org.briarproject.bramble.api.event.Event; -import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.event.EventListener; import org.briarproject.bramble.api.identity.Author; import org.briarproject.bramble.api.identity.AuthorId; @@ -25,25 +24,20 @@ import org.briarproject.bramble.api.identity.AuthorInfo; import org.briarproject.bramble.api.identity.IdentityManager; import org.briarproject.bramble.api.identity.LocalAuthor; import org.briarproject.bramble.api.nullsafety.NotNullByDefault; -import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionClosedEvent; -import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionOpenedEvent; -import org.briarproject.bramble.api.rendezvous.event.RendezvousFailedEvent; import org.briarproject.bramble.api.transport.KeyManager; import java.security.GeneralSecurityException; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; import javax.annotation.Nullable; import javax.annotation.concurrent.ThreadSafe; import javax.inject.Inject; -import static org.briarproject.bramble.api.contact.PendingContactState.ADDING_CONTACT; -import static org.briarproject.bramble.api.contact.PendingContactState.FAILED; import static org.briarproject.bramble.api.contact.PendingContactState.WAITING_FOR_CONNECTION; import static org.briarproject.bramble.api.identity.AuthorConstants.MAX_AUTHOR_NAME_LENGTH; import static org.briarproject.bramble.api.identity.AuthorInfo.Status.OURSELVES; @@ -60,23 +54,20 @@ class ContactManagerImpl implements ContactManager, EventListener { private final KeyManager keyManager; private final IdentityManager identityManager; private final PendingContactFactory pendingContactFactory; - private final EventBus eventBus; private final List hooks = new CopyOnWriteArrayList<>(); - private final ConcurrentMap states = + private final Map states = new ConcurrentHashMap<>(); @Inject ContactManagerImpl(DatabaseComponent db, KeyManager keyManager, IdentityManager identityManager, - PendingContactFactory pendingContactFactory, - EventBus eventBus) { + PendingContactFactory pendingContactFactory) { this.db = db; this.keyManager = keyManager; this.identityManager = identityManager; this.pendingContactFactory = pendingContactFactory; - this.eventBus = eventBus; } @Override @@ -156,7 +147,7 @@ class ContactManagerImpl implements ContactManager, EventListener { } finally { db.endTransaction(txn); } - setState(p.getId(), WAITING_FOR_CONNECTION); + states.put(p.getId(), WAITING_FOR_CONNECTION); return p; } @@ -286,46 +277,10 @@ class ContactManagerImpl implements ContactManager, EventListener { @Override public void eventOccurred(Event e) { - if (e instanceof RendezvousConnectionOpenedEvent) { - RendezvousConnectionOpenedEvent r = - (RendezvousConnectionOpenedEvent) e; - setStateConnected(r.getPendingContactId()); - } else if (e instanceof RendezvousConnectionClosedEvent) { - RendezvousConnectionClosedEvent r = - (RendezvousConnectionClosedEvent) e; - // We're only interested in failures - if the rendezvous succeeds - // the pending contact will be removed - if (!r.isSuccess()) setStateDisconnected(r.getPendingContactId()); - } else if (e instanceof RendezvousFailedEvent) { - RendezvousFailedEvent r = (RendezvousFailedEvent) e; - setState(r.getPendingContactId(), FAILED); - } - } - - /** - * Sets the state of the given pending contact and broadcasts an event. - */ - private void setState(PendingContactId p, PendingContactState state) { - states.put(p, state); - eventBus.broadcast(new PendingContactStateChangedEvent(p, state)); - } - - private void setStateConnected(PendingContactId p) { - // Set the state to ADDING_CONTACT if there's no current state or the - // current state is WAITING_FOR_CONNECTION - if (states.putIfAbsent(p, ADDING_CONTACT) == null || - states.replace(p, WAITING_FOR_CONNECTION, ADDING_CONTACT)) { - eventBus.broadcast(new PendingContactStateChangedEvent(p, - ADDING_CONTACT)); - } - } - - private void setStateDisconnected(PendingContactId p) { - // Set the state to WAITING_FOR_CONNECTION if the current state is - // ADDING_CONTACT - if (states.replace(p, ADDING_CONTACT, WAITING_FOR_CONNECTION)) { - eventBus.broadcast(new PendingContactStateChangedEvent(p, - WAITING_FOR_CONNECTION)); + if (e instanceof PendingContactStateChangedEvent) { + PendingContactStateChangedEvent p = + (PendingContactStateChangedEvent) e; + states.put(p.getId(), p.getPendingContactState()); } } } diff --git a/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousModule.java b/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousModule.java index 41bf14f66..2d868f881 100644 --- a/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousModule.java +++ b/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousModule.java @@ -2,6 +2,7 @@ package org.briarproject.bramble.rendezvous; import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.lifecycle.LifecycleManager; +import org.briarproject.bramble.api.rendezvous.RendezvousPoller; import javax.inject.Inject; import javax.inject.Singleton; diff --git a/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPoller.java b/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPoller.java deleted file mode 100644 index 4ecfc32a6..000000000 --- a/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPoller.java +++ /dev/null @@ -1,7 +0,0 @@ -package org.briarproject.bramble.rendezvous; - -/** - * Empty interface for injecting the rendezvous poller. - */ -interface RendezvousPoller { -} diff --git a/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPollerImpl.java b/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPollerImpl.java index b0d0c3461..797929fce 100644 --- a/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPollerImpl.java +++ b/bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPollerImpl.java @@ -4,8 +4,10 @@ import org.briarproject.bramble.PoliteExecutor; import org.briarproject.bramble.api.Pair; import org.briarproject.bramble.api.contact.PendingContact; import org.briarproject.bramble.api.contact.PendingContactId; +import org.briarproject.bramble.api.contact.PendingContactState; import org.briarproject.bramble.api.contact.event.PendingContactAddedEvent; import org.briarproject.bramble.api.contact.event.PendingContactRemovedEvent; +import org.briarproject.bramble.api.contact.event.PendingContactStateChangedEvent; import org.briarproject.bramble.api.crypto.KeyPair; import org.briarproject.bramble.api.crypto.SecretKey; import org.briarproject.bramble.api.crypto.TransportCrypto; @@ -34,7 +36,10 @@ import org.briarproject.bramble.api.plugin.event.TransportEnabledEvent; import org.briarproject.bramble.api.properties.TransportProperties; import org.briarproject.bramble.api.rendezvous.KeyMaterialSource; import org.briarproject.bramble.api.rendezvous.RendezvousEndpoint; -import org.briarproject.bramble.api.rendezvous.event.RendezvousFailedEvent; +import org.briarproject.bramble.api.rendezvous.RendezvousPoller; +import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionClosedEvent; +import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionOpenedEvent; +import org.briarproject.bramble.api.rendezvous.event.RendezvousPollEvent; import org.briarproject.bramble.api.system.Clock; import org.briarproject.bramble.api.system.Scheduler; @@ -45,6 +50,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; @@ -58,6 +64,9 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.logging.Level.INFO; import static java.util.logging.Level.WARNING; import static java.util.logging.Logger.getLogger; +import static org.briarproject.bramble.api.contact.PendingContactState.ADDING_CONTACT; +import static org.briarproject.bramble.api.contact.PendingContactState.FAILED; +import static org.briarproject.bramble.api.contact.PendingContactState.WAITING_FOR_CONNECTION; import static org.briarproject.bramble.api.nullsafety.NullSafety.requireNull; import static org.briarproject.bramble.rendezvous.RendezvousConstants.POLLING_INTERVAL_MS; import static org.briarproject.bramble.rendezvous.RendezvousConstants.RENDEZVOUS_TIMEOUT_MS; @@ -81,6 +90,9 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { private final Clock clock; private final AtomicBoolean used = new AtomicBoolean(false); + private final Map lastPollTimes = + new ConcurrentHashMap<>(); + // Executor that runs one task at a time private final Executor worker; // The following fields are only accessed on the worker @@ -113,6 +125,12 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { worker = new PoliteExecutor("RendezvousPoller", ioExecutor, 1); } + @Override + public long getLastPollTime(TransportId t) { + Long time = lastPollTimes.get(t); + return time == null ? 0 : time; + } + @Override public void startService() throws ServiceException { if (used.getAndSet(true)) throw new IllegalStateException(); @@ -140,8 +158,10 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { private void addPendingContact(PendingContact p) { long now = clock.currentTimeMillis(); long expiry = p.getTimestamp() + RENDEZVOUS_TIMEOUT_MS; - if (expiry <= now) { - eventBus.broadcast(new RendezvousFailedEvent(p.getId())); + if (expiry > now) { + broadcastState(p.getId(), WAITING_FOR_CONNECTION); + } else { + broadcastState(p.getId(), FAILED); return; } try { @@ -168,6 +188,10 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { } } + private void broadcastState(PendingContactId p, PendingContactState state) { + eventBus.broadcast(new PendingContactStateChangedEvent(p, state)); + } + @Nullable private RendezvousEndpoint createEndpoint(DuplexPlugin plugin, PendingContactId p, CryptoState cs) { @@ -195,7 +219,7 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { } for (PendingContactId p : expired) { removePendingContact(p); - eventBus.broadcast(new RendezvousFailedEvent(p)); + broadcastState(p, FAILED); } } @@ -211,16 +235,21 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { // Worker private void poll(PluginState ps) { + if (ps.endpoints.isEmpty()) return; + TransportId t = ps.plugin.getId(); List> properties = new ArrayList<>(); for (Entry e : ps.endpoints.entrySet()) { TransportProperties props = e.getValue().getRemoteTransportProperties(); - Handler h = new Handler(e.getKey(), ps.plugin.getId(), false); + Handler h = new Handler(e.getKey(), t, false); properties.add(new Pair<>(props, h)); } - if (!properties.isEmpty()) ps.plugin.poll(properties); + lastPollTimes.put(t, clock.currentTimeMillis()); + eventBus.broadcast(new RendezvousPollEvent(t, + new ArrayList<>(ps.endpoints.keySet()))); + ps.plugin.poll(properties); } @Override @@ -241,6 +270,14 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { } else if (e instanceof TransportDisabledEvent) { TransportDisabledEvent t = (TransportDisabledEvent) e; removeTransportAsync(t.getTransportId()); + } else if (e instanceof RendezvousConnectionOpenedEvent) { + RendezvousConnectionOpenedEvent r = + (RendezvousConnectionOpenedEvent) e; + connectionOpenedAsync(r.getPendingContactId()); + } else if (e instanceof RendezvousConnectionClosedEvent) { + RendezvousConnectionClosedEvent r = + (RendezvousConnectionClosedEvent) e; + if (!r.isSuccess()) connectionFailedAsync(r.getPendingContactId()); } } @@ -307,6 +344,29 @@ class RendezvousPollerImpl implements RendezvousPoller, Service, EventListener { } } + @EventExecutor + private void connectionOpenedAsync(PendingContactId p) { + worker.execute(() -> connectionOpened(p)); + } + + // Worker + private void connectionOpened(PendingContactId p) { + // Check that the pending contact hasn't expired + if (cryptoStates.containsKey(p)) broadcastState(p, ADDING_CONTACT); + } + + @EventExecutor + private void connectionFailedAsync(PendingContactId p) { + worker.execute(() -> connectionFailed(p)); + } + + // Worker + private void connectionFailed(PendingContactId p) { + // Check that the pending contact hasn't expired + if (cryptoStates.containsKey(p)) + broadcastState(p, WAITING_FOR_CONNECTION); + } + private static class PluginState { private final DuplexPlugin plugin; diff --git a/bramble-core/src/test/java/org/briarproject/bramble/contact/ContactManagerImplTest.java b/bramble-core/src/test/java/org/briarproject/bramble/contact/ContactManagerImplTest.java index b0a50ec16..c828f2698 100644 --- a/bramble-core/src/test/java/org/briarproject/bramble/contact/ContactManagerImplTest.java +++ b/bramble-core/src/test/java/org/briarproject/bramble/contact/ContactManagerImplTest.java @@ -5,26 +5,20 @@ import org.briarproject.bramble.api.contact.Contact; import org.briarproject.bramble.api.contact.ContactId; import org.briarproject.bramble.api.contact.PendingContact; import org.briarproject.bramble.api.contact.PendingContactState; -import org.briarproject.bramble.api.contact.event.PendingContactStateChangedEvent; import org.briarproject.bramble.api.crypto.KeyPair; import org.briarproject.bramble.api.crypto.SecretKey; import org.briarproject.bramble.api.db.DatabaseComponent; import org.briarproject.bramble.api.db.DbException; import org.briarproject.bramble.api.db.NoSuchContactException; import org.briarproject.bramble.api.db.Transaction; -import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.identity.Author; import org.briarproject.bramble.api.identity.AuthorId; import org.briarproject.bramble.api.identity.AuthorInfo; import org.briarproject.bramble.api.identity.IdentityManager; import org.briarproject.bramble.api.identity.LocalAuthor; -import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionClosedEvent; -import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionOpenedEvent; -import org.briarproject.bramble.api.rendezvous.event.RendezvousFailedEvent; import org.briarproject.bramble.api.transport.KeyManager; import org.briarproject.bramble.test.BrambleMockTestCase; import org.briarproject.bramble.test.DbExpectations; -import org.briarproject.bramble.test.PredicateMatcher; import org.jmock.Expectations; import org.junit.Before; import org.junit.Test; @@ -35,8 +29,6 @@ import java.util.Random; import static java.util.Collections.emptyList; import static java.util.Collections.singletonList; import static org.briarproject.bramble.api.contact.HandshakeLinkConstants.BASE32_LINK_BYTES; -import static org.briarproject.bramble.api.contact.PendingContactState.ADDING_CONTACT; -import static org.briarproject.bramble.api.contact.PendingContactState.FAILED; import static org.briarproject.bramble.api.contact.PendingContactState.WAITING_FOR_CONNECTION; import static org.briarproject.bramble.api.identity.AuthorConstants.MAX_AUTHOR_NAME_LENGTH; import static org.briarproject.bramble.api.identity.AuthorInfo.Status.OURSELVES; @@ -65,7 +57,6 @@ public class ContactManagerImplTest extends BrambleMockTestCase { context.mock(IdentityManager.class); private final PendingContactFactory pendingContactFactory = context.mock(PendingContactFactory.class); - private final EventBus eventBus = context.mock(EventBus.class); private final Author remote = getAuthor(); private final LocalAuthor localAuthor = getLocalAuthor(); @@ -85,7 +76,7 @@ public class ContactManagerImplTest extends BrambleMockTestCase { @Before public void setUp() { contactManager = new ContactManagerImpl(db, keyManager, - identityManager, pendingContactFactory, eventBus); + identityManager, pendingContactFactory); } @Test @@ -328,143 +319,4 @@ public class ContactManagerImplTest extends BrambleMockTestCase { assertEquals(WAITING_FOR_CONNECTION, pair.getSecond()); } - @Test - public void testPendingContactExpiresBeforeConnection() { - // The pending contact expires - the FAILED state is broadcast - context.checking(new Expectations() {{ - oneOf(eventBus).broadcast(with(new PredicateMatcher<>( - PendingContactStateChangedEvent.class, e -> - e.getPendingContactState() == FAILED))); - }}); - contactManager.eventOccurred(new RendezvousFailedEvent( - pendingContact.getId())); - context.assertIsSatisfied(); - - // A rendezvous connection is opened - no state is broadcast - contactManager.eventOccurred(new RendezvousConnectionOpenedEvent( - pendingContact.getId())); - context.assertIsSatisfied(); - - // The rendezvous connection fails - no state is broadcast - contactManager.eventOccurred(new RendezvousConnectionClosedEvent( - pendingContact.getId(), false)); - } - - @Test - public void testPendingContactExpiresDuringFailedConnection() { - // A rendezvous connection is opened - the ADDING_CONTACT state is - // broadcast - context.checking(new Expectations() {{ - oneOf(eventBus).broadcast(with(new PredicateMatcher<>( - PendingContactStateChangedEvent.class, e -> - e.getPendingContactState() == ADDING_CONTACT))); - }}); - - contactManager.eventOccurred(new RendezvousConnectionOpenedEvent( - pendingContact.getId())); - context.assertIsSatisfied(); - - // The pending contact expires - the FAILED state is broadcast - context.checking(new Expectations() {{ - oneOf(eventBus).broadcast(with(new PredicateMatcher<>( - PendingContactStateChangedEvent.class, e -> - e.getPendingContactState() == FAILED))); - }}); - - contactManager.eventOccurred(new RendezvousFailedEvent( - pendingContact.getId())); - context.assertIsSatisfied(); - - // The rendezvous connection fails - no state is broadcast - contactManager.eventOccurred(new RendezvousConnectionClosedEvent( - pendingContact.getId(), false)); - } - - @Test - public void testPendingContactExpiresDuringSuccessfulConnection() - throws Exception { - Transaction txn = new Transaction(null, false); - - // A rendezvous connection is opened - the ADDING_CONTACT state is - // broadcast - context.checking(new Expectations() {{ - oneOf(eventBus).broadcast(with(new PredicateMatcher<>( - PendingContactStateChangedEvent.class, e -> - e.getPendingContactState() == ADDING_CONTACT))); - }}); - - contactManager.eventOccurred(new RendezvousConnectionOpenedEvent( - pendingContact.getId())); - context.assertIsSatisfied(); - - // The pending contact expires - the FAILED state is broadcast - context.checking(new Expectations() {{ - oneOf(eventBus).broadcast(with(new PredicateMatcher<>( - PendingContactStateChangedEvent.class, e -> - e.getPendingContactState() == FAILED))); - }}); - - contactManager.eventOccurred(new RendezvousFailedEvent( - pendingContact.getId())); - context.assertIsSatisfied(); - - // The pending contact is converted to a contact - no state is broadcast - context.checking(new DbExpectations() {{ - oneOf(db).getPendingContact(txn, pendingContact.getId()); - will(returnValue(pendingContact)); - oneOf(db).removePendingContact(txn, pendingContact.getId()); - oneOf(db).addContact(txn, remote, local, - pendingContact.getPublicKey(), verified); - will(returnValue(contactId)); - oneOf(db).setContactAlias(txn, contactId, - pendingContact.getAlias()); - oneOf(identityManager).getHandshakeKeys(txn); - will(returnValue(handshakeKeyPair)); - oneOf(keyManager).addContact(txn, contactId, - pendingContact.getPublicKey(), handshakeKeyPair); - oneOf(keyManager).addRotationKeys(txn, contactId, rootKey, - timestamp, alice, active); - oneOf(db).getContact(txn, contactId); - will(returnValue(contact)); - }}); - - contactManager.addContact(txn, pendingContact.getId(), remote, - local, rootKey, timestamp, alice, verified, active); - context.assertIsSatisfied(); - - // The rendezvous connection succeeds - no state is broadcast - contactManager.eventOccurred(new RendezvousConnectionClosedEvent( - pendingContact.getId(), true)); - } - - @Test - public void testPendingContactRemovedDuringFailedConnection() - throws Exception { - Transaction txn = new Transaction(null, false); - - // A rendezvous connection is opened - the ADDING_CONTACT state is - // broadcast - context.checking(new Expectations() {{ - oneOf(eventBus).broadcast(with(new PredicateMatcher<>( - PendingContactStateChangedEvent.class, e -> - e.getPendingContactState() == ADDING_CONTACT))); - }}); - - contactManager.eventOccurred(new RendezvousConnectionOpenedEvent( - pendingContact.getId())); - context.assertIsSatisfied(); - - // The pending contact is removed - no state is broadcast - context.checking(new DbExpectations() {{ - oneOf(db).transaction(with(false), withDbRunnable(txn)); - oneOf(db).removePendingContact(txn, pendingContact.getId()); - }}); - - contactManager.removePendingContact(pendingContact.getId()); - context.assertIsSatisfied(); - - // The rendezvous connection fails - no state is broadcast - contactManager.eventOccurred(new RendezvousConnectionClosedEvent( - pendingContact.getId(), false)); - } } diff --git a/bramble-core/src/test/java/org/briarproject/bramble/rendezvous/RendezvousPollerImplTest.java b/bramble-core/src/test/java/org/briarproject/bramble/rendezvous/RendezvousPollerImplTest.java index e5b4a6463..902c9b8ea 100644 --- a/bramble-core/src/test/java/org/briarproject/bramble/rendezvous/RendezvousPollerImplTest.java +++ b/bramble-core/src/test/java/org/briarproject/bramble/rendezvous/RendezvousPollerImplTest.java @@ -1,8 +1,10 @@ package org.briarproject.bramble.rendezvous; import org.briarproject.bramble.api.contact.PendingContact; +import org.briarproject.bramble.api.contact.PendingContactState; import org.briarproject.bramble.api.contact.event.PendingContactAddedEvent; import org.briarproject.bramble.api.contact.event.PendingContactRemovedEvent; +import org.briarproject.bramble.api.contact.event.PendingContactStateChangedEvent; import org.briarproject.bramble.api.crypto.KeyPair; import org.briarproject.bramble.api.crypto.SecretKey; import org.briarproject.bramble.api.crypto.TransportCrypto; @@ -20,12 +22,14 @@ import org.briarproject.bramble.api.plugin.event.TransportEnabledEvent; import org.briarproject.bramble.api.properties.TransportProperties; import org.briarproject.bramble.api.rendezvous.KeyMaterialSource; import org.briarproject.bramble.api.rendezvous.RendezvousEndpoint; -import org.briarproject.bramble.api.rendezvous.event.RendezvousFailedEvent; +import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionClosedEvent; +import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionOpenedEvent; import org.briarproject.bramble.api.system.Clock; import org.briarproject.bramble.test.BrambleMockTestCase; import org.briarproject.bramble.test.CaptureArgumentAction; import org.briarproject.bramble.test.DbExpectations; import org.briarproject.bramble.test.ImmediateExecutor; +import org.briarproject.bramble.test.PredicateMatcher; import org.jmock.Expectations; import org.junit.Before; import org.junit.Test; @@ -38,6 +42,9 @@ import java.util.concurrent.atomic.AtomicReference; import static java.util.Collections.emptyList; import static java.util.Collections.singletonList; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.briarproject.bramble.api.contact.PendingContactState.ADDING_CONTACT; +import static org.briarproject.bramble.api.contact.PendingContactState.FAILED; +import static org.briarproject.bramble.api.contact.PendingContactState.WAITING_FOR_CONNECTION; import static org.briarproject.bramble.rendezvous.RendezvousConstants.POLLING_INTERVAL_MS; import static org.briarproject.bramble.rendezvous.RendezvousConstants.RENDEZVOUS_TIMEOUT_MS; import static org.briarproject.bramble.test.CollectionMatcher.collectionOf; @@ -110,7 +117,10 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { // The pending contact has not expired oneOf(clock).currentTimeMillis(); will(returnValue(beforeExpiry)); - // Capture the poll task, we'll run it later + oneOf(eventBus).broadcast(with(new PredicateMatcher<>( + PendingContactStateChangedEvent.class, e -> + e.getPendingContactState() == WAITING_FOR_CONNECTION))); + // Capture the poll task oneOf(scheduler).scheduleAtFixedRate(with(any(Runnable.class)), with(POLLING_INTERVAL_MS), with(POLLING_INTERVAL_MS), with(MILLISECONDS)); @@ -124,11 +134,7 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { context.assertIsSatisfied(); // Run the poll task - pending contact expires - context.checking(new Expectations() {{ - oneOf(clock).currentTimeMillis(); - will(returnValue(afterExpiry)); - oneOf(eventBus).broadcast(with(any(RendezvousFailedEvent.class))); - }}); + expectPendingContactExpires(afterExpiry); capturePollTask.get().run(); } @@ -147,7 +153,9 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { // The pending contact has already expired oneOf(clock).currentTimeMillis(); will(returnValue(atExpiry)); - oneOf(eventBus).broadcast(with(any(RendezvousFailedEvent.class))); + oneOf(eventBus).broadcast(with(new PredicateMatcher<>( + PendingContactStateChangedEvent.class, e -> + e.getPendingContactState() == FAILED))); // Schedule the poll task oneOf(scheduler).scheduleAtFixedRate(with(any(Runnable.class)), with(POLLING_INTERVAL_MS), with(POLLING_INTERVAL_MS), @@ -175,11 +183,7 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { context.assertIsSatisfied(); // Add the pending contact - endpoint should be created and polled - context.checking(new Expectations() {{ - oneOf(clock).currentTimeMillis(); - will(returnValue(beforeExpiry)); - }}); - + expectAddUnexpiredPendingContact(beforeExpiry); expectDeriveRendezvousKey(); expectCreateEndpoint(); @@ -230,11 +234,7 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { context.assertIsSatisfied(); // Add the pending contact - endpoint should be created and polled - context.checking(new Expectations() {{ - oneOf(clock).currentTimeMillis(); - will(returnValue(beforeExpiry)); - }}); - + expectAddUnexpiredPendingContact(beforeExpiry); expectDeriveRendezvousKey(); expectCreateEndpoint(); @@ -252,11 +252,10 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { context.assertIsSatisfied(); // Run the poll task - pending contact expires, endpoint is closed + expectPendingContactExpires(afterExpiry); + context.checking(new Expectations() {{ - oneOf(clock).currentTimeMillis(); - will(returnValue(afterExpiry)); oneOf(rendezvousEndpoint).close(); - oneOf(eventBus).broadcast(with(any(RendezvousFailedEvent.class))); }}); capturePollTask.get().run(); @@ -283,11 +282,7 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { context.assertIsSatisfied(); // Add the pending contact - no endpoints should be created yet - context.checking(new DbExpectations() {{ - oneOf(clock).currentTimeMillis(); - will(returnValue(beforeExpiry)); - }}); - + expectAddUnexpiredPendingContact(beforeExpiry); expectDeriveRendezvousKey(); rendezvousPoller.eventOccurred( @@ -314,15 +309,162 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { new PendingContactRemovedEvent(pendingContact.getId())); } + @Test + public void testRendezvousConnectionEvents() throws Exception { + long beforeExpiry = pendingContact.getTimestamp(); + + // Start the service + expectStartupWithPendingContact(beforeExpiry); + + rendezvousPoller.startService(); + context.assertIsSatisfied(); + + // Connection is opened - event should be broadcast + expectStateChangedEvent(ADDING_CONTACT); + + rendezvousPoller.eventOccurred( + new RendezvousConnectionOpenedEvent(pendingContact.getId())); + context.assertIsSatisfied(); + + // Connection fails - event should be broadcast + expectStateChangedEvent(WAITING_FOR_CONNECTION); + + rendezvousPoller.eventOccurred(new RendezvousConnectionClosedEvent( + pendingContact.getId(), false)); + } + + @Test + public void testPendingContactExpiresBeforeConnection() throws Exception { + long beforeExpiry = pendingContact.getTimestamp() + + RENDEZVOUS_TIMEOUT_MS - 1000; + long afterExpiry = beforeExpiry + POLLING_INTERVAL_MS; + + // Start the service, capturing the poll task + AtomicReference capturePollTask = + expectStartupWithPendingContact(beforeExpiry); + + rendezvousPoller.startService(); + context.assertIsSatisfied(); + + // Run the poll task - pending contact expires + expectPendingContactExpires(afterExpiry); + + capturePollTask.get().run(); + context.assertIsSatisfied(); + + // Connection is opened - no event should be broadcast + rendezvousPoller.eventOccurred( + new RendezvousConnectionOpenedEvent(pendingContact.getId())); + context.assertIsSatisfied(); + + // Connection fails - no event should be broadcast + rendezvousPoller.eventOccurred(new RendezvousConnectionClosedEvent( + pendingContact.getId(), false)); + } + + @Test + public void testPendingContactExpiresDuringFailedConnection() + throws Exception { + long beforeExpiry = pendingContact.getTimestamp() + + RENDEZVOUS_TIMEOUT_MS - 1000; + long afterExpiry = beforeExpiry + POLLING_INTERVAL_MS; + + // Start the service, capturing the poll task + AtomicReference capturePollTask = + expectStartupWithPendingContact(beforeExpiry); + + rendezvousPoller.startService(); + context.assertIsSatisfied(); + + // Connection is opened - event should be broadcast + expectStateChangedEvent(ADDING_CONTACT); + + rendezvousPoller.eventOccurred( + new RendezvousConnectionOpenedEvent(pendingContact.getId())); + context.assertIsSatisfied(); + + // Run the poll task - pending contact expires + expectPendingContactExpires(afterExpiry); + + capturePollTask.get().run(); + context.assertIsSatisfied(); + + // Connection fails - no event should be broadcast + rendezvousPoller.eventOccurred(new RendezvousConnectionClosedEvent( + pendingContact.getId(), false)); + } + + @Test + public void testPendingContactExpiresDuringSuccessfulConnection() + throws Exception { + long beforeExpiry = pendingContact.getTimestamp() + + RENDEZVOUS_TIMEOUT_MS - 1000; + long afterExpiry = beforeExpiry + POLLING_INTERVAL_MS; + + // Start the service, capturing the poll task + AtomicReference capturePollTask = + expectStartupWithPendingContact(beforeExpiry); + + rendezvousPoller.startService(); + context.assertIsSatisfied(); + + // Connection is opened - event should be broadcast + expectStateChangedEvent(ADDING_CONTACT); + + rendezvousPoller.eventOccurred( + new RendezvousConnectionOpenedEvent(pendingContact.getId())); + context.assertIsSatisfied(); + + // Run the poll task - pending contact expires + expectPendingContactExpires(afterExpiry); + + capturePollTask.get().run(); + context.assertIsSatisfied(); + + // Pending contact is removed - no event should be broadcast + rendezvousPoller.eventOccurred( + new PendingContactRemovedEvent(pendingContact.getId())); + } + + @Test + public void testPendingContactRemovedDuringFailedConnection() + throws Exception { + long beforeExpiry = pendingContact.getTimestamp(); + + // Start the service + expectStartupWithPendingContact(beforeExpiry); + + rendezvousPoller.startService(); + context.assertIsSatisfied(); + + // Connection is opened - event should be broadcast + expectStateChangedEvent(ADDING_CONTACT); + + rendezvousPoller.eventOccurred( + new RendezvousConnectionOpenedEvent(pendingContact.getId())); + context.assertIsSatisfied(); + + // Pending contact is removed - no event should be broadcast + rendezvousPoller.eventOccurred( + new PendingContactRemovedEvent(pendingContact.getId())); + context.assertIsSatisfied(); + + // Connection fails - no event should be broadcast + rendezvousPoller.eventOccurred(new RendezvousConnectionClosedEvent( + pendingContact.getId(), false)); + } + private AtomicReference expectStartupWithNoPendingContacts() throws Exception { Transaction txn = new Transaction(null, true); AtomicReference capturePollTask = new AtomicReference<>(); context.checking(new DbExpectations() {{ + // Load the pending contacts oneOf(db).transaction(with(true), withDbRunnable(txn)); oneOf(db).getPendingContacts(txn); will(returnValue(emptyList())); + // Capture the poll task oneOf(scheduler).scheduleAtFixedRate(with(any(Runnable.class)), with(POLLING_INTERVAL_MS), with(POLLING_INTERVAL_MS), with(MILLISECONDS)); @@ -333,6 +475,16 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { return capturePollTask; } + private void expectAddUnexpiredPendingContact(long now) { + context.checking(new Expectations() {{ + oneOf(clock).currentTimeMillis(); + will(returnValue(now)); + oneOf(eventBus).broadcast(with(new PredicateMatcher<>( + PendingContactStateChangedEvent.class, e -> + e.getPendingContactState() == WAITING_FOR_CONNECTION))); + }}); + } + private void expectDeriveRendezvousKey() throws Exception { Transaction txn = new Transaction(null, true); @@ -371,4 +523,50 @@ public class RendezvousPollerImplTest extends BrambleMockTestCase { will(returnValue(transportId)); }}); } + + private AtomicReference expectStartupWithPendingContact(long now) + throws Exception { + Transaction txn = new Transaction(null, true); + AtomicReference capturePollTask = new AtomicReference<>(); + + context.checking(new DbExpectations() {{ + // Load the pending contacts + oneOf(db).transaction(with(true), withDbRunnable(txn)); + oneOf(db).getPendingContacts(txn); + will(returnValue(singletonList(pendingContact))); + // The pending contact has not expired + oneOf(clock).currentTimeMillis(); + will(returnValue(now)); + oneOf(eventBus).broadcast(with(new PredicateMatcher<>( + PendingContactStateChangedEvent.class, e -> + e.getPendingContactState() == WAITING_FOR_CONNECTION))); + // Capture the poll task + oneOf(scheduler).scheduleAtFixedRate(with(any(Runnable.class)), + with(POLLING_INTERVAL_MS), with(POLLING_INTERVAL_MS), + with(MILLISECONDS)); + will(new CaptureArgumentAction<>(capturePollTask, Runnable.class, + 0)); + }}); + + expectDeriveRendezvousKey(); + + return capturePollTask; + } + + private void expectPendingContactExpires(long now) { + context.checking(new Expectations() {{ + oneOf(clock).currentTimeMillis(); + will(returnValue(now)); + }}); + + expectStateChangedEvent(FAILED); + } + + private void expectStateChangedEvent(PendingContactState state) { + context.checking(new Expectations() {{ + oneOf(eventBus).broadcast(with(new PredicateMatcher<>( + PendingContactStateChangedEvent.class, e -> + e.getPendingContactState() == state))); + }}); + } }