Commit 8bd4278a authored by akwizgran's avatar akwizgran

Add support for pending contacts to connection registry.

parent 015f5005
Pipeline #3397 passed with stage
in 7 minutes and 25 seconds
package org.briarproject.bramble.api.plugin; package org.briarproject.bramble.api.plugin;
import org.briarproject.bramble.api.contact.ContactId; import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.contact.PendingContactId;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.event.ConnectionClosedEvent;
import org.briarproject.bramble.api.plugin.event.ConnectionOpenedEvent;
import org.briarproject.bramble.api.plugin.event.ContactConnectedEvent;
import org.briarproject.bramble.api.plugin.event.ContactDisconnectedEvent;
import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionClosedEvent;
import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionOpenedEvent;
import java.util.Collection; import java.util.Collection;
...@@ -11,13 +18,50 @@ import java.util.Collection; ...@@ -11,13 +18,50 @@ import java.util.Collection;
@NotNullByDefault @NotNullByDefault
public interface ConnectionRegistry { public interface ConnectionRegistry {
/**
* Registers a connection with the given contact over the given transport.
* Broadcasts {@link ConnectionOpenedEvent}. Also broadcasts
* {@link ContactConnectedEvent} if this is the only connection with the
* contact.
*/
void registerConnection(ContactId c, TransportId t, boolean incoming); void registerConnection(ContactId c, TransportId t, boolean incoming);
/**
* Unregisters a connection with the given contact over the given transport.
* Broadcasts {@link ConnectionClosedEvent}. Also broadcasts
* {@link ContactDisconnectedEvent} if this is the only connection with
* the contact.
*/
void unregisterConnection(ContactId c, TransportId t, boolean incoming); void unregisterConnection(ContactId c, TransportId t, boolean incoming);
/**
* Returns any contacts that are connected via the given transport.
*/
Collection<ContactId> getConnectedContacts(TransportId t); Collection<ContactId> getConnectedContacts(TransportId t);
/**
* Returns true if the given contact is connected via the given transport.
*/
boolean isConnected(ContactId c, TransportId t); boolean isConnected(ContactId c, TransportId t);
/**
* Returns true if the given contact is connected via any transport.
*/
boolean isConnected(ContactId c); boolean isConnected(ContactId c);
/**
* Registers a connection with the given pending contact. Broadcasts
* {@link RendezvousConnectionOpenedEvent} if this is the only connection
* with the pending contact.
*
* @return True if this is the only connection with the pending contact,
* false if it is redundant and should be closed
*/
boolean registerConnection(PendingContactId p);
/**
* Unregisters a connection with the given pending contact. Broadcasts
* {@link RendezvousConnectionClosedEvent}.
*/
void unregisterConnection(PendingContactId p, boolean success);
} }
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 connection is closed.
*/
@Immutable
@NotNullByDefault
public class RendezvousConnectionClosedEvent extends Event {
private final PendingContactId pendingContactId;
private final boolean success;
public RendezvousConnectionClosedEvent(PendingContactId pendingContactId,
boolean success) {
this.pendingContactId = pendingContactId;
this.success = success;
}
public PendingContactId getPendingContactId() {
return pendingContactId;
}
public boolean isSuccess() {
return success;
}
}
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 connection is opened.
*/
@Immutable
@NotNullByDefault
public class RendezvousConnectionOpenedEvent extends Event {
private final PendingContactId pendingContactId;
public RendezvousConnectionOpenedEvent(PendingContactId pendingContactId) {
this.pendingContactId = pendingContactId;
}
public PendingContactId getPendingContactId() {
return pendingContactId;
}
}
...@@ -2,6 +2,7 @@ package org.briarproject.bramble.plugin; ...@@ -2,6 +2,7 @@ package org.briarproject.bramble.plugin;
import org.briarproject.bramble.api.Multiset; import org.briarproject.bramble.api.Multiset;
import org.briarproject.bramble.api.contact.ContactId; import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.contact.PendingContactId;
import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.event.EventBus;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.ConnectionRegistry; import org.briarproject.bramble.api.plugin.ConnectionRegistry;
...@@ -10,41 +11,49 @@ import org.briarproject.bramble.api.plugin.event.ConnectionClosedEvent; ...@@ -10,41 +11,49 @@ import org.briarproject.bramble.api.plugin.event.ConnectionClosedEvent;
import org.briarproject.bramble.api.plugin.event.ConnectionOpenedEvent; import org.briarproject.bramble.api.plugin.event.ConnectionOpenedEvent;
import org.briarproject.bramble.api.plugin.event.ContactConnectedEvent; import org.briarproject.bramble.api.plugin.event.ContactConnectedEvent;
import org.briarproject.bramble.api.plugin.event.ContactDisconnectedEvent; import org.briarproject.bramble.api.plugin.event.ContactDisconnectedEvent;
import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionClosedEvent;
import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionOpenedEvent;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.Collections; import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.locks.Lock; import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger; import java.util.logging.Logger;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe; import javax.annotation.concurrent.ThreadSafe;
import javax.inject.Inject; import javax.inject.Inject;
import static java.util.logging.Level.INFO; import static java.util.logging.Level.INFO;
import static java.util.logging.Logger.getLogger;
@ThreadSafe @ThreadSafe
@NotNullByDefault @NotNullByDefault
class ConnectionRegistryImpl implements ConnectionRegistry { class ConnectionRegistryImpl implements ConnectionRegistry {
private static final Logger LOG = private static final Logger LOG =
Logger.getLogger(ConnectionRegistryImpl.class.getName()); getLogger(ConnectionRegistryImpl.class.getName());
private final EventBus eventBus; private final EventBus eventBus;
private final Lock lock = new ReentrantLock();
// The following are locking: lock private final Object lock = new Object();
private final Map<TransportId, Multiset<ContactId>> connections; @GuardedBy("lock")
private final Map<TransportId, Multiset<ContactId>> contactConnections;
@GuardedBy("lock")
private final Multiset<ContactId> contactCounts; private final Multiset<ContactId> contactCounts;
@GuardedBy("lock")
private final Set<PendingContactId> connectedPendingContacts;
@Inject @Inject
ConnectionRegistryImpl(EventBus eventBus) { ConnectionRegistryImpl(EventBus eventBus) {
this.eventBus = eventBus; this.eventBus = eventBus;
connections = new HashMap<>(); contactConnections = new HashMap<>();
contactCounts = new Multiset<>(); contactCounts = new Multiset<>();
connectedPendingContacts = new HashSet<>();
} }
@Override @Override
...@@ -55,17 +64,14 @@ class ConnectionRegistryImpl implements ConnectionRegistry { ...@@ -55,17 +64,14 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
else LOG.info("Outgoing connection registered: " + t); else LOG.info("Outgoing connection registered: " + t);
} }
boolean firstConnection = false; boolean firstConnection = false;
lock.lock(); synchronized (lock) {
try { Multiset<ContactId> m = contactConnections.get(t);
Multiset<ContactId> m = connections.get(t);
if (m == null) { if (m == null) {
m = new Multiset<>(); m = new Multiset<>();
connections.put(t, m); contactConnections.put(t, m);
} }
m.add(c); m.add(c);
if (contactCounts.add(c) == 1) firstConnection = true; if (contactCounts.add(c) == 1) firstConnection = true;
} finally {
lock.unlock();
} }
eventBus.broadcast(new ConnectionOpenedEvent(c, t, incoming)); eventBus.broadcast(new ConnectionOpenedEvent(c, t, incoming));
if (firstConnection) { if (firstConnection) {
...@@ -82,14 +88,12 @@ class ConnectionRegistryImpl implements ConnectionRegistry { ...@@ -82,14 +88,12 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
else LOG.info("Outgoing connection unregistered: " + t); else LOG.info("Outgoing connection unregistered: " + t);
} }
boolean lastConnection = false; boolean lastConnection = false;
lock.lock(); synchronized (lock) {
try { Multiset<ContactId> m = contactConnections.get(t);
Multiset<ContactId> m = connections.get(t); if (m == null || !m.contains(c))
if (m == null) throw new IllegalArgumentException(); throw new IllegalArgumentException();
m.remove(c); m.remove(c);
if (contactCounts.remove(c) == 0) lastConnection = true; if (contactCounts.remove(c) == 0) lastConnection = true;
} finally {
lock.unlock();
} }
eventBus.broadcast(new ConnectionClosedEvent(c, t, incoming)); eventBus.broadcast(new ConnectionClosedEvent(c, t, incoming));
if (lastConnection) { if (lastConnection) {
...@@ -100,37 +104,47 @@ class ConnectionRegistryImpl implements ConnectionRegistry { ...@@ -100,37 +104,47 @@ class ConnectionRegistryImpl implements ConnectionRegistry {
@Override @Override
public Collection<ContactId> getConnectedContacts(TransportId t) { public Collection<ContactId> getConnectedContacts(TransportId t) {
lock.lock(); synchronized (lock) {
try { Multiset<ContactId> m = contactConnections.get(t);
Multiset<ContactId> m = connections.get(t);
if (m == null) return Collections.emptyList(); if (m == null) return Collections.emptyList();
List<ContactId> ids = new ArrayList<>(m.keySet()); List<ContactId> ids = new ArrayList<>(m.keySet());
if (LOG.isLoggable(INFO)) if (LOG.isLoggable(INFO))
LOG.info(ids.size() + " contacts connected: " + t); LOG.info(ids.size() + " contacts connected: " + t);
return ids; return ids;
} finally {
lock.unlock();
} }
} }
@Override @Override
public boolean isConnected(ContactId c, TransportId t) { public boolean isConnected(ContactId c, TransportId t) {
lock.lock(); synchronized (lock) {
try { Multiset<ContactId> m = contactConnections.get(t);
Multiset<ContactId> m = connections.get(t);
return m != null && m.contains(c); return m != null && m.contains(c);
} finally {
lock.unlock();
} }
} }
@Override @Override
public boolean isConnected(ContactId c) { public boolean isConnected(ContactId c) {
lock.lock(); synchronized (lock) {
try {
return contactCounts.contains(c); return contactCounts.contains(c);
} finally {
lock.unlock();
} }
} }
@Override
public boolean registerConnection(PendingContactId p) {
boolean added;
synchronized (lock) {
added = connectedPendingContacts.add(p);
}
if (added) eventBus.broadcast(new RendezvousConnectionOpenedEvent(p));
return added;
}
@Override
public void unregisterConnection(PendingContactId p, boolean success) {
synchronized (lock) {
if (!connectedPendingContacts.remove(p))
throw new IllegalArgumentException();
}
eventBus.broadcast(new RendezvousConnectionClosedEvent(p, success));
}
} }
package org.briarproject.bramble.plugin; package org.briarproject.bramble.plugin;
import org.briarproject.bramble.api.contact.ContactId; import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.contact.PendingContactId;
import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.event.EventBus;
import org.briarproject.bramble.api.plugin.ConnectionRegistry; import org.briarproject.bramble.api.plugin.ConnectionRegistry;
import org.briarproject.bramble.api.plugin.TransportId; import org.briarproject.bramble.api.plugin.TransportId;
...@@ -8,93 +9,106 @@ import org.briarproject.bramble.api.plugin.event.ConnectionClosedEvent; ...@@ -8,93 +9,106 @@ import org.briarproject.bramble.api.plugin.event.ConnectionClosedEvent;
import org.briarproject.bramble.api.plugin.event.ConnectionOpenedEvent; import org.briarproject.bramble.api.plugin.event.ConnectionOpenedEvent;
import org.briarproject.bramble.api.plugin.event.ContactConnectedEvent; import org.briarproject.bramble.api.plugin.event.ContactConnectedEvent;
import org.briarproject.bramble.api.plugin.event.ContactDisconnectedEvent; import org.briarproject.bramble.api.plugin.event.ContactDisconnectedEvent;
import org.briarproject.bramble.test.BrambleTestCase; import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionClosedEvent;
import org.briarproject.bramble.api.rendezvous.event.RendezvousConnectionOpenedEvent;
import org.briarproject.bramble.test.BrambleMockTestCase;
import org.jmock.Expectations; import org.jmock.Expectations;
import org.jmock.Mockery;
import org.junit.Test; import org.junit.Test;
import java.util.Collection; import java.util.Collection;
import java.util.Collections;
import java.util.NoSuchElementException;
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static org.briarproject.bramble.test.TestUtils.getContactId; import static org.briarproject.bramble.test.TestUtils.getContactId;
import static org.briarproject.bramble.test.TestUtils.getRandomId;
import static org.briarproject.bramble.test.TestUtils.getTransportId; import static org.briarproject.bramble.test.TestUtils.getTransportId;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail; import static org.junit.Assert.fail;
public class ConnectionRegistryImplTest extends BrambleTestCase { public class ConnectionRegistryImplTest extends BrambleMockTestCase {
private final ContactId contactId, contactId1; private final EventBus eventBus = context.mock(EventBus.class);
private final TransportId transportId, transportId1;
public ConnectionRegistryImplTest() { private final ContactId contactId = getContactId();
contactId = getContactId(); private final ContactId contactId1 = getContactId();
contactId1 = getContactId(); private final TransportId transportId = getTransportId();
transportId = getTransportId(); private final TransportId transportId1 = getTransportId();
transportId1 = getTransportId(); private final PendingContactId pendingContactId =
} new PendingContactId(getRandomId());
@Test @Test
public void testRegisterAndUnregister() { public void testRegisterAndUnregister() {
Mockery context = new Mockery();
EventBus eventBus = context.mock(EventBus.class);
context.checking(new Expectations() {{
exactly(5).of(eventBus).broadcast(with(any(
ConnectionOpenedEvent.class)));
exactly(2).of(eventBus).broadcast(with(any(
ConnectionClosedEvent.class)));
exactly(3).of(eventBus).broadcast(with(any(
ContactConnectedEvent.class)));
oneOf(eventBus).broadcast(with(any(
ContactDisconnectedEvent.class)));
}});
ConnectionRegistry c = new ConnectionRegistryImpl(eventBus); ConnectionRegistry c = new ConnectionRegistryImpl(eventBus);
// The registry should be empty // The registry should be empty
assertEquals(Collections.emptyList(), assertEquals(emptyList(), c.getConnectedContacts(transportId));
c.getConnectedContacts(transportId)); assertEquals(emptyList(), c.getConnectedContacts(transportId1));
assertEquals(Collections.emptyList(),
c.getConnectedContacts(transportId1));
// Check that a registered connection shows up - this should // Check that a registered connection shows up - this should
// broadcast a ConnectionOpenedEvent and a ContactConnectedEvent // broadcast a ConnectionOpenedEvent and a ContactConnectedEvent
context.checking(new Expectations() {{
oneOf(eventBus).broadcast(with(any(ConnectionOpenedEvent.class)));
oneOf(eventBus).broadcast(with(any(ContactConnectedEvent.class)));
}});
c.registerConnection(contactId, transportId, true); c.registerConnection(contactId, transportId, true);
assertEquals(Collections.singletonList(contactId), assertEquals(singletonList(contactId),
c.getConnectedContacts(transportId)); c.getConnectedContacts(transportId));
assertEquals(Collections.emptyList(), assertEquals(emptyList(), c.getConnectedContacts(transportId1));
c.getConnectedContacts(transportId1)); context.assertIsSatisfied();
// Register an identical connection - this should broadcast a // Register an identical connection - this should broadcast a
// ConnectionOpenedEvent and lookup should be unaffected // ConnectionOpenedEvent and lookup should be unaffected
context.checking(new Expectations() {{
oneOf(eventBus).broadcast(with(any(ConnectionOpenedEvent.class)));
}});
c.registerConnection(contactId, transportId, true); c.registerConnection(contactId, transportId, true);
assertEquals(Collections.singletonList(contactId), assertEquals(singletonList(contactId),
c.getConnectedContacts(transportId)); c.getConnectedContacts(transportId));
assertEquals(Collections.emptyList(), assertEquals(emptyList(), c.getConnectedContacts(transportId1));
c.getConnectedContacts(transportId1)); context.assertIsSatisfied();
// Unregister one of the connections - this should broadcast a // Unregister one of the connections - this should broadcast a
// ConnectionClosedEvent and lookup should be unaffected // ConnectionClosedEvent and lookup should be unaffected
context.checking(new Expectations() {{
oneOf(eventBus).broadcast(with(any(ConnectionClosedEvent.class)));
}});
c.unregisterConnection(contactId, transportId, true); c.unregisterConnection(contactId, transportId, true);
assertEquals(Collections.singletonList(contactId), assertEquals(singletonList(contactId),
c.getConnectedContacts(transportId)); c.getConnectedContacts(transportId));
assertEquals(Collections.emptyList(), assertEquals(emptyList(), c.getConnectedContacts(transportId1));
c.getConnectedContacts(transportId1)); context.assertIsSatisfied();
// Unregister the other connection - this should broadcast a // Unregister the other connection - this should broadcast a
// ConnectionClosedEvent and a ContactDisconnectedEvent // ConnectionClosedEvent and a ContactDisconnectedEvent
context.checking(new Expectations() {{
oneOf(eventBus).broadcast(with(any(ConnectionClosedEvent.class)));
oneOf(eventBus).broadcast(with(any(
ContactDisconnectedEvent.class)));
}});
c.unregisterConnection(contactId, transportId, true); c.unregisterConnection(contactId, transportId, true);
assertEquals(Collections.emptyList(), assertEquals(emptyList(), c.getConnectedContacts(transportId));
c.getConnectedContacts(transportId)); assertEquals(emptyList(), c.getConnectedContacts(transportId1));
assertEquals(Collections.emptyList(), context.assertIsSatisfied();
c.getConnectedContacts(transportId1));
// Try to unregister the connection again - exception should be thrown // Try to unregister the connection again - exception should be thrown
try { try {
c.unregisterConnection(contactId, transportId, true); c.unregisterConnection(contactId, transportId, true);
fail(); fail();
} catch (NoSuchElementException expected) { } catch (IllegalArgumentException expected) {
// Expected // Expected
} }
// Register both contacts with one transport, one contact with both - // Register both contacts with one transport, one contact with both -
// this should broadcast three ConnectionOpenedEvents and two // this should broadcast three ConnectionOpenedEvents and two
// ContactConnectedEvents // ContactConnectedEvents
context.checking(new Expectations() {{
exactly(3).of(eventBus).broadcast(with(any(
ConnectionOpenedEvent.class)));
exactly(2).of(eventBus).broadcast(with(any(
ContactConnectedEvent.class)));
}});
c.registerConnection(contactId, transportId, true); c.registerConnection(contactId, transportId, true);
c.registerConnection(contactId1, transportId, true); c.registerConnection(contactId1, transportId, true);
c.registerConnection(contactId1, transportId1, true); c.registerConnection(contactId1, transportId1, true);
...@@ -102,8 +116,34 @@ public class ConnectionRegistryImplTest extends BrambleTestCase { ...@@ -102,8 +116,34 @@ public class ConnectionRegistryImplTest extends BrambleTestCase {
assertEquals(2, connected.size()); assertEquals(2, connected.size());
assertTrue(connected.contains(contactId)); assertTrue(connected.contains(contactId));
assertTrue(connected.contains(contactId1)); assertTrue(connected.contains(contactId1));
assertEquals(Collections.singletonList(contactId1), assertEquals(singletonList(contactId1),
c.getConnectedContacts(transportId1)); c.getConnectedContacts(transportId1));
}
@Test
public void testRegisterAndUnregisterPendingContacts() {
ConnectionRegistry c = new ConnectionRegistryImpl(eventBus);
context.checking(new Expectations() {{
oneOf(eventBus).broadcast(with(any(
RendezvousConnectionOpenedEvent.class)));
}});
assertTrue(c.registerConnection(pendingContactId));
assertFalse(c.registerConnection(pendingContactId)); // Redundant
context.assertIsSatisfied(); context.assertIsSatisfied();
context.checking(new Expectations() {{
oneOf(eventBus).broadcast(with(any(
RendezvousConnectionClosedEvent.class)));
}});
c.unregisterConnection(pendingContactId, true);
context.assertIsSatisfied();
try {
c.unregisterConnection(pendingContactId, true);
fail();
} catch (IllegalArgumentException expected) {
// Expected
}
} }
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment