Compare commits

...

23 Commits

Author SHA1 Message Date
akwizgran
173b6006c4 Don't treat an incoming connection as an attempt to raise the limit. 2020-05-11 17:15:25 +01:00
akwizgran
99edb893f7 Check for stability whenever connections are closed. 2020-05-11 17:15:25 +01:00
akwizgran
f063feedd4 Simplify backoff. 2020-05-11 17:15:25 +01:00
akwizgran
126f515760 Move responsibility for closing connections from limiter to plugin. 2020-05-11 17:15:25 +01:00
akwizgran
e2b61483d6 Always accept incoming connections. 2020-05-11 17:15:25 +01:00
akwizgran
9771825c45 Back off between attempts to raise connection limit. 2020-05-11 17:15:24 +01:00
akwizgran
e376744487 Update constructor args. 2020-05-11 17:15:24 +01:00
akwizgran
13cca9ca61 Occasionally try to raise the limit by allowing an extra connection. 2020-05-11 17:15:24 +01:00
akwizgran
e464f9e7bd Close connections cleanly when starting key agreement. 2020-05-11 17:15:24 +01:00
akwizgran
bd86ff2d5f Let the limiter know whether connections closed cleanly. 2020-05-11 17:15:24 +01:00
akwizgran
bda3b2100a Raise the connection limit if connections are stable. 2020-05-11 17:15:24 +01:00
akwizgran
104a82aea9 Add unit test for connection limiter. 2020-05-11 17:15:24 +01:00
akwizgran
d905451f48 Impose a fixed limit on the number of Bluetooth connections. 2020-05-11 17:15:24 +01:00
Torsten Grote
708452713d Merge branch '1712-detect-dead-bluetooth-connections' into 'master'
Detect and close dead Bluetooth connections

See merge request briar/briar!1246
2020-05-11 15:55:07 +00:00
akwizgran
c80d3196af Use milliseconds for timing. 2020-05-11 15:42:23 +01:00
Torsten Grote
d1c2eb89a1 Merge branch '1712-fix-double-connection-counting' into 'master'
Don't count Bluetooth connections twice

See merge request briar/briar!1245
2020-05-11 14:06:30 +00:00
akwizgran
c4273d22ed Delegate all other methods to wrapped InputStream. 2020-05-08 16:22:46 +01:00
akwizgran
21f3a9f3c7 Add javadoc. 2020-05-08 16:22:46 +01:00
akwizgran
0281eec0da Add unit test for TimeoutInputStream. 2020-05-08 16:22:46 +01:00
akwizgran
d3fd309609 Only check timeouts when we have some streams to monitor. 2020-05-08 16:22:46 +01:00
akwizgran
f2f278c393 Add timeout monitor for Bluetooth connections. 2020-05-08 16:22:46 +01:00
akwizgran
e204d5a996 Don't count connections twice. 2020-05-08 15:17:27 +01:00
akwizgran
876efee1a8 Use keepalives to detect dead connections. 2020-05-08 14:21:41 +01:00
25 changed files with 891 additions and 129 deletions

View File

@@ -9,6 +9,7 @@ import android.content.Context;
import android.content.Intent; import android.content.Intent;
import android.content.IntentFilter; import android.content.IntentFilter;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.nullsafety.MethodsNotNullByDefault; import org.briarproject.bramble.api.nullsafety.MethodsNotNullByDefault;
import org.briarproject.bramble.api.nullsafety.ParametersNotNullByDefault; import org.briarproject.bramble.api.nullsafety.ParametersNotNullByDefault;
import org.briarproject.bramble.api.plugin.Backoff; import org.briarproject.bramble.api.plugin.Backoff;
@@ -76,11 +77,12 @@ class AndroidBluetoothPlugin extends BluetoothPlugin<BluetoothServerSocket> {
private volatile BluetoothAdapter adapter = null; private volatile BluetoothAdapter adapter = null;
AndroidBluetoothPlugin(BluetoothConnectionLimiter connectionLimiter, AndroidBluetoothPlugin(BluetoothConnectionLimiter connectionLimiter,
Executor ioExecutor, AndroidExecutor androidExecutor, TimeoutMonitor timeoutMonitor, Executor ioExecutor,
Context appContext, SecureRandom secureRandom, Clock clock, SecureRandom secureRandom, AndroidExecutor androidExecutor,
Backoff backoff, PluginCallback callback, int maxLatency) { Context appContext, Clock clock, Backoff backoff,
super(connectionLimiter, ioExecutor, secureRandom, backoff, callback, PluginCallback callback, int maxLatency, int maxIdleTime) {
maxLatency); super(connectionLimiter, timeoutMonitor, ioExecutor, secureRandom,
backoff, callback, maxLatency, maxIdleTime);
this.androidExecutor = androidExecutor; this.androidExecutor = androidExecutor;
this.appContext = appContext; this.appContext = appContext;
this.clock = clock; this.clock = clock;
@@ -172,9 +174,10 @@ class AndroidBluetoothPlugin extends BluetoothPlugin<BluetoothServerSocket> {
return wrapSocket(ss.accept()); return wrapSocket(ss.accept());
} }
private DuplexTransportConnection wrapSocket(BluetoothSocket s) { private DuplexTransportConnection wrapSocket(BluetoothSocket s)
return new AndroidBluetoothTransportConnection(this, throws IOException {
connectionLimiter, s); return new AndroidBluetoothTransportConnection(this, connectionLimiter,
timeoutMonitor, s);
} }
@Override @Override

View File

@@ -3,6 +3,7 @@ package org.briarproject.bramble.plugin.bluetooth;
import android.content.Context; import android.content.Context;
import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.event.EventBus;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.Backoff; import org.briarproject.bramble.api.plugin.Backoff;
import org.briarproject.bramble.api.plugin.BackoffFactory; import org.briarproject.bramble.api.plugin.BackoffFactory;
@@ -25,6 +26,7 @@ import static org.briarproject.bramble.api.plugin.BluetoothConstants.ID;
public class AndroidBluetoothPluginFactory implements DuplexPluginFactory { public class AndroidBluetoothPluginFactory implements DuplexPluginFactory {
private static final int MAX_LATENCY = 30 * 1000; // 30 seconds private static final int MAX_LATENCY = 30 * 1000; // 30 seconds
private static final int MAX_IDLE_TIME = 30 * 1000; // 30 seconds
private static final int MIN_POLLING_INTERVAL = 60 * 1000; // 1 minute private static final int MIN_POLLING_INTERVAL = 60 * 1000; // 1 minute
private static final int MAX_POLLING_INTERVAL = 10 * 60 * 1000; // 10 mins private static final int MAX_POLLING_INTERVAL = 10 * 60 * 1000; // 10 mins
private static final double BACKOFF_BASE = 1.2; private static final double BACKOFF_BASE = 1.2;
@@ -35,18 +37,20 @@ public class AndroidBluetoothPluginFactory implements DuplexPluginFactory {
private final SecureRandom secureRandom; private final SecureRandom secureRandom;
private final EventBus eventBus; private final EventBus eventBus;
private final Clock clock; private final Clock clock;
private final TimeoutMonitor timeoutMonitor;
private final BackoffFactory backoffFactory; private final BackoffFactory backoffFactory;
public AndroidBluetoothPluginFactory(Executor ioExecutor, public AndroidBluetoothPluginFactory(Executor ioExecutor,
AndroidExecutor androidExecutor, Context appContext, AndroidExecutor androidExecutor, Context appContext,
SecureRandom secureRandom, EventBus eventBus, Clock clock, SecureRandom secureRandom, EventBus eventBus, Clock clock,
BackoffFactory backoffFactory) { TimeoutMonitor timeoutMonitor, BackoffFactory backoffFactory) {
this.ioExecutor = ioExecutor; this.ioExecutor = ioExecutor;
this.androidExecutor = androidExecutor; this.androidExecutor = androidExecutor;
this.appContext = appContext; this.appContext = appContext;
this.secureRandom = secureRandom; this.secureRandom = secureRandom;
this.eventBus = eventBus; this.eventBus = eventBus;
this.clock = clock; this.clock = clock;
this.timeoutMonitor = timeoutMonitor;
this.backoffFactory = backoffFactory; this.backoffFactory = backoffFactory;
} }
@@ -63,12 +67,13 @@ public class AndroidBluetoothPluginFactory implements DuplexPluginFactory {
@Override @Override
public DuplexPlugin createPlugin(PluginCallback callback) { public DuplexPlugin createPlugin(PluginCallback callback) {
BluetoothConnectionLimiter connectionLimiter = BluetoothConnectionLimiter connectionLimiter =
new BluetoothConnectionLimiterImpl(); new BluetoothConnectionLimiterImpl(eventBus, clock);
Backoff backoff = backoffFactory.createBackoff(MIN_POLLING_INTERVAL, Backoff backoff = backoffFactory.createBackoff(MIN_POLLING_INTERVAL,
MAX_POLLING_INTERVAL, BACKOFF_BASE); MAX_POLLING_INTERVAL, BACKOFF_BASE);
AndroidBluetoothPlugin plugin = new AndroidBluetoothPlugin( AndroidBluetoothPlugin plugin = new AndroidBluetoothPlugin(
connectionLimiter, ioExecutor, androidExecutor, appContext, connectionLimiter, timeoutMonitor, ioExecutor, secureRandom,
secureRandom, clock, backoff, callback, MAX_LATENCY); androidExecutor, appContext, clock, backoff,
callback, MAX_LATENCY, MAX_IDLE_TIME);
eventBus.addListener(plugin); eventBus.addListener(plugin);
return plugin; return plugin;
} }

View File

@@ -2,6 +2,7 @@ package org.briarproject.bramble.plugin.bluetooth;
import android.bluetooth.BluetoothSocket; import android.bluetooth.BluetoothSocket;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.Plugin; import org.briarproject.bramble.api.plugin.Plugin;
import org.briarproject.bramble.api.plugin.duplex.AbstractDuplexTransportConnection; import org.briarproject.bramble.api.plugin.duplex.AbstractDuplexTransportConnection;
@@ -17,22 +18,26 @@ import static org.briarproject.bramble.util.AndroidUtils.isValidBluetoothAddress
class AndroidBluetoothTransportConnection class AndroidBluetoothTransportConnection
extends AbstractDuplexTransportConnection { extends AbstractDuplexTransportConnection {
private final BluetoothConnectionLimiter connectionManager; private final BluetoothConnectionLimiter connectionLimiter;
private final BluetoothSocket socket; private final BluetoothSocket socket;
private final InputStream in;
AndroidBluetoothTransportConnection(Plugin plugin, AndroidBluetoothTransportConnection(Plugin plugin,
BluetoothConnectionLimiter connectionManager, BluetoothConnectionLimiter connectionLimiter,
BluetoothSocket socket) { TimeoutMonitor timeoutMonitor, BluetoothSocket socket)
throws IOException {
super(plugin); super(plugin);
this.connectionManager = connectionManager; this.connectionLimiter = connectionLimiter;
this.socket = socket; this.socket = socket;
in = timeoutMonitor.createTimeoutInputStream(
socket.getInputStream(), plugin.getMaxIdleTime() * 2);
String address = socket.getRemoteDevice().getAddress(); String address = socket.getRemoteDevice().getAddress();
if (isValidBluetoothAddress(address)) remote.put(PROP_ADDRESS, address); if (isValidBluetoothAddress(address)) remote.put(PROP_ADDRESS, address);
} }
@Override @Override
protected InputStream getInputStream() throws IOException { protected InputStream getInputStream() {
return socket.getInputStream(); return in;
} }
@Override @Override
@@ -45,7 +50,7 @@ class AndroidBluetoothTransportConnection
try { try {
socket.close(); socket.close();
} finally { } finally {
connectionManager.connectionClosed(this); connectionLimiter.connectionClosed(this, exception);
} }
} }
} }

View File

@@ -0,0 +1,15 @@
package org.briarproject.bramble.api.io;
import java.io.InputStream;
public interface TimeoutMonitor {
/**
* Returns an {@link InputStream} that wraps the given stream and allows
* read timeouts to be detected.
*
* @param timeoutMs The read timeout in milliseconds. Timeouts will be
* detected eventually but are not guaranteed to be detected immediately.
*/
InputStream createTimeoutInputStream(InputStream in, long timeoutMs);
}

View File

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

View File

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

View File

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

View File

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

View File

@@ -0,0 +1,104 @@
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 timeoutMs;
private final CloseListener listener;
private final Object lock = new Object();
@GuardedBy("lock")
private long readStartedMs = -1;
TimeoutInputStream(Clock clock, InputStream in, long timeoutMs,
CloseListener listener) {
this.clock = clock;
this.in = in;
this.timeoutMs = timeoutMs;
this.listener = listener;
}
@Override
public int read() throws IOException {
synchronized (lock) {
readStartedMs = clock.currentTimeMillis();
}
int input = in.read();
synchronized (lock) {
readStartedMs = -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) {
readStartedMs = clock.currentTimeMillis();
}
int read = in.read(b, off, len);
synchronized (lock) {
readStartedMs = -1;
}
return read;
}
@Override
public void close() throws IOException {
try {
in.close();
} finally {
listener.onClose(this);
}
}
@Override
public int available() throws IOException {
return in.available();
}
@Override
public void mark(int readlimit) {
in.mark(readlimit);
}
@Override
public boolean markSupported() {
return in.markSupported();
}
@Override
public void reset() throws IOException {
in.reset();
}
@Override
public long skip(long n) throws IOException {
return in.skip(n);
}
boolean hasTimedOut() {
synchronized (lock) {
return readStartedMs != -1 &&
clock.currentTimeMillis() - readStartedMs > timeoutMs;
}
}
interface CloseListener {
void onClose(TimeoutInputStream closed);
}
}

View File

@@ -0,0 +1,96 @@
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.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.logging.Logger;
import javax.annotation.concurrent.GuardedBy;
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 ScheduledExecutorService scheduler;
private final Executor ioExecutor;
private final Clock clock;
private final Object lock = new Object();
@GuardedBy("lock")
private final List<TimeoutInputStream> streams = new ArrayList<>();
@GuardedBy("lock")
private Future<?> task = null;
@Inject
TimeoutMonitorImpl(@Scheduler ScheduledExecutorService scheduler,
@IoExecutor Executor ioExecutor, Clock clock) {
this.scheduler = scheduler;
this.ioExecutor = ioExecutor;
this.clock = clock;
}
@Override
public InputStream createTimeoutInputStream(InputStream in,
long timeoutMs) {
TimeoutInputStream stream = new TimeoutInputStream(clock, in,
timeoutMs, this::removeStream);
synchronized (lock) {
if (streams.isEmpty()) {
task = scheduler.scheduleWithFixedDelay(this::checkTimeouts,
CHECK_INTERVAL_MS, CHECK_INTERVAL_MS, MILLISECONDS);
}
streams.add(stream);
}
return stream;
}
private void removeStream(TimeoutInputStream stream) {
Future<?> toCancel = null;
synchronized (lock) {
if (streams.remove(stream) && streams.isEmpty()) {
toCancel = task;
task = null;
}
}
if (toCancel != null) toCancel.cancel(false);
}
@Scheduler
private void checkTimeouts() {
ioExecutor.execute(() -> {
List<TimeoutInputStream> snapshot;
synchronized (lock) {
snapshot = new ArrayList<>(streams);
}
for (TimeoutInputStream stream : snapshot) {
if (stream.hasTimedOut()) {
LOG.info("Input stream has timed out");
try {
stream.close();
} catch (IOException e) {
logException(LOG, INFO, e);
}
}
}
});
}
}

View File

@@ -130,8 +130,8 @@ class ConnectionManagerImpl implements ConnectionManager {
TransportConnectionWriter w) throws IOException { TransportConnectionWriter w) throws IOException {
StreamWriter streamWriter = streamWriterFactory.createStreamWriter( StreamWriter streamWriter = streamWriterFactory.createStreamWriter(
w.getOutputStream(), ctx); w.getOutputStream(), ctx);
ContactId c = requireNonNull(ctx.getContactId()); return syncSessionFactory.createSimplexOutgoingSession(
return syncSessionFactory.createSimplexOutgoingSession(c, requireNonNull(ctx.getContactId()), ctx.getTransportId(),
w.getMaxLatency(), streamWriter); w.getMaxLatency(), streamWriter);
} }
@@ -139,8 +139,8 @@ class ConnectionManagerImpl implements ConnectionManager {
TransportConnectionWriter w) throws IOException { TransportConnectionWriter w) throws IOException {
StreamWriter streamWriter = streamWriterFactory.createStreamWriter( StreamWriter streamWriter = streamWriterFactory.createStreamWriter(
w.getOutputStream(), ctx); w.getOutputStream(), ctx);
ContactId c = requireNonNull(ctx.getContactId()); return syncSessionFactory.createDuplexOutgoingSession(
return syncSessionFactory.createDuplexOutgoingSession(c, requireNonNull(ctx.getContactId()), ctx.getTransportId(),
w.getMaxLatency(), w.getMaxIdleTime(), streamWriter); w.getMaxLatency(), w.getMaxIdleTime(), streamWriter);
} }

View File

@@ -3,9 +3,30 @@ package org.briarproject.bramble.plugin.bluetooth;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection; import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection;
import static java.util.concurrent.TimeUnit.DAYS;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
@NotNullByDefault @NotNullByDefault
interface BluetoothConnectionLimiter { interface BluetoothConnectionLimiter {
/**
* How long a connection must remain open before it's considered stable.
*/
long STABILITY_PERIOD_MS = SECONDS.toMillis(90);
/**
* The minimum interval between attempts to raise the connection limit.
* This is longer than {@link #STABILITY_PERIOD_MS} so we don't start
* another attempt before knowing the outcome of the last one.
*/
long MIN_ATTEMPT_INTERVAL_MS = MINUTES.toMillis(2);
/**
* The maximum interval between attempts to raise the connection limit.
*/
long MAX_ATTEMPT_INTERVAL_MS = DAYS.toMillis(2);
/** /**
* Informs the limiter that key agreement has started. * Informs the limiter that key agreement has started.
*/ */
@@ -23,12 +44,12 @@ interface BluetoothConnectionLimiter {
boolean canOpenContactConnection(); boolean canOpenContactConnection();
/** /**
* Informs the limiter that a contact connection has been opened. The * Informs the limiter that a contact connection has been opened.
* limiter may close the new connection if key agreement is in progress.
* <p/> * <p/>
* Returns false if the limiter has closed the new connection. * Returns true if the connection is allowed.
*/ */
boolean contactConnectionOpened(DuplexTransportConnection conn); boolean contactConnectionOpened(DuplexTransportConnection conn,
boolean incoming);
/** /**
* Informs the limiter that a key agreement connection has been opened. * Informs the limiter that a key agreement connection has been opened.
@@ -37,11 +58,13 @@ interface BluetoothConnectionLimiter {
/** /**
* Informs the limiter that the given connection has been closed. * Informs the limiter that the given connection has been closed.
*
* @param exception True if the connection was closed due to an exception.
*/ */
void connectionClosed(DuplexTransportConnection conn); void connectionClosed(DuplexTransportConnection conn, boolean exception);
/** /**
* Informs the limiter that all connections have been closed. * Informs the limiter that the Bluetooth adapter has been disabled.
*/ */
void allConnectionsClosed(); void bluetoothDisabled();
} }

View File

@@ -1,46 +1,59 @@
package org.briarproject.bramble.plugin.bluetooth; 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.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection; import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection;
import org.briarproject.bramble.api.sync.event.CloseSyncConnectionsEvent;
import org.briarproject.bramble.api.system.Clock;
import java.io.IOException; import java.util.Iterator;
import java.util.ArrayList;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
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 static java.lang.Math.min;
import static java.util.logging.Level.INFO; 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.util.LogUtils.logException; import static org.briarproject.bramble.api.plugin.BluetoothConstants.ID;
@NotNullByDefault @NotNullByDefault
@ThreadSafe @ThreadSafe
class BluetoothConnectionLimiterImpl implements BluetoothConnectionLimiter { class BluetoothConnectionLimiterImpl implements BluetoothConnectionLimiter {
private static final Logger LOG = private static final Logger LOG =
Logger.getLogger(BluetoothConnectionLimiterImpl.class.getName()); getLogger(BluetoothConnectionLimiterImpl.class.getName());
private final EventBus eventBus;
private final Clock clock;
private final Object lock = new Object(); private final Object lock = new Object();
// The following are locking: lock @GuardedBy("lock")
private final LinkedList<DuplexTransportConnection> connections = private final List<ConnectionRecord> connections = new LinkedList<>();
new LinkedList<>(); @GuardedBy("lock")
private boolean keyAgreementInProgress = false; private boolean keyAgreementInProgress = false;
@GuardedBy("lock")
private int connectionLimit = 1;
@GuardedBy("lock")
private long timeOfLastAttempt = 0,
attemptInterval = MIN_ATTEMPT_INTERVAL_MS;
@Inject
BluetoothConnectionLimiterImpl(EventBus eventBus, Clock clock) {
this.eventBus = eventBus;
this.clock = clock;
}
@Override @Override
public void keyAgreementStarted() { public void keyAgreementStarted() {
List<DuplexTransportConnection> close;
synchronized (lock) { synchronized (lock) {
keyAgreementInProgress = true; keyAgreementInProgress = true;
close = new ArrayList<>(connections);
connections.clear();
} }
if (LOG.isLoggable(INFO)) { LOG.info("Key agreement started");
LOG.info("Key agreement started, closing " + close.size() + eventBus.broadcast(new CloseSyncConnectionsEvent(ID));
" connections");
}
for (DuplexTransportConnection conn : close) tryToClose(conn);
} }
@Override @Override
@@ -55,62 +68,128 @@ class BluetoothConnectionLimiterImpl implements BluetoothConnectionLimiter {
public boolean canOpenContactConnection() { public boolean canOpenContactConnection() {
synchronized (lock) { synchronized (lock) {
if (keyAgreementInProgress) { if (keyAgreementInProgress) {
LOG.info("Can't open contact connection during key agreement"); LOG.info("Refusing contact connection during key agreement");
return false; return false;
} else { } else {
LOG.info("Can open contact connection"); long now = clock.currentTimeMillis();
return true; return isContactConnectionAllowedByLimit(now);
} }
} }
} }
@Override @Override
public boolean contactConnectionOpened(DuplexTransportConnection conn) { public boolean contactConnectionOpened(DuplexTransportConnection conn,
boolean accept = true; boolean incoming) {
synchronized (lock) { synchronized (lock) {
if (keyAgreementInProgress) { if (keyAgreementInProgress) {
LOG.info("Refusing contact connection during key agreement"); LOG.info("Refusing contact connection during key agreement");
accept = false; return false;
} else { } else {
LOG.info("Accepting contact connection"); long now = clock.currentTimeMillis();
connections.add(conn); if (incoming || isContactConnectionAllowedByLimit(now)) {
connections.add(new ConnectionRecord(conn, now));
if (!incoming && connections.size() > connectionLimit) {
LOG.info("Attempting to raise connection limit");
timeOfLastAttempt = now;
}
return true;
} else {
return false;
}
} }
} }
if (!accept) tryToClose(conn);
return accept;
} }
@Override @Override
public void keyAgreementConnectionOpened(DuplexTransportConnection conn) { public void keyAgreementConnectionOpened(DuplexTransportConnection conn) {
synchronized (lock) { synchronized (lock) {
LOG.info("Accepting key agreement connection"); LOG.info("Accepting key agreement connection");
connections.add(conn); connections.add(
} new ConnectionRecord(conn, clock.currentTimeMillis()));
}
private void tryToClose(DuplexTransportConnection conn) {
try {
conn.getWriter().dispose(false);
conn.getReader().dispose(false, false);
} catch (IOException e) {
logException(LOG, WARNING, e);
} }
} }
@Override @Override
public void connectionClosed(DuplexTransportConnection conn) { public void connectionClosed(DuplexTransportConnection conn,
boolean exception) {
synchronized (lock) { synchronized (lock) {
connections.remove(conn); Iterator<ConnectionRecord> it = connections.iterator();
while (it.hasNext()) {
if (it.next().connection == conn) {
long now = clock.currentTimeMillis();
if (exception) connectionFailed(now);
else considerRaisingConnectionLimit(now);
it.remove();
break;
}
}
if (LOG.isLoggable(INFO)) if (LOG.isLoggable(INFO))
LOG.info("Connection closed, " + connections.size() + " open"); LOG.info("Connection closed, " + connections.size() + " open");
} }
} }
@Override @Override
public void allConnectionsClosed() { public void bluetoothDisabled() {
synchronized (lock) { synchronized (lock) {
LOG.info("Bluetooth disabled");
considerRaisingConnectionLimit(clock.currentTimeMillis());
connections.clear(); connections.clear();
LOG.info("All connections closed"); }
}
@GuardedBy("lock")
private boolean isContactConnectionAllowedByLimit(long now) {
considerRaisingConnectionLimit(now);
if (connections.size() > connectionLimit) {
LOG.info("Refusing contact connection, above limit");
return false;
} else if (connections.size() < connectionLimit) {
LOG.info("Allowing contact connection, below limit");
return true;
} else if (now - timeOfLastAttempt >= attemptInterval) {
LOG.info("Allowing contact connection, at limit");
return true;
} else {
LOG.info("Refusing contact connection, at limit");
return false;
}
}
@GuardedBy("lock")
private void considerRaisingConnectionLimit(long now) {
int stable = 0;
for (ConnectionRecord rec : connections) {
if (now - rec.timeOpened >= STABILITY_PERIOD_MS) stable++;
}
if (stable > connectionLimit) {
LOG.info("Raising connection limit");
connectionLimit = stable;
attemptInterval = MIN_ATTEMPT_INTERVAL_MS;
}
if (LOG.isLoggable(INFO)) {
LOG.info(stable + " connections are stable, limit is "
+ connectionLimit);
}
}
@GuardedBy("lock")
private void connectionFailed(long now) {
if (connections.size() > connectionLimit &&
now - timeOfLastAttempt < STABILITY_PERIOD_MS) {
LOG.info("Connection failed above limit, increasing interval");
attemptInterval = min(attemptInterval * 2, MAX_ATTEMPT_INTERVAL_MS);
}
}
private static final class ConnectionRecord {
private final DuplexTransportConnection connection;
private final long timeOpened;
private ConnectionRecord(DuplexTransportConnection connection,
long timeOpened) {
this.connection = connection;
this.timeOpened = timeOpened;
} }
} }
} }

View File

@@ -5,6 +5,7 @@ import org.briarproject.bramble.api.Pair;
import org.briarproject.bramble.api.data.BdfList; import org.briarproject.bramble.api.data.BdfList;
import org.briarproject.bramble.api.event.Event; import org.briarproject.bramble.api.event.Event;
import org.briarproject.bramble.api.event.EventListener; 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.KeyAgreementConnection;
import org.briarproject.bramble.api.keyagreement.KeyAgreementListener; import org.briarproject.bramble.api.keyagreement.KeyAgreementListener;
import org.briarproject.bramble.api.keyagreement.event.KeyAgreementListeningEvent; import org.briarproject.bramble.api.keyagreement.event.KeyAgreementListeningEvent;
@@ -60,12 +61,13 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
getLogger(BluetoothPlugin.class.getName()); getLogger(BluetoothPlugin.class.getName());
final BluetoothConnectionLimiter connectionLimiter; final BluetoothConnectionLimiter connectionLimiter;
final TimeoutMonitor timeoutMonitor;
private final Executor ioExecutor; private final Executor ioExecutor;
private final SecureRandom secureRandom; private final SecureRandom secureRandom;
private final Backoff backoff; private final Backoff backoff;
private final PluginCallback callback; private final PluginCallback callback;
private final int maxLatency; private final int maxLatency, maxIdleTime;
private final AtomicBoolean used = new AtomicBoolean(false); private final AtomicBoolean used = new AtomicBoolean(false);
private volatile boolean running = false, contactConnections = false; private volatile boolean running = false, contactConnections = false;
@@ -105,14 +107,17 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
abstract DuplexTransportConnection discoverAndConnect(String uuid); abstract DuplexTransportConnection discoverAndConnect(String uuid);
BluetoothPlugin(BluetoothConnectionLimiter connectionLimiter, BluetoothPlugin(BluetoothConnectionLimiter connectionLimiter,
Executor ioExecutor, SecureRandom secureRandom, TimeoutMonitor timeoutMonitor, Executor ioExecutor,
Backoff backoff, PluginCallback callback, int maxLatency) { SecureRandom secureRandom, Backoff backoff,
PluginCallback callback, int maxLatency, int maxIdleTime) {
this.connectionLimiter = connectionLimiter; this.connectionLimiter = connectionLimiter;
this.timeoutMonitor = timeoutMonitor;
this.ioExecutor = ioExecutor; this.ioExecutor = ioExecutor;
this.secureRandom = secureRandom; this.secureRandom = secureRandom;
this.backoff = backoff; this.backoff = backoff;
this.callback = callback; this.callback = callback;
this.maxLatency = maxLatency; this.maxLatency = maxLatency;
this.maxIdleTime = maxIdleTime;
} }
void onAdapterEnabled() { void onAdapterEnabled() {
@@ -125,7 +130,7 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
void onAdapterDisabled() { void onAdapterDisabled() {
LOG.info("Bluetooth disabled"); LOG.info("Bluetooth disabled");
tryToClose(socket); tryToClose(socket);
connectionLimiter.allConnectionsClosed(); connectionLimiter.bluetoothDisabled();
callback.transportDisabled(); callback.transportDisabled();
} }
@@ -141,8 +146,7 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
@Override @Override
public int getMaxIdleTime() { public int getMaxIdleTime() {
// Bluetooth detects dead connections so we don't need keepalives return maxIdleTime;
return Integer.MAX_VALUE;
} }
@Override @Override
@@ -227,13 +231,26 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
if (LOG.isLoggable(INFO)) LOG.info(e.toString()); if (LOG.isLoggable(INFO)) LOG.info(e.toString());
return; return;
} }
backoff.reset(); LOG.info("Connection received");
if (connectionLimiter.contactConnectionOpened(conn)) if (connectionLimiter.contactConnectionOpened(conn, true)) {
backoff.reset();
callback.handleConnection(conn); callback.handleConnection(conn);
} else {
tryToClose(conn);
}
if (!running) return; if (!running) return;
} }
} }
private void tryToClose(DuplexTransportConnection conn) {
try {
conn.getWriter().dispose(false);
conn.getReader().dispose(false, false);
} catch (IOException e) {
logException(LOG, WARNING, e);
}
}
@Override @Override
public void stop() { public void stop() {
running = false; running = false;
@@ -273,13 +290,10 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
String uuid = p.get(PROP_UUID); String uuid = p.get(PROP_UUID);
if (isNullOrEmpty(uuid)) return; if (isNullOrEmpty(uuid)) return;
ioExecutor.execute(() -> { ioExecutor.execute(() -> {
if (!isRunning() || !shouldAllowContactConnections()) return;
if (!connectionLimiter.canOpenContactConnection()) return;
DuplexTransportConnection d = createConnection(p); DuplexTransportConnection d = createConnection(p);
if (d != null) { if (d != null) {
backoff.reset(); backoff.reset();
if (connectionLimiter.contactConnectionOpened(d)) h.handleConnection(d);
h.handleConnection(d);
} }
}); });
} }
@@ -325,8 +339,12 @@ abstract class BluetoothPlugin<SS> implements DuplexPlugin, EventListener {
if (isNullOrEmpty(uuid)) return null; if (isNullOrEmpty(uuid)) return null;
DuplexTransportConnection conn = connect(address, uuid); DuplexTransportConnection conn = connect(address, uuid);
if (conn == null) return null; if (conn == null) return null;
// TODO: Why don't we reset the backoff here? if (connectionLimiter.contactConnectionOpened(conn, false)) {
return connectionLimiter.contactConnectionOpened(conn) ? conn : null; return conn;
} else {
tryToClose(conn);
return null;
}
} }
@Override @Override

View File

@@ -11,6 +11,7 @@ import org.briarproject.bramble.api.event.EventListener;
import org.briarproject.bramble.api.lifecycle.IoExecutor; import org.briarproject.bramble.api.lifecycle.IoExecutor;
import org.briarproject.bramble.api.lifecycle.event.LifecycleEvent; import org.briarproject.bramble.api.lifecycle.event.LifecycleEvent;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; 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.Ack;
import org.briarproject.bramble.api.sync.Message; import org.briarproject.bramble.api.sync.Message;
import org.briarproject.bramble.api.sync.Offer; import org.briarproject.bramble.api.sync.Offer;
@@ -18,6 +19,7 @@ import org.briarproject.bramble.api.sync.Request;
import org.briarproject.bramble.api.sync.SyncRecordWriter; import org.briarproject.bramble.api.sync.SyncRecordWriter;
import org.briarproject.bramble.api.sync.SyncSession; import org.briarproject.bramble.api.sync.SyncSession;
import org.briarproject.bramble.api.sync.Versions; 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.GroupVisibilityUpdatedEvent;
import org.briarproject.bramble.api.sync.event.MessageRequestedEvent; import org.briarproject.bramble.api.sync.event.MessageRequestedEvent;
import org.briarproject.bramble.api.sync.event.MessageSharedEvent; import org.briarproject.bramble.api.sync.event.MessageSharedEvent;
@@ -71,6 +73,7 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
private final EventBus eventBus; private final EventBus eventBus;
private final Clock clock; private final Clock clock;
private final ContactId contactId; private final ContactId contactId;
private final TransportId transportId;
private final int maxLatency, maxIdleTime; private final int maxLatency, maxIdleTime;
private final StreamWriter streamWriter; private final StreamWriter streamWriter;
private final SyncRecordWriter recordWriter; private final SyncRecordWriter recordWriter;
@@ -86,14 +89,15 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
private volatile boolean interrupted = false; private volatile boolean interrupted = false;
DuplexOutgoingSession(DatabaseComponent db, Executor dbExecutor, DuplexOutgoingSession(DatabaseComponent db, Executor dbExecutor,
EventBus eventBus, Clock clock, ContactId contactId, int maxLatency, EventBus eventBus, Clock clock, ContactId contactId,
int maxIdleTime, StreamWriter streamWriter, TransportId transportId, int maxLatency, int maxIdleTime,
SyncRecordWriter recordWriter) { StreamWriter streamWriter, SyncRecordWriter recordWriter) {
this.db = db; this.db = db;
this.dbExecutor = dbExecutor; this.dbExecutor = dbExecutor;
this.eventBus = eventBus; this.eventBus = eventBus;
this.clock = clock; this.clock = clock;
this.contactId = contactId; this.contactId = contactId;
this.transportId = transportId;
this.maxLatency = maxLatency; this.maxLatency = maxLatency;
this.maxIdleTime = maxIdleTime; this.maxIdleTime = maxIdleTime;
this.streamWriter = streamWriter; this.streamWriter = streamWriter;
@@ -223,6 +227,9 @@ class DuplexOutgoingSession implements SyncSession, EventListener {
} else if (e instanceof LifecycleEvent) { } else if (e instanceof LifecycleEvent) {
LifecycleEvent l = (LifecycleEvent) e; LifecycleEvent l = (LifecycleEvent) e;
if (l.getLifecycleState() == STOPPING) interrupt(); if (l.getLifecycleState() == STOPPING) interrupt();
} else if (e instanceof CloseSyncConnectionsEvent) {
CloseSyncConnectionsEvent c = (CloseSyncConnectionsEvent) e;
if (c.getTransportId().equals(transportId)) interrupt();
} }
} }

View File

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

View File

@@ -5,6 +5,7 @@ import org.briarproject.bramble.api.db.DatabaseComponent;
import org.briarproject.bramble.api.db.DatabaseExecutor; import org.briarproject.bramble.api.db.DatabaseExecutor;
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.TransportId;
import org.briarproject.bramble.api.sync.SyncRecordReader; import org.briarproject.bramble.api.sync.SyncRecordReader;
import org.briarproject.bramble.api.sync.SyncRecordReaderFactory; import org.briarproject.bramble.api.sync.SyncRecordReaderFactory;
import org.briarproject.bramble.api.sync.SyncRecordWriter; import org.briarproject.bramble.api.sync.SyncRecordWriter;
@@ -53,22 +54,23 @@ class SyncSessionFactoryImpl implements SyncSessionFactory {
} }
@Override @Override
public SyncSession createSimplexOutgoingSession(ContactId c, public SyncSession createSimplexOutgoingSession(ContactId c, TransportId t,
int maxLatency, StreamWriter streamWriter) { int maxLatency, StreamWriter streamWriter) {
OutputStream out = streamWriter.getOutputStream(); OutputStream out = streamWriter.getOutputStream();
SyncRecordWriter recordWriter = SyncRecordWriter recordWriter =
recordWriterFactory.createRecordWriter(out); recordWriterFactory.createRecordWriter(out);
return new SimplexOutgoingSession(db, dbExecutor, eventBus, c, return new SimplexOutgoingSession(db, dbExecutor, eventBus, c, t,
maxLatency, streamWriter, recordWriter); maxLatency, streamWriter, recordWriter);
} }
@Override @Override
public SyncSession createDuplexOutgoingSession(ContactId c, int maxLatency, public SyncSession createDuplexOutgoingSession(ContactId c,
int maxIdleTime, StreamWriter streamWriter) { TransportId t, int maxLatency, int maxIdleTime,
StreamWriter streamWriter) {
OutputStream out = streamWriter.getOutputStream(); OutputStream out = streamWriter.getOutputStream();
SyncRecordWriter recordWriter = SyncRecordWriter recordWriter =
recordWriterFactory.createRecordWriter(out); recordWriterFactory.createRecordWriter(out);
return new DuplexOutgoingSession(db, dbExecutor, eventBus, clock, c, return new DuplexOutgoingSession(db, dbExecutor, eventBus, clock, c, t,
maxLatency, maxIdleTime, streamWriter, recordWriter); maxLatency, maxIdleTime, streamWriter, recordWriter);
} }
} }

View File

@@ -0,0 +1,143 @@
package org.briarproject.bramble.io;
import org.briarproject.bramble.test.BrambleTestCase;
import org.briarproject.bramble.test.SettableClock;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class TimeoutInputStreamTest extends BrambleTestCase {
private static final long TIMEOUT_MS = MINUTES.toMillis(1);
private final long now = System.currentTimeMillis();
private AtomicLong time;
private UnresponsiveInputStream in;
private AtomicBoolean listenerCalled;
private TimeoutInputStream stream;
private CountDownLatch readReturned;
@Before
public void setUp() {
time = new AtomicLong(now);
in = new UnresponsiveInputStream();
listenerCalled = new AtomicBoolean(false);
stream = new TimeoutInputStream(new SettableClock(time), in,
TIMEOUT_MS, stream -> listenerCalled.set(true));
readReturned = new CountDownLatch(1);
}
@Test
public void testTimeoutIsReportedIfReadDoesNotReturn() throws Exception {
startReading();
try {
// The stream should not report a timeout
assertFalse(stream.hasTimedOut());
// Time passes
time.set(now + TIMEOUT_MS);
// The stream still shouldn't report a timeout
assertFalse(stream.hasTimedOut());
// Time passes
time.set(now + TIMEOUT_MS + 1);
// The stream should report a timeout
assertTrue(stream.hasTimedOut());
// The listener should not have been called yet
assertFalse(listenerCalled.get());
// Close the stream
stream.close();
// The listener should have been called
assertTrue(listenerCalled.get());
} finally {
// Allow the read to return
in.readFinished.countDown();
}
}
@Test
public void testTimeoutIsNotReportedIfReadReturns() throws Exception {
startReading();
try {
// The stream should not report a timeout
assertFalse(stream.hasTimedOut());
// Time passes
time.set(now + TIMEOUT_MS);
// The stream still shouldn't report a timeout
assertFalse(stream.hasTimedOut());
// Allow the read to finish and wait for it to return
in.readFinished.countDown();
readReturned.await(10, SECONDS);
// Time passes
time.set(now + TIMEOUT_MS + 1);
// The stream should not report a timeout as the read has returned
assertFalse(stream.hasTimedOut());
// The listener should not have been called yet
assertFalse(listenerCalled.get());
// Close the stream
stream.close();
// The listener should have been called
assertTrue(listenerCalled.get());
} finally {
// Allow the read to return in case an assertion was thrown
in.readFinished.countDown();
}
}
private void startReading() throws Exception {
// Start a background thread to read from the unresponsive stream
new Thread(() -> {
try {
assertEquals(123, stream.read());
readReturned.countDown();
} catch (IOException e) {
fail();
}
}).start();
// Wait for the background thread to start reading
assertTrue(in.readStarted.await(10, SECONDS));
}
private class UnresponsiveInputStream extends InputStream {
private final CountDownLatch readStarted = new CountDownLatch(1);
private final CountDownLatch readFinished = new CountDownLatch(1);
@Override
public int read() throws IOException {
readStarted.countDown();
try {
readFinished.await();
return 123;
} catch (InterruptedException e) {
throw new IOException(e);
}
}
}
}

View File

@@ -0,0 +1,182 @@
package org.briarproject.bramble.plugin.bluetooth;
import org.briarproject.bramble.api.event.EventBus;
import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection;
import org.briarproject.bramble.api.sync.event.CloseSyncConnectionsEvent;
import org.briarproject.bramble.api.system.Clock;
import org.briarproject.bramble.test.BrambleMockTestCase;
import org.briarproject.bramble.test.SettableClock;
import org.jmock.Expectations;
import org.junit.Before;
import org.junit.Test;
import java.util.concurrent.atomic.AtomicLong;
import static org.briarproject.bramble.plugin.bluetooth.BluetoothConnectionLimiter.MIN_ATTEMPT_INTERVAL_MS;
import static org.briarproject.bramble.plugin.bluetooth.BluetoothConnectionLimiter.STABILITY_PERIOD_MS;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class BluetoothConnectionLimiterImplTest extends BrambleMockTestCase {
private final EventBus eventBus = context.mock(EventBus.class);
private final DuplexTransportConnection conn1 =
context.mock(DuplexTransportConnection.class, "conn1");
private final DuplexTransportConnection conn2 =
context.mock(DuplexTransportConnection.class, "conn2");
private final DuplexTransportConnection conn3 =
context.mock(DuplexTransportConnection.class, "conn3");
private final long now = System.currentTimeMillis();
private AtomicLong time;
private BluetoothConnectionLimiter limiter;
@Before
public void setUp() {
time = new AtomicLong(now);
Clock clock = new SettableClock(time);
limiter = new BluetoothConnectionLimiterImpl(eventBus, clock);
}
@Test
public void testLimiterDoesNotAllowContactConnectionsDuringKeyAgreement() {
assertTrue(limiter.canOpenContactConnection());
expectCloseSyncConnectionsEvent();
limiter.keyAgreementStarted();
assertFalse(limiter.canOpenContactConnection());
limiter.keyAgreementEnded();
assertTrue(limiter.canOpenContactConnection());
}
@Test
public void testLimiterAllowsAttemptToRaiseLimitAtStartup() {
// First outgoing connection is allowed - we're below the limit of 1
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn1, false));
// Second outgoing connection is allowed - it's time to try raising
// the limit to 2
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn2, false));
// Third outgoing connection is not allowed - we're above the limit of 1
assertFalse(limiter.canOpenContactConnection());
}
@Test
public void testLimiterAllowsThirdConnectionAfterFirstTwoAreClosed() {
// First outgoing connection is allowed - we're below the limit of 1
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn1, false));
// Second outgoing connection is allowed - it's time to try raising
// the limit to 2
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn2, false));
// Third outgoing connection is not allowed - we're above the limit of 1
assertFalse(limiter.canOpenContactConnection());
// Close the first connection
limiter.connectionClosed(conn1, false);
// Third outgoing connection is not allowed - we're at the limit of 1
assertFalse(limiter.canOpenContactConnection());
// Close the second connection
limiter.connectionClosed(conn2, false);
// Third outgoing connection is allowed - we're below the limit of 1
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn3, false));
}
@Test
public void testLimiterRaisesLimitWhenConnectionsAreStable() {
// First outgoing connection is allowed - we're below the limit of 1
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn1, false));
// Second outgoing connection is allowed - it's time to try raising
// the limit to 2
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn2, false));
// Third outgoing connection is not allowed - we're above the limit of 1
assertFalse(limiter.canOpenContactConnection());
// Time passes
time.set(now + STABILITY_PERIOD_MS);
// Third outgoing connection is still not allowed - first two are now
// stable so limit is raised to 2, but we're already at the new limit
assertFalse(limiter.canOpenContactConnection());
// Time passes
time.set(now + MIN_ATTEMPT_INTERVAL_MS);
// Third outgoing connection is allowed - it's time to try raising
// the limit to 3
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn3, false));
// Fourth outgoing connection is not allowed - we're above the limit
// of 2
assertFalse(limiter.canOpenContactConnection());
}
@Test
public void testLimiterIncreasesIntervalWhenConnectionFailsAboveLimit() {
// First outgoing connection is allowed - we're below the limit of 1
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn1, false));
// Time passes
time.set(now + 1);
// Second outgoing connection is allowed - it's time to try raising
// the limit to 2
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn2, false));
// Time passes - the first connection is stable, the second isn't
time.set(now + STABILITY_PERIOD_MS);
// First connection fails. The second connection isn't stable yet, so
// the limiter considers this a failed attempt and doubles the interval
// between attempts
limiter.connectionClosed(conn1, true);
// Third outgoing connection is not allowed - we're still at the limit
// of 1
assertFalse(limiter.canOpenContactConnection());
// Time passes - nearly time for the second attempt
time.set(now + MIN_ATTEMPT_INTERVAL_MS * 2);
// Third outgoing connection is not allowed - we're still at the limit
// of 1
assertFalse(limiter.canOpenContactConnection());
// Time passes - now it's time for the second attempt
time.set(now + 1 + MIN_ATTEMPT_INTERVAL_MS * 2);
// Third outgoing connection is allowed - it's time to try raising the
// limit to 2 again
assertTrue(limiter.canOpenContactConnection());
assertTrue(limiter.contactConnectionOpened(conn3, false));
}
private void expectCloseSyncConnectionsEvent() {
context.checking(new Expectations() {{
oneOf(eventBus).broadcast(with(any(
CloseSyncConnectionsEvent.class)));
}});
}
}

View File

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

View File

@@ -1,6 +1,7 @@
package org.briarproject.bramble.plugin; package org.briarproject.bramble.plugin;
import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.event.EventBus;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.lifecycle.IoExecutor; import org.briarproject.bramble.api.lifecycle.IoExecutor;
import org.briarproject.bramble.api.lifecycle.ShutdownManager; import org.briarproject.bramble.api.lifecycle.ShutdownManager;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
@@ -9,6 +10,7 @@ import org.briarproject.bramble.api.plugin.PluginConfig;
import org.briarproject.bramble.api.plugin.duplex.DuplexPluginFactory; import org.briarproject.bramble.api.plugin.duplex.DuplexPluginFactory;
import org.briarproject.bramble.api.plugin.simplex.SimplexPluginFactory; import org.briarproject.bramble.api.plugin.simplex.SimplexPluginFactory;
import org.briarproject.bramble.api.reliability.ReliabilityLayerFactory; import org.briarproject.bramble.api.reliability.ReliabilityLayerFactory;
import org.briarproject.bramble.api.system.Clock;
import org.briarproject.bramble.plugin.bluetooth.JavaBluetoothPluginFactory; import org.briarproject.bramble.plugin.bluetooth.JavaBluetoothPluginFactory;
import org.briarproject.bramble.plugin.modem.ModemPluginFactory; import org.briarproject.bramble.plugin.modem.ModemPluginFactory;
import org.briarproject.bramble.plugin.tcp.LanTcpPluginFactory; import org.briarproject.bramble.plugin.tcp.LanTcpPluginFactory;
@@ -31,10 +33,11 @@ public class DesktopPluginModule extends PluginModule {
PluginConfig getPluginConfig(@IoExecutor Executor ioExecutor, PluginConfig getPluginConfig(@IoExecutor Executor ioExecutor,
SecureRandom random, BackoffFactory backoffFactory, SecureRandom random, BackoffFactory backoffFactory,
ReliabilityLayerFactory reliabilityFactory, ReliabilityLayerFactory reliabilityFactory,
ShutdownManager shutdownManager, EventBus eventBus) { ShutdownManager shutdownManager, EventBus eventBus, Clock clock,
DuplexPluginFactory bluetooth = TimeoutMonitor timeoutMonitor) {
new JavaBluetoothPluginFactory(ioExecutor, random, eventBus, DuplexPluginFactory bluetooth = new JavaBluetoothPluginFactory(
backoffFactory); ioExecutor, random, eventBus, clock, timeoutMonitor,
backoffFactory);
DuplexPluginFactory modem = new ModemPluginFactory(ioExecutor, DuplexPluginFactory modem = new ModemPluginFactory(ioExecutor,
reliabilityFactory); reliabilityFactory);
DuplexPluginFactory lan = new LanTcpPluginFactory(ioExecutor, DuplexPluginFactory lan = new LanTcpPluginFactory(ioExecutor,

View File

@@ -1,5 +1,6 @@
package org.briarproject.bramble.plugin.bluetooth; package org.briarproject.bramble.plugin.bluetooth;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.nullsafety.MethodsNotNullByDefault; import org.briarproject.bramble.api.nullsafety.MethodsNotNullByDefault;
import org.briarproject.bramble.api.nullsafety.ParametersNotNullByDefault; import org.briarproject.bramble.api.nullsafety.ParametersNotNullByDefault;
import org.briarproject.bramble.api.plugin.Backoff; import org.briarproject.bramble.api.plugin.Backoff;
@@ -34,10 +35,11 @@ class JavaBluetoothPlugin extends BluetoothPlugin<StreamConnectionNotifier> {
private volatile LocalDevice localDevice = null; private volatile LocalDevice localDevice = null;
JavaBluetoothPlugin(BluetoothConnectionLimiter connectionManager, JavaBluetoothPlugin(BluetoothConnectionLimiter connectionManager,
Executor ioExecutor, SecureRandom secureRandom, TimeoutMonitor timeoutMonitor, Executor ioExecutor,
Backoff backoff, PluginCallback callback, int maxLatency) { SecureRandom secureRandom, Backoff backoff,
super(connectionManager, ioExecutor, secureRandom, backoff, callback, PluginCallback callback, int maxLatency, int maxIdleTime) {
maxLatency); super(connectionManager, timeoutMonitor, ioExecutor, secureRandom,
backoff, callback, maxLatency, maxIdleTime);
} }
@Override @Override
@@ -119,7 +121,9 @@ class JavaBluetoothPlugin extends BluetoothPlugin<StreamConnectionNotifier> {
return "btspp://" + address + ":" + uuid + ";name=RFCOMM"; return "btspp://" + address + ":" + uuid + ";name=RFCOMM";
} }
private DuplexTransportConnection wrapSocket(StreamConnection s) { private DuplexTransportConnection wrapSocket(StreamConnection s)
return new JavaBluetoothTransportConnection(this, connectionLimiter, s); throws IOException {
return new JavaBluetoothTransportConnection(this, connectionLimiter,
timeoutMonitor, s);
} }
} }

View File

@@ -1,6 +1,7 @@
package org.briarproject.bramble.plugin.bluetooth; package org.briarproject.bramble.plugin.bluetooth;
import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.event.EventBus;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.Backoff; import org.briarproject.bramble.api.plugin.Backoff;
import org.briarproject.bramble.api.plugin.BackoffFactory; import org.briarproject.bramble.api.plugin.BackoffFactory;
@@ -8,6 +9,7 @@ import org.briarproject.bramble.api.plugin.PluginCallback;
import org.briarproject.bramble.api.plugin.TransportId; import org.briarproject.bramble.api.plugin.TransportId;
import org.briarproject.bramble.api.plugin.duplex.DuplexPlugin; import org.briarproject.bramble.api.plugin.duplex.DuplexPlugin;
import org.briarproject.bramble.api.plugin.duplex.DuplexPluginFactory; import org.briarproject.bramble.api.plugin.duplex.DuplexPluginFactory;
import org.briarproject.bramble.api.system.Clock;
import java.security.SecureRandom; import java.security.SecureRandom;
import java.util.concurrent.Executor; import java.util.concurrent.Executor;
@@ -21,22 +23,27 @@ import static org.briarproject.bramble.api.plugin.BluetoothConstants.ID;
public class JavaBluetoothPluginFactory implements DuplexPluginFactory { public class JavaBluetoothPluginFactory implements DuplexPluginFactory {
private static final int MAX_LATENCY = 30 * 1000; // 30 seconds private static final int MAX_LATENCY = 30 * 1000; // 30 seconds
private static final int MAX_IDLE_TIME = 30 * 1000; // 30 seconds
private static final int MIN_POLLING_INTERVAL = 60 * 1000; // 1 minute private static final int MIN_POLLING_INTERVAL = 60 * 1000; // 1 minute
private static final int MAX_POLLING_INTERVAL = 10 * 60 * 1000; // 10 mins private static final int MAX_POLLING_INTERVAL = 10 * 60 * 1000; // 10 mins
private static final double BACKOFF_BASE = 1.2; private static final double BACKOFF_BASE = 1.2;
private final Executor ioExecutor; private final Executor ioExecutor;
private final SecureRandom secureRandom; private final SecureRandom secureRandom;
private final BackoffFactory backoffFactory;
private final EventBus eventBus; private final EventBus eventBus;
private final Clock clock;
private final TimeoutMonitor timeoutMonitor;
private final BackoffFactory backoffFactory;
public JavaBluetoothPluginFactory(Executor ioExecutor, public JavaBluetoothPluginFactory(Executor ioExecutor,
SecureRandom secureRandom, EventBus eventBus, SecureRandom secureRandom, EventBus eventBus, Clock clock,
BackoffFactory backoffFactory) { TimeoutMonitor timeoutMonitor, BackoffFactory backoffFactory) {
this.ioExecutor = ioExecutor; this.ioExecutor = ioExecutor;
this.secureRandom = secureRandom; this.secureRandom = secureRandom;
this.backoffFactory = backoffFactory;
this.eventBus = eventBus; this.eventBus = eventBus;
this.clock = clock;
this.timeoutMonitor = timeoutMonitor;
this.backoffFactory = backoffFactory;
} }
@Override @Override
@@ -52,11 +59,12 @@ public class JavaBluetoothPluginFactory implements DuplexPluginFactory {
@Override @Override
public DuplexPlugin createPlugin(PluginCallback callback) { public DuplexPlugin createPlugin(PluginCallback callback) {
BluetoothConnectionLimiter connectionLimiter = BluetoothConnectionLimiter connectionLimiter =
new BluetoothConnectionLimiterImpl(); new BluetoothConnectionLimiterImpl(eventBus, clock);
Backoff backoff = backoffFactory.createBackoff(MIN_POLLING_INTERVAL, Backoff backoff = backoffFactory.createBackoff(MIN_POLLING_INTERVAL,
MAX_POLLING_INTERVAL, BACKOFF_BASE); MAX_POLLING_INTERVAL, BACKOFF_BASE);
JavaBluetoothPlugin plugin = new JavaBluetoothPlugin(connectionLimiter, JavaBluetoothPlugin plugin = new JavaBluetoothPlugin(connectionLimiter,
ioExecutor, secureRandom, backoff, callback, MAX_LATENCY); timeoutMonitor, ioExecutor, secureRandom, backoff, callback,
MAX_LATENCY, MAX_IDLE_TIME);
eventBus.addListener(plugin); eventBus.addListener(plugin);
return plugin; return plugin;
} }

View File

@@ -1,5 +1,6 @@
package org.briarproject.bramble.plugin.bluetooth; package org.briarproject.bramble.plugin.bluetooth;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.Plugin; import org.briarproject.bramble.api.plugin.Plugin;
import org.briarproject.bramble.api.plugin.duplex.AbstractDuplexTransportConnection; import org.briarproject.bramble.api.plugin.duplex.AbstractDuplexTransportConnection;
@@ -14,20 +15,24 @@ import javax.microedition.io.StreamConnection;
class JavaBluetoothTransportConnection class JavaBluetoothTransportConnection
extends AbstractDuplexTransportConnection { extends AbstractDuplexTransportConnection {
private final BluetoothConnectionLimiter connectionManager; private final BluetoothConnectionLimiter connectionLimiter;
private final StreamConnection stream; private final StreamConnection stream;
private final InputStream in;
JavaBluetoothTransportConnection(Plugin plugin, JavaBluetoothTransportConnection(Plugin plugin,
BluetoothConnectionLimiter connectionManager, BluetoothConnectionLimiter connectionLimiter,
StreamConnection stream) { TimeoutMonitor timeoutMonitor,
StreamConnection stream) throws IOException {
super(plugin); super(plugin);
this.connectionLimiter = connectionLimiter;
this.stream = stream; this.stream = stream;
this.connectionManager = connectionManager; in = timeoutMonitor.createTimeoutInputStream(
stream.openInputStream(), plugin.getMaxIdleTime() * 2);
} }
@Override @Override
protected InputStream getInputStream() throws IOException { protected InputStream getInputStream() {
return stream.openInputStream(); return in;
} }
@Override @Override
@@ -40,7 +45,7 @@ class JavaBluetoothTransportConnection
try { try {
stream.close(); stream.close();
} finally { } finally {
connectionManager.connectionClosed(this); connectionLimiter.connectionClosed(this, exception);
} }
} }
} }

View File

@@ -14,6 +14,7 @@ import org.briarproject.bramble.api.crypto.KeyStrengthener;
import org.briarproject.bramble.api.crypto.PublicKey; import org.briarproject.bramble.api.crypto.PublicKey;
import org.briarproject.bramble.api.db.DatabaseConfig; import org.briarproject.bramble.api.db.DatabaseConfig;
import org.briarproject.bramble.api.event.EventBus; import org.briarproject.bramble.api.event.EventBus;
import org.briarproject.bramble.api.io.TimeoutMonitor;
import org.briarproject.bramble.api.lifecycle.IoExecutor; import org.briarproject.bramble.api.lifecycle.IoExecutor;
import org.briarproject.bramble.api.lifecycle.LifecycleManager; import org.briarproject.bramble.api.lifecycle.LifecycleManager;
import org.briarproject.bramble.api.network.NetworkManager; import org.briarproject.bramble.api.network.NetworkManager;
@@ -122,11 +123,12 @@ public class AppModule {
LocationUtils locationUtils, EventBus eventBus, LocationUtils locationUtils, EventBus eventBus,
ResourceProvider resourceProvider, ResourceProvider resourceProvider,
CircumventionProvider circumventionProvider, CircumventionProvider circumventionProvider,
BatteryManager batteryManager, Clock clock) { BatteryManager batteryManager, Clock clock,
TimeoutMonitor timeoutMonitor) {
Context appContext = app.getApplicationContext(); Context appContext = app.getApplicationContext();
DuplexPluginFactory bluetooth = DuplexPluginFactory bluetooth = new AndroidBluetoothPluginFactory(
new AndroidBluetoothPluginFactory(ioExecutor, androidExecutor, ioExecutor, androidExecutor, appContext, random, eventBus,
appContext, random, eventBus, clock, backoffFactory); clock, timeoutMonitor, backoffFactory);
DuplexPluginFactory tor = new AndroidTorPluginFactory(ioExecutor, DuplexPluginFactory tor = new AndroidTorPluginFactory(ioExecutor,
scheduler, appContext, networkManager, locationUtils, eventBus, scheduler, appContext, networkManager, locationUtils, eventBus,
torSocketFactory, backoffFactory, resourceProvider, torSocketFactory, backoffFactory, resourceProvider,