Add timeout monitor for Bluetooth connections.

bramble-core/src/main/java/org/briarproject/bramble/BrambleCoreModule.java

@@ -9,6 +9,7 @@ import org.briarproject.bramble.db.DatabaseExecutorModule;
 import org.briarproject.bramble.db.DatabaseModule;
 import org.briarproject.bramble.event.EventModule;
 import org.briarproject.bramble.identity.IdentityModule;
+import org.briarproject.bramble.io.IoModule;
 import org.briarproject.bramble.keyagreement.KeyAgreementModule;
 import org.briarproject.bramble.lifecycle.LifecycleModule;
 import org.briarproject.bramble.plugin.PluginModule;
@@ -35,6 +36,7 @@ import dagger.Module;
 		DatabaseExecutorModule.class,
 		EventModule.class,
 		IdentityModule.class,
+		IoModule.class,
 		KeyAgreementModule.class,
 		LifecycleModule.class,
 		PluginModule.class,

bramble-core/src/main/java/org/briarproject/bramble/io/IoModule.java

@@ -0,0 +1,18 @@
+package org.briarproject.bramble.io;
+
+import org.briarproject.bramble.api.io.TimeoutMonitor;
+
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+
+@Module
+public class IoModule {
+
+	@Provides
+	@Singleton
+	TimeoutMonitor provideTimeoutMonitor(TimeoutMonitorImpl timeoutMonitor) {
+		return timeoutMonitor;
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/io/TimeoutInputStream.java

@@ -0,0 +1,79 @@
+package org.briarproject.bramble.io;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.system.Clock;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import javax.annotation.concurrent.GuardedBy;
+
+@NotNullByDefault
+class TimeoutInputStream extends InputStream {
+
+	private final Clock clock;
+	private final InputStream in;
+	private final long timeoutNs;
+	private final CloseListener listener;
+	private final Object lock = new Object();
+	@GuardedBy("lock")
+	private long readStartedNs = -1;
+
+	TimeoutInputStream(Clock clock, InputStream in, long timeoutNs,
+			CloseListener listener) {
+		this.clock = clock;
+		this.in = in;
+		this.timeoutNs = timeoutNs;
+		this.listener = listener;
+	}
+
+	@Override
+	public int read() throws IOException {
+		synchronized (lock) {
+			readStartedNs = clock.nanoTime();
+		}
+		int input = in.read();
+		synchronized (lock) {
+			readStartedNs = -1;
+		}
+		return input;
+	}
+
+	@Override
+	public int read(byte[] b) throws IOException {
+		return read(b, 0, b.length);
+	}
+
+	@Override
+	public int read(byte[] b, int off, int len) throws IOException {
+		synchronized (lock) {
+			readStartedNs = clock.nanoTime();
+		}
+		int read = in.read(b, off, len);
+		synchronized (lock) {
+			readStartedNs = -1;
+		}
+		return read;
+	}
+
+	@Override
+	public void close() throws IOException {
+		try {
+			in.close();
+		} finally {
+			listener.onClose(this);
+		}
+	}
+
+	boolean hasTimedOut() {
+		synchronized (lock) {
+			return readStartedNs != -1 &&
+					clock.nanoTime() - readStartedNs > timeoutNs;
+		}
+	}
+
+	interface CloseListener {
+
+		void onClose(TimeoutInputStream closed);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/io/TimeoutMonitorImpl.java

@@ -0,0 +1,71 @@
+package org.briarproject.bramble.io;
+
+import org.briarproject.bramble.api.io.TimeoutMonitor;
+import org.briarproject.bramble.api.lifecycle.IoExecutor;
+import org.briarproject.bramble.api.system.Clock;
+import org.briarproject.bramble.api.system.Scheduler;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.logging.Logger;
+
+import javax.inject.Inject;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.logging.Level.INFO;
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.util.LogUtils.logException;
+
+class TimeoutMonitorImpl implements TimeoutMonitor {
+
+	private static final Logger LOG =
+			getLogger(TimeoutMonitorImpl.class.getName());
+
+	private static final long CHECK_INTERVAL_MS = SECONDS.toMillis(10);
+
+	private final Executor ioExecutor;
+	private final Clock clock;
+
+	private final List<TimeoutInputStream> streams =
+			new CopyOnWriteArrayList<>();
+
+	@Inject
+	TimeoutMonitorImpl(@IoExecutor Executor ioExecutor, Clock clock,
+			@Scheduler ScheduledExecutorService scheduler) {
+		this.ioExecutor = ioExecutor;
+		this.clock = clock;
+		scheduler.scheduleWithFixedDelay(this::checkTimeouts,
+				CHECK_INTERVAL_MS, CHECK_INTERVAL_MS, MILLISECONDS);
+	}
+
+	@Override
+	public InputStream createTimeoutInputStream(InputStream in,
+			long timeoutMs) {
+		TimeoutInputStream stream = new TimeoutInputStream(clock, in,
+				timeoutMs * 1_000_000, streams::remove);
+		streams.add(stream);
+		return stream;
+	}
+
+	@Scheduler
+	private void checkTimeouts() {
+		ioExecutor.execute(() -> {
+			LOG.info("Checking input stream timeouts");
+			for (TimeoutInputStream stream : streams) {
+				if (stream.hasTimedOut()) {
+					LOG.info("Input stream has timed out");
+					try {
+						stream.close();
+					} catch (IOException e) {
+						logException(LOG, INFO, e);
+					}
+				}
+			}
+		});
+	}
+}

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

@@ -5,6 +5,7 @@ import org.briarproject.bramble.api.Pair;
 import org.briarproject.bramble.api.data.BdfList;
 import org.briarproject.bramble.api.event.Event;
 import org.briarproject.bramble.api.event.EventListener;
+import org.briarproject.bramble.api.io.TimeoutMonitor;
 import org.briarproject.bramble.api.keyagreement.KeyAgreementConnection;
 import org.briarproject.bramble.api.keyagreement.KeyAgreementListener;
 import org.briarproject.bramble.api.keyagreement.event.KeyAgreementListeningEvent;
@@ -60,6 +61,7 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 			getLogger(BluetoothPlugin.class.getName());
 
 	final BluetoothConnectionLimiter connectionLimiter;
+	final TimeoutMonitor timeoutMonitor;
 
 	private final Executor ioExecutor;
 	private final SecureRandom secureRandom;
@@ -105,10 +107,11 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
 	abstract DuplexTransportConnection discoverAndConnect(String uuid);
 
 	BluetoothPlugin(BluetoothConnectionLimiter connectionLimiter,
-			Executor ioExecutor, SecureRandom secureRandom,
-			Backoff backoff, PluginCallback callback, int maxLatency,
-			int maxIdleTime) {
+			TimeoutMonitor timeoutMonitor, Executor ioExecutor,
+			SecureRandom secureRandom, Backoff backoff,
+			PluginCallback callback, int maxLatency, int maxIdleTime) {
 		this.connectionLimiter = connectionLimiter;
+		this.timeoutMonitor = timeoutMonitor;
 		this.ioExecutor = ioExecutor;
 		this.secureRandom = secureRandom;
 		this.backoff = backoff;

bramble-core/src/main/java/org/briarproject/bramble/system/SystemClock.java

@@ -12,6 +12,11 @@ public class SystemClock implements Clock {
 		return System.currentTimeMillis();
 	}
 
+	@Override
+	public long nanoTime() {
+		return System.nanoTime();
+	}
+
 	@Override
 	public void sleep(long milliseconds) throws InterruptedException {
 		Thread.sleep(milliseconds);

bramble-core/src/test/java/org/briarproject/bramble/db/JdbcDatabaseTest.java

@@ -2420,6 +2420,11 @@ public abstract class JdbcDatabaseTest extends BrambleTestCase {
 			return time;
 		}
 
+		@Override
+		public long nanoTime() {
+			return time * 1_000_000;
+		}
+
 		@Override
 		public void sleep(long milliseconds) throws InterruptedException {
 			Thread.sleep(milliseconds);