Simple connection limiter that closes connections cleanly.

bramble-android/src/main/java/org/briarproject/bramble/plugin/bluetooth/AndroidBluetoothPluginFactory.java

@@ -67,7 +67,7 @@ public class AndroidBluetoothPluginFactory implements DuplexPluginFactory {
 	@Override
 	public DuplexPlugin createPlugin(PluginCallback callback) {
 		BluetoothConnectionLimiter connectionLimiter =
-				new BluetoothConnectionLimiterImpl();
+				new BluetoothConnectionLimiterImpl(eventBus);
 		Backoff backoff = backoffFactory.createBackoff(MIN_POLLING_INTERVAL,
 				MAX_POLLING_INTERVAL, BACKOFF_BASE);
 		AndroidBluetoothPlugin plugin = new AndroidBluetoothPlugin(

bramble-api/src/main/java/org/briarproject/bramble/api/sync/SyncSessionFactory.java

@@ -2,6 +2,7 @@ package org.briarproject.bramble.api.sync;
 
 import org.briarproject.bramble.api.contact.ContactId;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
 import org.briarproject.bramble.api.transport.StreamWriter;
 
 import java.io.InputStream;
@@ -14,10 +15,10 @@ public interface SyncSessionFactory {
 	SyncSession createIncomingSession(ContactId c, InputStream in,
 			PriorityHandler handler);
 
-	SyncSession createSimplexOutgoingSession(ContactId c, int maxLatency,
-			StreamWriter streamWriter);
+	SyncSession createSimplexOutgoingSession(ContactId c, TransportId t,
+			int maxLatency, StreamWriter streamWriter);
 
-	SyncSession createDuplexOutgoingSession(ContactId c, int maxLatency,
-			int maxIdleTime, StreamWriter streamWriter,
+	SyncSession createDuplexOutgoingSession(ContactId c, TransportId t,
+			int maxLatency, int maxIdleTime, StreamWriter streamWriter,
 			@Nullable Priority priority);
 }

bramble-api/src/main/java/org/briarproject/bramble/api/sync/event/CloseSyncConnectionsEvent.java

@@ -0,0 +1,26 @@
+package org.briarproject.bramble.api.sync.event;
+
+import org.briarproject.bramble.api.event.Event;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
+
+import javax.annotation.concurrent.Immutable;
+
+/**
+ * An event that is broadcast when all sync connections using a given
+ * transport should be closed.
+ */
+@Immutable
+@NotNullByDefault
+public class CloseSyncConnectionsEvent extends Event {
+
+	private final TransportId transportId;
+
+	public CloseSyncConnectionsEvent(TransportId transportId) {
+		this.transportId = transportId;
+	}
+
+	public TransportId getTransportId() {
+		return transportId;
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/connection/DuplexSyncConnection.java

@@ -47,20 +47,24 @@ abstract class DuplexSyncConnection extends SyncConnection
 
 	@Override
 	public void interruptOutgoingSession() {
+		SyncSession out = null;
 		synchronized (interruptLock) {
 			if (outgoingSession == null) interruptWaiting = true;
-			else outgoingSession.interrupt();
+			else out = outgoingSession;
 		}
+		if (out != null) out.interrupt();
 	}
 
 	void setOutgoingSession(SyncSession outgoingSession) {
+		boolean interruptWasWaiting = false;
 		synchronized (interruptLock) {
 			this.outgoingSession = outgoingSession;
 			if (interruptWaiting) {
-				outgoingSession.interrupt();
+				interruptWasWaiting = true;
 				interruptWaiting = false;
 			}
 		}
+		if (interruptWasWaiting) outgoingSession.interrupt();
 	}
 
 	DuplexSyncConnection(KeyManager keyManager,
@@ -99,6 +103,7 @@ abstract class DuplexSyncConnection extends SyncConnection
 				w.getOutputStream(), ctx);
 		ContactId c = requireNonNull(ctx.getContactId());
 		return syncSessionFactory.createDuplexOutgoingSession(c,
-				w.getMaxLatency(), w.getMaxIdleTime(), streamWriter, priority);
+				ctx.getTransportId(), w.getMaxLatency(), w.getMaxIdleTime(),
+				streamWriter, priority);
 	}
 }

bramble-core/src/main/java/org/briarproject/bramble/connection/OutgoingSimplexSyncConnection.java

@@ -72,7 +72,7 @@ class OutgoingSimplexSyncConnection extends SyncConnection implements Runnable {
 				w.getOutputStream(), ctx);
 		ContactId c = requireNonNull(ctx.getContactId());
 		return syncSessionFactory.createSimplexOutgoingSession(c,
-				w.getMaxLatency(), streamWriter);
+				ctx.getTransportId(), w.getMaxLatency(), streamWriter);
 	}
 }
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/bluetooth/BluetoothConnectionLimiter.java

@@ -23,17 +23,9 @@ interface BluetoothConnectionLimiter {
 	boolean canOpenContactConnection();
 
 	/**
-	 * Informs the limiter that a contact connection has been opened. The
-	 * limiter may close the new connection if key agreement is in progress.
-	 * <p/>
-	 * Returns false if the limiter has closed the new connection.
+	 * Informs the limiter that the given connection has been opened.
 	 */
-	boolean contactConnectionOpened(DuplexTransportConnection conn);
-
-	/**
-	 * Informs the limiter that a key agreement connection has been opened.
-	 */
-	void keyAgreementConnectionOpened(DuplexTransportConnection conn);
+	void connectionOpened(DuplexTransportConnection conn);
 
 	/**
 	 * Informs the limiter that the given connection has been closed.

bramble-core/src/main/java/org/briarproject/bramble/plugin/bluetooth/BluetoothConnectionLimiterImpl.java

@@ -1,46 +1,48 @@
 package org.briarproject.bramble.plugin.bluetooth;
 
+import org.briarproject.bramble.api.event.EventBus;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
 import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection;
+import org.briarproject.bramble.api.sync.event.CloseSyncConnectionsEvent;
 
-import java.io.IOException;
-import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.logging.Logger;
 
+import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
 
 import static java.util.logging.Level.INFO;
-import static java.util.logging.Level.WARNING;
-import static org.briarproject.bramble.util.LogUtils.logException;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.api.plugin.BluetoothConstants.ID;
 
 @NotNullByDefault
 @ThreadSafe
 class BluetoothConnectionLimiterImpl implements BluetoothConnectionLimiter {
 
 	private static final Logger LOG =
-			Logger.getLogger(BluetoothConnectionLimiterImpl.class.getName());
+			getLogger(BluetoothConnectionLimiterImpl.class.getName());
+
+	private final EventBus eventBus;
 
 	private final Object lock = new Object();
-	// The following are locking: lock
-	private final LinkedList<DuplexTransportConnection> connections =
+	@GuardedBy("lock")
+	private final List<DuplexTransportConnection> connections =
 			new LinkedList<>();
+	@GuardedBy("lock")
 	private boolean keyAgreementInProgress = false;
 
+	BluetoothConnectionLimiterImpl(EventBus eventBus) {
+		this.eventBus = eventBus;
+	}
+
 	@Override
 	public void keyAgreementStarted() {
-		List<DuplexTransportConnection> close;
 		synchronized (lock) {
 			keyAgreementInProgress = true;
-			close = new ArrayList<>(connections);
-			connections.clear();
 		}
-		if (LOG.isLoggable(INFO)) {
-			LOG.info("Key agreement started, closing " + close.size() +
-					" connections");
-		}
-		for (DuplexTransportConnection conn : close) tryToClose(conn);
+		LOG.info("Key agreement started");
+		eventBus.broadcast(new CloseSyncConnectionsEvent(ID));
 	}
 
 	@Override
@@ -65,35 +67,12 @@ class BluetoothConnectionLimiterImpl implements BluetoothConnectionLimiter {
 	}
 
 	@Override
-	public boolean contactConnectionOpened(DuplexTransportConnection conn) {
-		boolean accept = true;
+	public void connectionOpened(DuplexTransportConnection conn) {
 		synchronized (lock) {
-			if (keyAgreementInProgress) {
-				LOG.info("Refusing contact connection during key agreement");
-				accept = false;
-			} else {
-				LOG.info("Accepting contact connection");
-				connections.add(conn);
-			}
-		}
-		if (!accept) tryToClose(conn);
-		return accept;
-	}
-
-	@Override
-	public void keyAgreementConnectionOpened(DuplexTransportConnection conn) {
-		synchronized (lock) {
-			LOG.info("Accepting key agreement connection");
 			connections.add(conn);
-		}
-	}
-
-	private void tryToClose(DuplexTransportConnection conn) {
-		try {
-			conn.getWriter().dispose(false);
-			conn.getReader().dispose(false, false);
-		} catch (IOException e) {
-			logException(LOG, WARNING, e);
+			if (LOG.isLoggable(INFO)) {
+				LOG.info("Connection opened, " + connections.size() + " open");
+			}
 		}
 	}
 
@@ -101,8 +80,9 @@ class BluetoothConnectionLimiterImpl implements BluetoothConnectionLimiter {
 	public void connectionClosed(DuplexTransportConnection conn) {
 		synchronized (lock) {
 			connections.remove(conn);
-			if (LOG.isLoggable(INFO))
+			if (LOG.isLoggable(INFO)) {
 				LOG.info("Connection closed, " + connections.size() + " open");
+			}
 		}
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/bluetooth/BluetoothPlugin.java

@@ -232,10 +232,9 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 				return;
 			}
 			LOG.info("Connection received");
-			if (connectionLimiter.contactConnectionOpened(conn)) {
-				backoff.reset();
-				callback.handleConnection(conn);
-			}
+			connectionLimiter.connectionOpened(conn);
+			backoff.reset();
+			callback.handleConnection(conn);
 			if (!running) return;
 		}
 	}
@@ -327,8 +326,8 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 		String uuid = p.get(PROP_UUID);
 		if (isNullOrEmpty(uuid)) return null;
 		DuplexTransportConnection conn = connect(address, uuid);
-		if (conn == null) return null;
-		return connectionLimiter.contactConnectionOpened(conn) ? conn : null;
+		if (conn != null) connectionLimiter.connectionOpened(conn);
+		return conn;
 	}
 
 	@Override
@@ -384,7 +383,7 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 				LOG.info("Connecting to key agreement UUID " + uuid);
 			conn = connect(address, uuid);
 		}
-		if (conn != null) connectionLimiter.keyAgreementConnectionOpened(conn);
+		if (conn != null) connectionLimiter.connectionOpened(conn);
 		return conn;
 	}
 
@@ -453,7 +452,7 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 		public KeyAgreementConnection accept() throws IOException {
 			DuplexTransportConnection conn = acceptConnection(ss);
 			if (LOG.isLoggable(INFO)) LOG.info(ID + ": Incoming connection");
-			connectionLimiter.keyAgreementConnectionOpened(conn);
+			connectionLimiter.connectionOpened(conn);
 			return new KeyAgreementConnection(conn, ID);
 		}
 

bramble-core/src/main/java/org/briarproject/bramble/sync/DuplexOutgoingSession.java

@@ -11,6 +11,7 @@ import org.briarproject.bramble.api.event.EventListener;
 import org.briarproject.bramble.api.lifecycle.IoExecutor;
 import org.briarproject.bramble.api.lifecycle.event.LifecycleEvent;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
 import org.briarproject.bramble.api.sync.Ack;
 import org.briarproject.bramble.api.sync.Message;
 import org.briarproject.bramble.api.sync.Offer;
@@ -19,6 +20,7 @@ import org.briarproject.bramble.api.sync.Request;
 import org.briarproject.bramble.api.sync.SyncRecordWriter;
 import org.briarproject.bramble.api.sync.SyncSession;
 import org.briarproject.bramble.api.sync.Versions;
+import org.briarproject.bramble.api.sync.event.CloseSyncConnectionsEvent;
 import org.briarproject.bramble.api.sync.event.GroupVisibilityUpdatedEvent;
 import org.briarproject.bramble.api.sync.event.MessageRequestedEvent;
 import org.briarproject.bramble.api.sync.event.MessageSharedEvent;
@@ -73,6 +75,7 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
 	private final EventBus eventBus;
 	private final Clock clock;
 	private final ContactId contactId;
+	private final TransportId transportId;
 	private final int maxLatency, maxIdleTime;
 	private final StreamWriter streamWriter;
 	private final SyncRecordWriter recordWriter;
@@ -90,14 +93,16 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
 	private volatile boolean interrupted = false;
 
 	DuplexOutgoingSession(DatabaseComponent db, Executor dbExecutor,
-			EventBus eventBus, Clock clock, ContactId contactId, int maxLatency,
-			int maxIdleTime, StreamWriter streamWriter,
-			SyncRecordWriter recordWriter, @Nullable Priority priority) {
+			EventBus eventBus, Clock clock, ContactId contactId,
+			TransportId transportId, int maxLatency, int maxIdleTime,
+			StreamWriter streamWriter, SyncRecordWriter recordWriter,
+			@Nullable Priority priority) {
 		this.db = db;
 		this.dbExecutor = dbExecutor;
 		this.eventBus = eventBus;
 		this.clock = clock;
 		this.contactId = contactId;
+		this.transportId = transportId;
 		this.maxLatency = maxLatency;
 		this.maxIdleTime = maxIdleTime;
 		this.streamWriter = streamWriter;
@@ -230,6 +235,9 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
 		} else if (e instanceof LifecycleEvent) {
 			LifecycleEvent l = (LifecycleEvent) e;
 			if (l.getLifecycleState() == STOPPING) interrupt();
+		} else if (e instanceof CloseSyncConnectionsEvent) {
+			CloseSyncConnectionsEvent c = (CloseSyncConnectionsEvent) e;
+			if (c.getTransportId().equals(transportId)) interrupt();
 		}
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/sync/SimplexOutgoingSession.java

@@ -11,11 +11,13 @@ import org.briarproject.bramble.api.event.EventListener;
 import org.briarproject.bramble.api.lifecycle.IoExecutor;
 import org.briarproject.bramble.api.lifecycle.event.LifecycleEvent;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
 import org.briarproject.bramble.api.sync.Ack;
 import org.briarproject.bramble.api.sync.Message;
 import org.briarproject.bramble.api.sync.SyncRecordWriter;
 import org.briarproject.bramble.api.sync.SyncSession;
 import org.briarproject.bramble.api.sync.Versions;
+import org.briarproject.bramble.api.sync.event.CloseSyncConnectionsEvent;
 import org.briarproject.bramble.api.transport.StreamWriter;
 
 import java.io.IOException;
@@ -56,6 +58,7 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 	private final Executor dbExecutor;
 	private final EventBus eventBus;
 	private final ContactId contactId;
+	private final TransportId transportId;
 	private final int maxLatency;
 	private final StreamWriter streamWriter;
 	private final SyncRecordWriter recordWriter;
@@ -65,12 +68,14 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 	private volatile boolean interrupted = false;
 
 	SimplexOutgoingSession(DatabaseComponent db, Executor dbExecutor,
-			EventBus eventBus, ContactId contactId, int maxLatency,
-			StreamWriter streamWriter, SyncRecordWriter recordWriter) {
+			EventBus eventBus, ContactId contactId, TransportId transportId,
+			int maxLatency, StreamWriter streamWriter,
+			SyncRecordWriter recordWriter) {
 		this.db = db;
 		this.dbExecutor = dbExecutor;
 		this.eventBus = eventBus;
 		this.contactId = contactId;
+		this.transportId = transportId;
 		this.maxLatency = maxLatency;
 		this.streamWriter = streamWriter;
 		this.recordWriter = recordWriter;
@@ -123,6 +128,9 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
 		} else if (e instanceof LifecycleEvent) {
 			LifecycleEvent l = (LifecycleEvent) e;
 			if (l.getLifecycleState() == STOPPING) interrupt();
+		} else if (e instanceof CloseSyncConnectionsEvent) {
+			CloseSyncConnectionsEvent c = (CloseSyncConnectionsEvent) e;
+			if (c.getTransportId().equals(transportId)) interrupt();
 		}
 	}
 

bramble-core/src/main/java/org/briarproject/bramble/sync/SyncSessionFactoryImpl.java

@@ -5,6 +5,7 @@ import org.briarproject.bramble.api.db.DatabaseComponent;
 import org.briarproject.bramble.api.db.DatabaseExecutor;
 import org.briarproject.bramble.api.event.EventBus;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.TransportId;
 import org.briarproject.bramble.api.sync.Priority;
 import org.briarproject.bramble.api.sync.PriorityHandler;
 import org.briarproject.bramble.api.sync.SyncRecordReader;
@@ -58,23 +59,23 @@ class SyncSessionFactoryImpl implements SyncSessionFactory {
 	}
 
 	@Override
-	public SyncSession createSimplexOutgoingSession(ContactId c,
+	public SyncSession createSimplexOutgoingSession(ContactId c, TransportId t,
 			int maxLatency, StreamWriter streamWriter) {
 		OutputStream out = streamWriter.getOutputStream();
 		SyncRecordWriter recordWriter =
 				recordWriterFactory.createRecordWriter(out);
-		return new SimplexOutgoingSession(db, dbExecutor, eventBus, c,
+		return new SimplexOutgoingSession(db, dbExecutor, eventBus, c, t,
 				maxLatency, streamWriter, recordWriter);
 	}
 
 	@Override
-	public SyncSession createDuplexOutgoingSession(ContactId c, int maxLatency,
-			int maxIdleTime, StreamWriter streamWriter,
+	public SyncSession createDuplexOutgoingSession(ContactId c, TransportId t,
+			int maxLatency, int maxIdleTime, StreamWriter streamWriter,
 			@Nullable Priority priority) {
 		OutputStream out = streamWriter.getOutputStream();
 		SyncRecordWriter recordWriter =
 				recordWriterFactory.createRecordWriter(out);
-		return new DuplexOutgoingSession(db, dbExecutor, eventBus, clock, c,
+		return new DuplexOutgoingSession(db, dbExecutor, eventBus, clock, c, t,
 				maxLatency, maxIdleTime, streamWriter, recordWriter, priority);
 	}
 }

bramble-core/src/test/java/org/briarproject/bramble/sync/SimplexOutgoingSessionTest.java

@@ -4,6 +4,7 @@ import org.briarproject.bramble.api.contact.ContactId;
 import org.briarproject.bramble.api.db.DatabaseComponent;
 import org.briarproject.bramble.api.db.Transaction;
 import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.plugin.TransportId;
 import org.briarproject.bramble.api.sync.Ack;
 import org.briarproject.bramble.api.sync.GroupId;
 import org.briarproject.bramble.api.sync.Message;
@@ -23,6 +24,7 @@ import static org.briarproject.bramble.api.sync.SyncConstants.MAX_MESSAGE_IDS;
 import static org.briarproject.bramble.test.TestUtils.getContactId;
 import static org.briarproject.bramble.test.TestUtils.getMessage;
 import static org.briarproject.bramble.test.TestUtils.getRandomId;
+import static org.briarproject.bramble.test.TestUtils.getTransportId;
 
 public class SimplexOutgoingSessionTest extends BrambleMockTestCase {
 
@@ -36,14 +38,15 @@ public class SimplexOutgoingSessionTest extends BrambleMockTestCase {
 
 	private final Executor dbExecutor = new ImmediateExecutor();
 	private final ContactId contactId = getContactId();
+	private final TransportId transportId = getTransportId();
 	private final Message message = getMessage(new GroupId(getRandomId()));
 	private final MessageId messageId = message.getId();
 
 	@Test
 	public void testNothingToSend() throws Exception {
 		SimplexOutgoingSession session = new SimplexOutgoingSession(db,
-				dbExecutor, eventBus, contactId, MAX_LATENCY, streamWriter,
-				recordWriter);
+				dbExecutor, eventBus, contactId, transportId, MAX_LATENCY,
+				streamWriter, recordWriter);
 		Transaction noAckTxn = new Transaction(null, false);
 		Transaction noMsgTxn = new Transaction(null, false);
 
@@ -76,8 +79,8 @@ public class SimplexOutgoingSessionTest extends BrambleMockTestCase {
 	public void testSomethingToSend() throws Exception {
 		Ack ack = new Ack(singletonList(messageId));
 		SimplexOutgoingSession session = new SimplexOutgoingSession(db,
-				dbExecutor, eventBus, contactId, MAX_LATENCY, streamWriter,
-				recordWriter);
+				dbExecutor, eventBus, contactId, transportId, MAX_LATENCY,
+				streamWriter, recordWriter);
 		Transaction ackTxn = new Transaction(null, false);
 		Transaction noAckTxn = new Transaction(null, false);
 		Transaction msgTxn = new Transaction(null, false);

bramble-java/src/main/java/org/briarproject/bramble/plugin/bluetooth/JavaBluetoothPluginFactory.java

@@ -56,7 +56,7 @@ public class JavaBluetoothPluginFactory implements DuplexPluginFactory {
 	@Override
 	public DuplexPlugin createPlugin(PluginCallback callback) {
 		BluetoothConnectionLimiter connectionLimiter =
-				new BluetoothConnectionLimiterImpl();
+				new BluetoothConnectionLimiterImpl(eventBus);
 		Backoff backoff = backoffFactory.createBackoff(MIN_POLLING_INTERVAL,
 				MAX_POLLING_INTERVAL, BACKOFF_BASE);
 		JavaBluetoothPlugin plugin = new JavaBluetoothPlugin(connectionLimiter,