Skip to content

briar
briar
Commit 4fcce711 authored by akwizgran's avatar akwizgran
Browse files
Options

Decouple poller from plugin manager.

parent 421ca309
Pipeline #3306 passed with stage
    in 6 minutes and 50 seconds
    Hide whitespace changes
    Inline Side-by-side
    Showing with 308 additions and 341 deletions
    bramble-core/src/main/java/org/briarproject/bramble/plugin/PluginManagerImpl.java
    View file @ 4fcce711
    ......@@ -8,7 +8,6 @@ import org.briarproject.bramble.api.lifecycle.Service;
    import org.briarproject.bramble.api.lifecycle.ServiceException;
    import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
    import org.briarproject.bramble.api.plugin.ConnectionManager;
    import org.briarproject.bramble.api.plugin.ConnectionRegistry;
    import org.briarproject.bramble.api.plugin.Plugin;
    import org.briarproject.bramble.api.plugin.PluginCallback;
    import org.briarproject.bramble.api.plugin.PluginConfig;
    ......@@ -30,10 +29,7 @@ import org.briarproject.bramble.api.properties.TransportProperties;
    import org.briarproject.bramble.api.properties.TransportPropertyManager;
    import org.briarproject.bramble.api.settings.Settings;
    import org.briarproject.bramble.api.settings.SettingsManager;
    import org.briarproject.bramble.api.system.Clock;
    import org.briarproject.bramble.api.system.Scheduler;
    import java.security.SecureRandom;
    import java.util.ArrayList;
    import java.util.Collection;
    import java.util.List;
    ......@@ -42,7 +38,6 @@ import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.CopyOnWriteArrayList;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.atomic.AtomicBoolean;
    import java.util.logging.Logger;
    ......@@ -64,15 +59,11 @@ class PluginManagerImpl implements PluginManager, Service {
    Logger.getLogger(PluginManagerImpl.class.getName());
    private final Executor ioExecutor;
    private final ScheduledExecutorService scheduler;
    private final EventBus eventBus;
    private final PluginConfig pluginConfig;
    private final ConnectionManager connectionManager;
    private final ConnectionRegistry connectionRegistry;
    private final SettingsManager settingsManager;
    private final TransportPropertyManager transportPropertyManager;
    private final SecureRandom random;
    private final Clock clock;
    private final Map<TransportId, Plugin> plugins;
    private final List<SimplexPlugin> simplexPlugins;
    private final List<DuplexPlugin> duplexPlugins;
    ......@@ -80,41 +71,25 @@ class PluginManagerImpl implements PluginManager, Service {
    private final AtomicBoolean used = new AtomicBoolean(false);
    @Inject
    PluginManagerImpl(@IoExecutor Executor ioExecutor,
    @Scheduler ScheduledExecutorService scheduler, EventBus eventBus,
    PluginManagerImpl(@IoExecutor Executor ioExecutor, EventBus eventBus,
    PluginConfig pluginConfig, ConnectionManager connectionManager,
    ConnectionRegistry connectionRegistry,
    SettingsManager settingsManager,
    TransportPropertyManager transportPropertyManager,
    SecureRandom random, Clock clock) {
    TransportPropertyManager transportPropertyManager) {
    this.ioExecutor = ioExecutor;
    this.scheduler = scheduler;
    this.eventBus = eventBus;
    this.pluginConfig = pluginConfig;
    this.connectionManager = connectionManager;
    this.connectionRegistry = connectionRegistry;
    this.settingsManager = settingsManager;
    this.transportPropertyManager = transportPropertyManager;
    this.random = random;
    this.clock = clock;
    plugins = new ConcurrentHashMap<>();
    simplexPlugins = new CopyOnWriteArrayList<>();
    duplexPlugins = new CopyOnWriteArrayList<>();
    startLatches = new ConcurrentHashMap<>();
    }
    @Override
    public void startService() {
    if (used.getAndSet(true)) throw new IllegalStateException();
    // Instantiate the poller
    if (pluginConfig.shouldPoll()) {
    LOG.info("Starting poller");
    Poller poller = new Poller(ioExecutor, scheduler, connectionManager,
    connectionRegistry, this, transportPropertyManager, random,
    clock);
    eventBus.addListener(poller);
    }
    // Instantiate the simplex plugins and start them asynchronously
    LOG.info("Starting simplex plugins");
    for (SimplexPluginFactory f : pluginConfig.getSimplexFactories()) {
    ......
    bramble-core/src/main/java/org/briarproject/bramble/plugin/PluginModule.java
    View file @ 4fcce711
    package org.briarproject.bramble.plugin;
    import org.briarproject.bramble.api.event.EventBus;
    import org.briarproject.bramble.api.lifecycle.LifecycleManager;
    import org.briarproject.bramble.api.plugin.BackoffFactory;
    import org.briarproject.bramble.api.plugin.ConnectionManager;
    import org.briarproject.bramble.api.plugin.ConnectionRegistry;
    import org.briarproject.bramble.api.plugin.PluginConfig;
    import org.briarproject.bramble.api.plugin.PluginManager;
    import javax.inject.Inject;
    ......@@ -18,6 +20,8 @@ public class PluginModule {
    public static class EagerSingletons {
    @Inject
    PluginManager pluginManager;
    @Inject
    Poller poller;
    }
    @Provides
    ......@@ -46,4 +50,12 @@ public class PluginModule {
    lifecycleManager.registerService(pluginManager);
    return pluginManager;
    }
    @Provides
    @Singleton
    Poller providePoller(PluginConfig config, EventBus eventBus,
    PollerImpl poller) {
    if (config.shouldPoll()) eventBus.addListener(poller);
    return poller;
    }
    }
    bramble-core/src/main/java/org/briarproject/bramble/plugin/Poller.java
    View file @ 4fcce711
    package org.briarproject.bramble.plugin;
    import org.briarproject.bramble.api.contact.ContactId;
    import org.briarproject.bramble.api.contact.event.ContactAddedEvent;
    import org.briarproject.bramble.api.db.DbException;
    import org.briarproject.bramble.api.event.Event;
    import org.briarproject.bramble.api.event.EventListener;
    import org.briarproject.bramble.api.lifecycle.IoExecutor;
    import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
    import org.briarproject.bramble.api.plugin.ConnectionManager;
    import org.briarproject.bramble.api.plugin.ConnectionRegistry;
    import org.briarproject.bramble.api.plugin.Plugin;
    import org.briarproject.bramble.api.plugin.PluginManager;
    import org.briarproject.bramble.api.plugin.TransportConnectionWriter;
    import org.briarproject.bramble.api.plugin.TransportId;
    import org.briarproject.bramble.api.plugin.duplex.DuplexPlugin;
    import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection;
    import org.briarproject.bramble.api.plugin.event.ConnectionClosedEvent;
    import org.briarproject.bramble.api.plugin.event.ConnectionOpenedEvent;
    import org.briarproject.bramble.api.plugin.event.TransportDisabledEvent;
    import org.briarproject.bramble.api.plugin.event.TransportEnabledEvent;
    import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
    import org.briarproject.bramble.api.properties.TransportProperties;
    import org.briarproject.bramble.api.properties.TransportPropertyManager;
    import org.briarproject.bramble.api.system.Clock;
    import org.briarproject.bramble.api.system.Scheduler;
    import java.security.SecureRandom;
    import java.util.Collection;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.Executor;
    import java.util.concurrent.Future;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    import java.util.logging.Logger;
    import javax.annotation.concurrent.ThreadSafe;
    import static java.util.concurrent.TimeUnit.MILLISECONDS;
    import static java.util.logging.Level.INFO;
    import static java.util.logging.Level.WARNING;
    import static org.briarproject.bramble.util.LogUtils.logException;
    @ThreadSafe
    @NotNullByDefault
    class Poller implements EventListener {
    private static final Logger LOG = Logger.getLogger(Poller.class.getName());
    private final Executor ioExecutor;
    private final ScheduledExecutorService scheduler;
    private final ConnectionManager connectionManager;
    private final ConnectionRegistry connectionRegistry;
    private final PluginManager pluginManager;
    private final TransportPropertyManager transportPropertyManager;
    private final SecureRandom random;
    private final Clock clock;
    private final Lock lock;
    private final Map<TransportId, ScheduledPollTask> tasks; // Locking: lock
    Poller(@IoExecutor Executor ioExecutor,
    @Scheduler ScheduledExecutorService scheduler,
    ConnectionManager connectionManager,
    ConnectionRegistry connectionRegistry, PluginManager pluginManager,
    TransportPropertyManager transportPropertyManager,
    SecureRandom random, Clock clock) {
    this.ioExecutor = ioExecutor;
    this.scheduler = scheduler;
    this.connectionManager = connectionManager;
    this.connectionRegistry = connectionRegistry;
    this.pluginManager = pluginManager;
    this.transportPropertyManager = transportPropertyManager;
    this.random = random;
    this.clock = clock;
    lock = new ReentrantLock();
    tasks = new HashMap<>();
    }
    @Override
    public void eventOccurred(Event e) {
    if (e instanceof ContactAddedEvent) {
    ContactAddedEvent c = (ContactAddedEvent) e;
    // Connect to the newly activated contact
    connectToContact(c.getContactId());
    } else if (e instanceof ConnectionClosedEvent) {
    ConnectionClosedEvent c = (ConnectionClosedEvent) e;
    // Reschedule polling, the polling interval may have decreased
    reschedule(c.getTransportId());
    if (!c.isIncoming()) {
    // Connect to the disconnected contact
    connectToContact(c.getContactId(), c.getTransportId());
    }
    } else if (e instanceof ConnectionOpenedEvent) {
    ConnectionOpenedEvent c = (ConnectionOpenedEvent) e;
    // Reschedule polling, the polling interval may have decreased
    reschedule(c.getTransportId());
    } else if (e instanceof TransportEnabledEvent) {
    TransportEnabledEvent t = (TransportEnabledEvent) e;
    // Poll the newly enabled transport
    pollNow(t.getTransportId());
    } else if (e instanceof TransportDisabledEvent) {
    TransportDisabledEvent t = (TransportDisabledEvent) e;
    // Cancel polling for the disabled transport
    cancel(t.getTransportId());
    }
    }
    private void connectToContact(ContactId c) {
    for (SimplexPlugin s : pluginManager.getSimplexPlugins())
    if (s.shouldPoll()) connectToContact(c, s);
    for (DuplexPlugin d : pluginManager.getDuplexPlugins())
    if (d.shouldPoll()) connectToContact(c, d);
    }
    private void connectToContact(ContactId c, TransportId t) {
    Plugin p = pluginManager.getPlugin(t);
    if (p instanceof SimplexPlugin && p.shouldPoll())
    connectToContact(c, (SimplexPlugin) p);
    else if (p instanceof DuplexPlugin && p.shouldPoll())
    connectToContact(c, (DuplexPlugin) p);
    }
    private void connectToContact(ContactId c, SimplexPlugin p) {
    ioExecutor.execute(() -> {
    TransportId t = p.getId();
    if (connectionRegistry.isConnected(c, t)) return;
    try {
    TransportProperties props =
    transportPropertyManager.getRemoteProperties(c, t);
    TransportConnectionWriter w = p.createWriter(props);
    if (w != null)
    connectionManager.manageOutgoingConnection(c, t, w);
    } catch (DbException e) {
    logException(LOG, WARNING, e);
    }
    });
    }
    private void connectToContact(ContactId c, DuplexPlugin p) {
    ioExecutor.execute(() -> {
    TransportId t = p.getId();
    if (connectionRegistry.isConnected(c, t)) return;
    try {
    TransportProperties props =
    transportPropertyManager.getRemoteProperties(c, t);
    DuplexTransportConnection d = p.createConnection(props);
    if (d != null)
    connectionManager.manageOutgoingConnection(c, t, d);
    } catch (DbException e) {
    logException(LOG, WARNING, e);
    }
    });
    }
    private void reschedule(TransportId t) {
    Plugin p = pluginManager.getPlugin(t);
    if (p != null && p.shouldPoll())
    schedule(p, p.getPollingInterval(), false);
    }
    private void pollNow(TransportId t) {
    Plugin p = pluginManager.getPlugin(t);
    // Randomise next polling interval
    if (p != null && p.shouldPoll()) schedule(p, 0, true);
    }
    private void schedule(Plugin p, int delay, boolean randomiseNext) {
    // Replace any later scheduled task for this plugin
    long due = clock.currentTimeMillis() + delay;
    TransportId t = p.getId();
    lock.lock();
    try {
    ScheduledPollTask scheduled = tasks.get(t);
    if (scheduled == null || due < scheduled.task.due) {
    // If a later task exists, cancel it. If it's already started
    // it will abort safely when it finds it's been replaced
    if (scheduled != null) scheduled.future.cancel(false);
    PollTask task = new PollTask(p, due, randomiseNext);
    Future future = scheduler.schedule(
    () -> ioExecutor.execute(task), delay, MILLISECONDS);
    tasks.put(t, new ScheduledPollTask(task, future));
    }
    } finally {
    lock.unlock();
    }
    }
    private void cancel(TransportId t) {
    lock.lock();
    try {
    ScheduledPollTask scheduled = tasks.remove(t);
    if (scheduled != null) scheduled.future.cancel(false);
    } finally {
    lock.unlock();
    }
    }
    @IoExecutor
    private void poll(Plugin p) {
    TransportId t = p.getId();
    if (LOG.isLoggable(INFO)) LOG.info("Polling plugin " + t);
    try {
    Map<ContactId, TransportProperties> remote =
    transportPropertyManager.getRemoteProperties(t);
    Collection<ContactId> connected =
    connectionRegistry.getConnectedContacts(t);
    remote = new HashMap<>(remote);
    remote.keySet().removeAll(connected);
    if (!remote.isEmpty()) p.poll(remote);
    } catch (DbException e) {
    logException(LOG, WARNING, e);
    }
    }
    private class ScheduledPollTask {
    private final PollTask task;
    private final Future future;
    private ScheduledPollTask(PollTask task, Future future) {
    this.task = task;
    this.future = future;
    }
    }
    private class PollTask implements Runnable {
    private final Plugin plugin;
    private final long due;
    private final boolean randomiseNext;
    private PollTask(Plugin plugin, long due, boolean randomiseNext) {
    this.plugin = plugin;
    this.due = due;
    this.randomiseNext = randomiseNext;
    }
    @Override
    @IoExecutor
    public void run() {
    lock.lock();
    try {
    TransportId t = plugin.getId();
    ScheduledPollTask scheduled = tasks.get(t);
    if (scheduled != null && scheduled.task != this)
    return; // Replaced by another task
    tasks.remove(t);
    } finally {
    lock.unlock();
    }
    int delay = plugin.getPollingInterval();
    if (randomiseNext) delay = (int) (delay * random.nextDouble());
    schedule(plugin, delay, false);
    poll(plugin);
    }
    }
    /**
    * Empty interface used for injecting the poller.
    */
    interface Poller {
    }
    bramble-core/src/main/java/org/briarproject/bramble/plugin/PollerImpl.java 0 → 100644
    View file @ 4fcce711
    package org.briarproject.bramble.plugin;
    import org.briarproject.bramble.api.contact.ContactId;
    import org.briarproject.bramble.api.contact.event.ContactAddedEvent;
    import org.briarproject.bramble.api.db.DbException;
    import org.briarproject.bramble.api.event.Event;
    import org.briarproject.bramble.api.event.EventListener;
    import org.briarproject.bramble.api.lifecycle.IoExecutor;
    import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
    import org.briarproject.bramble.api.plugin.ConnectionManager;
    import org.briarproject.bramble.api.plugin.ConnectionRegistry;
    import org.briarproject.bramble.api.plugin.Plugin;
    import org.briarproject.bramble.api.plugin.PluginManager;
    import org.briarproject.bramble.api.plugin.TransportConnectionWriter;
    import org.briarproject.bramble.api.plugin.TransportId;
    import org.briarproject.bramble.api.plugin.duplex.DuplexPlugin;
    import org.briarproject.bramble.api.plugin.duplex.DuplexTransportConnection;
    import org.briarproject.bramble.api.plugin.event.ConnectionClosedEvent;
    import org.briarproject.bramble.api.plugin.event.ConnectionOpenedEvent;
    import org.briarproject.bramble.api.plugin.event.TransportDisabledEvent;
    import org.briarproject.bramble.api.plugin.event.TransportEnabledEvent;
    import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
    import org.briarproject.bramble.api.properties.TransportProperties;
    import org.briarproject.bramble.api.properties.TransportPropertyManager;
    import org.briarproject.bramble.api.system.Clock;
    import org.briarproject.bramble.api.system.Scheduler;
    import java.security.SecureRandom;
    import java.util.Collection;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.Executor;
    import java.util.concurrent.Future;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    import java.util.logging.Logger;
    import javax.annotation.concurrent.GuardedBy;
    import javax.annotation.concurrent.ThreadSafe;
    import javax.inject.Inject;
    import static java.util.concurrent.TimeUnit.MILLISECONDS;
    import static java.util.logging.Level.INFO;
    import static java.util.logging.Level.WARNING;
    import static java.util.logging.Logger.getLogger;
    import static org.briarproject.bramble.util.LogUtils.logException;
    @ThreadSafe
    @NotNullByDefault
    class PollerImpl implements Poller, EventListener {
    private static final Logger LOG = getLogger(PollerImpl.class.getName());
    private final Executor ioExecutor;
    private final ScheduledExecutorService scheduler;
    private final ConnectionManager connectionManager;
    private final ConnectionRegistry connectionRegistry;
    private final PluginManager pluginManager;
    private final TransportPropertyManager transportPropertyManager;
    private final SecureRandom random;
    private final Clock clock;
    private final Lock lock;
    @GuardedBy("lock")
    private final Map<TransportId, ScheduledPollTask> tasks;
    @Inject
    PollerImpl(@IoExecutor Executor ioExecutor,
    @Scheduler ScheduledExecutorService scheduler,
    ConnectionManager connectionManager,
    ConnectionRegistry connectionRegistry, PluginManager pluginManager,
    TransportPropertyManager transportPropertyManager,
    SecureRandom random, Clock clock) {
    this.ioExecutor = ioExecutor;
    this.scheduler = scheduler;
    this.connectionManager = connectionManager;
    this.connectionRegistry = connectionRegistry;
    this.pluginManager = pluginManager;
    this.transportPropertyManager = transportPropertyManager;
    this.random = random;
    this.clock = clock;
    lock = new ReentrantLock();
    tasks = new HashMap<>();
    }
    @Override
    public void eventOccurred(Event e) {
    if (e instanceof ContactAddedEvent) {
    ContactAddedEvent c = (ContactAddedEvent) e;
    // Connect to the newly added contact
    connectToContact(c.getContactId());
    } else if (e instanceof ConnectionClosedEvent) {
    ConnectionClosedEvent c = (ConnectionClosedEvent) e;
    // Reschedule polling, the polling interval may have decreased
    reschedule(c.getTransportId());
    if (!c.isIncoming()) {
    // Connect to the disconnected contact
    connectToContact(c.getContactId(), c.getTransportId());
    }
    } else if (e instanceof ConnectionOpenedEvent) {
    ConnectionOpenedEvent c = (ConnectionOpenedEvent) e;
    // Reschedule polling, the polling interval may have decreased
    reschedule(c.getTransportId());
    } else if (e instanceof TransportEnabledEvent) {
    TransportEnabledEvent t = (TransportEnabledEvent) e;
    // Poll the newly enabled transport
    pollNow(t.getTransportId());
    } else if (e instanceof TransportDisabledEvent) {
    TransportDisabledEvent t = (TransportDisabledEvent) e;
    // Cancel polling for the disabled transport
    cancel(t.getTransportId());
    }
    }
    private void connectToContact(ContactId c) {
    for (SimplexPlugin s : pluginManager.getSimplexPlugins())
    if (s.shouldPoll()) connectToContact(c, s);
    for (DuplexPlugin d : pluginManager.getDuplexPlugins())
    if (d.shouldPoll()) connectToContact(c, d);
    }
    private void connectToContact(ContactId c, TransportId t) {
    Plugin p = pluginManager.getPlugin(t);
    if (p instanceof SimplexPlugin && p.shouldPoll())
    connectToContact(c, (SimplexPlugin) p);
    else if (p instanceof DuplexPlugin && p.shouldPoll())
    connectToContact(c, (DuplexPlugin) p);
    }
    private void connectToContact(ContactId c, SimplexPlugin p) {
    ioExecutor.execute(() -> {
    TransportId t = p.getId();
    if (connectionRegistry.isConnected(c, t)) return;
    try {
    TransportProperties props =
    transportPropertyManager.getRemoteProperties(c, t);
    TransportConnectionWriter w = p.createWriter(props);
    if (w != null)
    connectionManager.manageOutgoingConnection(c, t, w);
    } catch (DbException e) {
    logException(LOG, WARNING, e);
    }
    });
    }
    private void connectToContact(ContactId c, DuplexPlugin p) {
    ioExecutor.execute(() -> {
    TransportId t = p.getId();
    if (connectionRegistry.isConnected(c, t)) return;
    try {
    TransportProperties props =
    transportPropertyManager.getRemoteProperties(c, t);
    DuplexTransportConnection d = p.createConnection(props);
    if (d != null)
    connectionManager.manageOutgoingConnection(c, t, d);
    } catch (DbException e) {
    logException(LOG, WARNING, e);
    }
    });
    }
    private void reschedule(TransportId t) {
    Plugin p = pluginManager.getPlugin(t);
    if (p != null && p.shouldPoll())
    schedule(p, p.getPollingInterval(), false);
    }
    private void pollNow(TransportId t) {
    Plugin p = pluginManager.getPlugin(t);
    // Randomise next polling interval
    if (p != null && p.shouldPoll()) schedule(p, 0, true);
    }
    private void schedule(Plugin p, int delay, boolean randomiseNext) {
    // Replace any later scheduled task for this plugin
    long due = clock.currentTimeMillis() + delay;
    TransportId t = p.getId();
    lock.lock();
    try {
    ScheduledPollTask scheduled = tasks.get(t);
    if (scheduled == null || due < scheduled.task.due) {
    // If a later task exists, cancel it. If it's already started
    // it will abort safely when it finds it's been replaced
    if (scheduled != null) scheduled.future.cancel(false);
    PollTask task = new PollTask(p, due, randomiseNext);
    Future future = scheduler.schedule(() ->
    ioExecutor.execute(task), delay, MILLISECONDS);
    tasks.put(t, new ScheduledPollTask(task, future));
    }
    } finally {
    lock.unlock();
    }
    }
    private void cancel(TransportId t) {
    lock.lock();
    try {
    ScheduledPollTask scheduled = tasks.remove(t);
    if (scheduled != null) scheduled.future.cancel(false);
    } finally {
    lock.unlock();
    }
    }
    @IoExecutor
    private void poll(Plugin p) {
    TransportId t = p.getId();
    if (LOG.isLoggable(INFO)) LOG.info("Polling plugin " + t);
    try {
    Map<ContactId, TransportProperties> remote =
    transportPropertyManager.getRemoteProperties(t);
    Collection<ContactId> connected =
    connectionRegistry.getConnectedContacts(t);
    remote = new HashMap<>(remote);
    remote.keySet().removeAll(connected);
    if (!remote.isEmpty()) p.poll(remote);
    } catch (DbException e) {
    logException(LOG, WARNING, e);
    }
    }
    private class ScheduledPollTask {
    private final PollTask task;
    private final Future future;
    private ScheduledPollTask(PollTask task, Future future) {
    this.task = task;
    this.future = future;
    }
    }
    private class PollTask implements Runnable {
    private final Plugin plugin;
    private final long due;
    private final boolean randomiseNext;
    private PollTask(Plugin plugin, long due, boolean randomiseNext) {
    this.plugin = plugin;
    this.due = due;
    this.randomiseNext = randomiseNext;
    }
    @Override
    @IoExecutor
    public void run() {
    lock.lock();
    try {
    TransportId t = plugin.getId();
    ScheduledPollTask scheduled = tasks.get(t);
    if (scheduled != null && scheduled.task != this)
    return; // Replaced by another task
    tasks.remove(t);
    } finally {
    lock.unlock();
    }
    int delay = plugin.getPollingInterval();
    if (randomiseNext) delay = (int) (delay * random.nextDouble());
    schedule(plugin, delay, false);
    poll(plugin);
    }
    }
    }
    bramble-core/src/test/java/org/briarproject/bramble/plugin/PluginManagerImplTest.java
    View file @ 4fcce711
    ......@@ -2,7 +2,6 @@ package org.briarproject.bramble.plugin;
    import org.briarproject.bramble.api.event.EventBus;
    import org.briarproject.bramble.api.plugin.ConnectionManager;
    import org.briarproject.bramble.api.plugin.ConnectionRegistry;
    import org.briarproject.bramble.api.plugin.PluginConfig;
    import org.briarproject.bramble.api.plugin.PluginException;
    import org.briarproject.bramble.api.plugin.TransportId;
    ......@@ -14,18 +13,15 @@ import org.briarproject.bramble.api.plugin.simplex.SimplexPluginCallback;
    import org.briarproject.bramble.api.plugin.simplex.SimplexPluginFactory;
    import org.briarproject.bramble.api.properties.TransportPropertyManager;
    import org.briarproject.bramble.api.settings.SettingsManager;
    import org.briarproject.bramble.api.system.Clock;
    import org.briarproject.bramble.test.BrambleTestCase;
    import org.jmock.Expectations;
    import org.jmock.Mockery;
    import org.jmock.lib.concurrent.Synchroniser;
    import org.junit.Test;
    import java.security.SecureRandom;
    import java.util.Arrays;
    import java.util.concurrent.Executor;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import static org.briarproject.bramble.test.TestUtils.getTransportId;
    ......@@ -37,16 +33,10 @@ public class PluginManagerImplTest extends BrambleTestCase {
    setThreadingPolicy(new Synchroniser());
    }};
    Executor ioExecutor = Executors.newSingleThreadExecutor();
    ScheduledExecutorService scheduler =
    context.mock(ScheduledExecutorService.class);