Move pending contact events to rendezvous poller.

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);
+}

bramble-api/src/main/java/org/briarproject/bramble/api/rendezvous/event/RendezvousFailedEvent.java

@@ -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;
-	}
-}

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<PendingContactId> pendingContacts;
+
+	public RendezvousPollEvent(TransportId transportId,
+			Collection<PendingContactId> pendingContacts) {
+		this.transportId = transportId;
+		this.pendingContacts = pendingContacts;
+	}
+
+	public TransportId getTransportId() {
+		return transportId;
+	}
+
+	public Collection<PendingContactId> getPendingContacts() {
+		return pendingContacts;
+	}
+}

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<ContactHook> hooks = new CopyOnWriteArrayList<>();
-	private final ConcurrentMap<PendingContactId, PendingContactState> states =
+	private final Map<PendingContactId, PendingContactState> states =
 			new ConcurrentHashMap<>();
 
 	@Inject
 	ContactManagerImpl(DatabaseComponent db,
 			KeyManager keyManager,
 			IdentityManager identityManager,
-			PendingContactFactory pendingContactFactory,
+			PendingContactFactory pendingContactFactory) {
-			EventBus eventBus) {
 		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) {
+		if (e instanceof PendingContactStateChangedEvent) {
-			RendezvousConnectionOpenedEvent r =
+			PendingContactStateChangedEvent p =
-					(RendezvousConnectionOpenedEvent) e;
+					(PendingContactStateChangedEvent) e;
-			setStateConnected(r.getPendingContactId());
+			states.put(p.getId(), p.getPendingContactState());
-		} 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));
 		}
 	}
 }

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;

bramble-core/src/main/java/org/briarproject/bramble/rendezvous/RendezvousPoller.java

@@ -1,7 +0,0 @@
-package org.briarproject.bramble.rendezvous;
-
-/**
- * Empty interface for injecting the rendezvous poller.
- */
-interface RendezvousPoller {
-}

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<TransportId, Long> 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) {
+		if (expiry > now) {
-			eventBus.broadcast(new RendezvousFailedEvent(p.getId()));
+			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<Pair<TransportProperties, ConnectionHandler>> properties =
 				new ArrayList<>();
 		for (Entry<PendingContactId, RendezvousEndpoint> 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;

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));
-	}
 }

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() {{
+		expectPendingContactExpires(afterExpiry);
-			oneOf(clock).currentTimeMillis();
-			will(returnValue(afterExpiry));
-			oneOf(eventBus).broadcast(with(any(RendezvousFailedEvent.class)));
-		}});
 
 		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() {{
+		expectAddUnexpiredPendingContact(beforeExpiry);
-			oneOf(clock).currentTimeMillis();
-			will(returnValue(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() {{
+		expectAddUnexpiredPendingContact(beforeExpiry);
-			oneOf(clock).currentTimeMillis();
-			will(returnValue(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() {{
+		expectAddUnexpiredPendingContact(beforeExpiry);
-			oneOf(clock).currentTimeMillis();
-			will(returnValue(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<Runnable> 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<Runnable> 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<Runnable> 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<Runnable> expectStartupWithNoPendingContacts()
 			throws Exception {
 		Transaction txn = new Transaction(null, true);
 		AtomicReference<Runnable> 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<Runnable> expectStartupWithPendingContact(long now)
+			throws Exception {
+		Transaction txn = new Transaction(null, true);
+		AtomicReference<Runnable> 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)));
+		}});
+	}
 }