Merge branch 'poller-refactoring' into 'master'

Poller refactoring, replace Timer with ScheduledExecutorService

* Replace Timer with ScheduledExecutorService (closes #258)
* Move automatic connection logic from PluginManager to Poller
* Reschedule polling when connections are opened or closed, making the poller more responsive to reductions in the polling interval


See merge request !180
This commit is contained in:
akwizgran
2016-05-11 14:45:50 +00:00
30 changed files with 876 additions and 648 deletions

View File

@@ -1,20 +1,13 @@
package org.briarproject.plugins;
import org.briarproject.BriarTestCase;
import org.briarproject.ImmediateExecutor;
import org.briarproject.api.TransportId;
import org.briarproject.api.contact.ContactId;
import org.briarproject.api.event.ContactStatusChangedEvent;
import org.briarproject.api.event.EventBus;
import org.briarproject.api.event.EventListener;
import org.briarproject.api.plugins.ConnectionManager;
import org.briarproject.api.plugins.ConnectionRegistry;
import org.briarproject.api.plugins.PluginConfig;
import org.briarproject.api.plugins.TransportConnectionWriter;
import org.briarproject.api.plugins.duplex.DuplexPlugin;
import org.briarproject.api.plugins.duplex.DuplexPluginCallback;
import org.briarproject.api.plugins.duplex.DuplexPluginFactory;
import org.briarproject.api.plugins.duplex.DuplexTransportConnection;
import org.briarproject.api.plugins.simplex.SimplexPlugin;
import org.briarproject.api.plugins.simplex.SimplexPluginCallback;
import org.briarproject.api.plugins.simplex.SimplexPluginFactory;
@@ -40,11 +33,8 @@ public class PluginManagerImplTest extends BriarTestCase {
final Executor ioExecutor = Executors.newSingleThreadExecutor();
final EventBus eventBus = context.mock(EventBus.class);
final PluginConfig pluginConfig = context.mock(PluginConfig.class);
final Poller poller = context.mock(Poller.class);
final ConnectionManager connectionManager =
context.mock(ConnectionManager.class);
final ConnectionRegistry connectionRegistry =
context.mock(ConnectionRegistry.class);
final SettingsManager settingsManager =
context.mock(SettingsManager.class);
final TransportPropertyManager transportPropertyManager =
@@ -72,6 +62,12 @@ public class PluginManagerImplTest extends BriarTestCase {
final TransportId duplexFailId = new TransportId("duplex1");
context.checking(new Expectations() {{
allowing(simplexPlugin).getId();
will(returnValue(simplexId));
allowing(simplexFailPlugin).getId();
will(returnValue(simplexFailId));
allowing(duplexPlugin).getId();
will(returnValue(duplexId));
// start()
// First simplex plugin
oneOf(pluginConfig).getSimplexFactories();
@@ -108,21 +104,16 @@ public class PluginManagerImplTest extends BriarTestCase {
oneOf(duplexFailFactory).createPlugin(with(any(
DuplexPluginCallback.class)));
will(returnValue(null)); // Failed to create a plugin
// Start listening for events
oneOf(eventBus).addListener(with(any(EventListener.class)));
// stop()
// Stop listening for events
oneOf(eventBus).removeListener(with(any(EventListener.class)));
// Stop the poller
oneOf(poller).stop();
// Stop the plugins
oneOf(simplexPlugin).stop();
oneOf(simplexFailPlugin).stop();
oneOf(duplexPlugin).stop();
}});
PluginManagerImpl p = new PluginManagerImpl(ioExecutor, eventBus,
pluginConfig, poller, connectionManager, connectionRegistry,
settingsManager, transportPropertyManager, uiCallback);
pluginConfig, connectionManager, settingsManager,
transportPropertyManager, uiCallback);
// Two plugins should be started and stopped
p.startService();
@@ -130,141 +121,4 @@ public class PluginManagerImplTest extends BriarTestCase {
context.assertIsSatisfied();
}
@Test
public void testConnectToNewContact() throws Exception {
Mockery context = new Mockery();
final Executor ioExecutor = new ImmediateExecutor();
final EventBus eventBus = context.mock(EventBus.class);
final PluginConfig pluginConfig = context.mock(PluginConfig.class);
final Poller poller = context.mock(Poller.class);
final ConnectionManager connectionManager =
context.mock(ConnectionManager.class);
final ConnectionRegistry connectionRegistry =
context.mock(ConnectionRegistry.class);
final SettingsManager settingsManager =
context.mock(SettingsManager.class);
final TransportPropertyManager transportPropertyManager =
context.mock(TransportPropertyManager.class);
final UiCallback uiCallback = context.mock(UiCallback.class);
final TransportConnectionWriter transportConnectionWriter =
context.mock(TransportConnectionWriter.class);
final DuplexTransportConnection duplexTransportConnection =
context.mock(DuplexTransportConnection.class);
final ContactId contactId = new ContactId(234);
// Two simplex plugins: one supports polling, the other doesn't
final SimplexPluginFactory simplexFactory =
context.mock(SimplexPluginFactory.class);
final SimplexPlugin simplexPlugin = context.mock(SimplexPlugin.class);
final TransportId simplexId = new TransportId("simplex");
final SimplexPluginFactory simplexFactory1 =
context.mock(SimplexPluginFactory.class, "simplexFactory1");
final SimplexPlugin simplexPlugin1 =
context.mock(SimplexPlugin.class, "simplexPlugin1");
final TransportId simplexId1 = new TransportId("simplex1");
// Two duplex plugins: one supports polling, the other doesn't
final DuplexPluginFactory duplexFactory =
context.mock(DuplexPluginFactory.class);
final DuplexPlugin duplexPlugin = context.mock(DuplexPlugin.class);
final TransportId duplexId = new TransportId("duplex");
final DuplexPluginFactory duplexFactory1 =
context.mock(DuplexPluginFactory.class, "duplexFactory1");
final DuplexPlugin duplexPlugin1 =
context.mock(DuplexPlugin.class, "duplexPlugin1");
final TransportId duplexId1 = new TransportId("duplex1");
context.checking(new Expectations() {{
// start()
// First simplex plugin
oneOf(pluginConfig).getSimplexFactories();
will(returnValue(Arrays.asList(simplexFactory, simplexFactory1)));
oneOf(simplexFactory).getId();
will(returnValue(simplexId));
oneOf(simplexFactory).createPlugin(with(any(
SimplexPluginCallback.class)));
will(returnValue(simplexPlugin)); // Created
oneOf(simplexPlugin).start();
will(returnValue(true)); // Started
// Second simplex plugin
oneOf(simplexFactory1).getId();
will(returnValue(simplexId1));
oneOf(simplexFactory1).createPlugin(with(any(
SimplexPluginCallback.class)));
will(returnValue(simplexPlugin1)); // Created
oneOf(simplexPlugin1).start();
will(returnValue(true)); // Started
// First duplex plugin
oneOf(pluginConfig).getDuplexFactories();
will(returnValue(Arrays.asList(duplexFactory, duplexFactory1)));
oneOf(duplexFactory).getId();
will(returnValue(duplexId));
oneOf(duplexFactory).createPlugin(with(any(
DuplexPluginCallback.class)));
will(returnValue(duplexPlugin)); // Created
oneOf(duplexPlugin).start();
will(returnValue(true)); // Started
// Second duplex plugin
oneOf(duplexFactory1).getId();
will(returnValue(duplexId1));
oneOf(duplexFactory1).createPlugin(with(any(
DuplexPluginCallback.class)));
will(returnValue(duplexPlugin1)); // Created
oneOf(duplexPlugin1).start();
will(returnValue(true)); // Started
// Start listening for events
oneOf(eventBus).addListener(with(any(EventListener.class)));
// eventOccurred()
// First simplex plugin
oneOf(simplexPlugin).shouldPoll();
will(returnValue(true));
oneOf(simplexPlugin).getId();
will(returnValue(simplexId));
oneOf(connectionRegistry).isConnected(contactId, simplexId);
will(returnValue(false));
oneOf(simplexPlugin).createWriter(contactId);
will(returnValue(transportConnectionWriter));
oneOf(connectionManager).manageOutgoingConnection(contactId,
simplexId, transportConnectionWriter);
// Second simplex plugin
oneOf(simplexPlugin1).shouldPoll();
will(returnValue(false));
// First duplex plugin
oneOf(duplexPlugin).shouldPoll();
will(returnValue(true));
oneOf(duplexPlugin).getId();
will(returnValue(duplexId));
oneOf(connectionRegistry).isConnected(contactId, duplexId);
will(returnValue(false));
oneOf(duplexPlugin).createConnection(contactId);
will(returnValue(duplexTransportConnection));
oneOf(connectionManager).manageOutgoingConnection(contactId,
duplexId, duplexTransportConnection);
// Second duplex plugin
oneOf(duplexPlugin1).shouldPoll();
will(returnValue(false));
// stop()
// Stop listening for events
oneOf(eventBus).removeListener(with(any(EventListener.class)));
// Stop the poller
oneOf(poller).stop();
// Stop the plugins
oneOf(simplexPlugin).stop();
oneOf(simplexPlugin1).stop();
oneOf(duplexPlugin).stop();
oneOf(duplexPlugin1).stop();
}});
PluginManagerImpl p = new PluginManagerImpl(ioExecutor, eventBus,
pluginConfig, poller, connectionManager, connectionRegistry,
settingsManager, transportPropertyManager, uiCallback);
p.startService();
p.eventOccurred(new ContactStatusChangedEvent(contactId, true));
p.stopService();
context.assertIsSatisfied();
}
}

View File

@@ -0,0 +1,354 @@
package org.briarproject.plugins;
import org.briarproject.BriarTestCase;
import org.briarproject.ImmediateExecutor;
import org.briarproject.RunAction;
import org.briarproject.api.TransportId;
import org.briarproject.api.contact.ContactId;
import org.briarproject.api.event.ConnectionClosedEvent;
import org.briarproject.api.event.ConnectionOpenedEvent;
import org.briarproject.api.event.ContactStatusChangedEvent;
import org.briarproject.api.event.TransportEnabledEvent;
import org.briarproject.api.plugins.ConnectionManager;
import org.briarproject.api.plugins.ConnectionRegistry;
import org.briarproject.api.plugins.Plugin;
import org.briarproject.api.plugins.PluginManager;
import org.briarproject.api.plugins.TransportConnectionWriter;
import org.briarproject.api.plugins.duplex.DuplexPlugin;
import org.briarproject.api.plugins.duplex.DuplexTransportConnection;
import org.briarproject.api.plugins.simplex.SimplexPlugin;
import org.briarproject.api.system.Clock;
import org.jmock.Expectations;
import org.jmock.Mockery;
import org.jmock.lib.legacy.ClassImposteriser;
import org.junit.Test;
import java.security.SecureRandom;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
public class PollerTest extends BriarTestCase {
private final ContactId contactId = new ContactId(234);
private final int pollingInterval = 60 * 1000;
private final long now = System.currentTimeMillis();
@Test
public void testConnectOnContactStatusChanged() throws Exception {
Mockery context = new Mockery();
context.setImposteriser(ClassImposteriser.INSTANCE);
final Executor ioExecutor = new ImmediateExecutor();
final ScheduledExecutorService scheduler =
context.mock(ScheduledExecutorService.class);
final ConnectionManager connectionManager =
context.mock(ConnectionManager.class);
final ConnectionRegistry connectionRegistry =
context.mock(ConnectionRegistry.class);
final PluginManager pluginManager = context.mock(PluginManager.class);
final SecureRandom random = context.mock(SecureRandom.class);
final Clock clock = context.mock(Clock.class);
// Two simplex plugins: one supports polling, the other doesn't
final SimplexPlugin simplexPlugin = context.mock(SimplexPlugin.class);
final SimplexPlugin simplexPlugin1 =
context.mock(SimplexPlugin.class, "simplexPlugin1");
final TransportId simplexId1 = new TransportId("simplex1");
final List<SimplexPlugin> simplexPlugins = Arrays.asList(simplexPlugin,
simplexPlugin1);
final TransportConnectionWriter simplexWriter =
context.mock(TransportConnectionWriter.class);
// Two duplex plugins: one supports polling, the other doesn't
final DuplexPlugin duplexPlugin = context.mock(DuplexPlugin.class);
final TransportId duplexId = new TransportId("duplex");
final DuplexPlugin duplexPlugin1 =
context.mock(DuplexPlugin.class, "duplexPlugin1");
final List<DuplexPlugin> duplexPlugins = Arrays.asList(duplexPlugin,
duplexPlugin1);
final DuplexTransportConnection duplexConnection =
context.mock(DuplexTransportConnection.class);
context.checking(new Expectations() {{
// Get the simplex plugins
oneOf(pluginManager).getSimplexPlugins();
will(returnValue(simplexPlugins));
// The first plugin doesn't support polling
oneOf(simplexPlugin).shouldPoll();
will(returnValue(false));
// The second plugin supports polling
oneOf(simplexPlugin1).shouldPoll();
will(returnValue(true));
// Check whether the contact is already connected
oneOf(simplexPlugin1).getId();
will(returnValue(simplexId1));
oneOf(connectionRegistry).isConnected(contactId, simplexId1);
will(returnValue(false));
// Connect to the contact
oneOf(simplexPlugin1).createWriter(contactId);
will(returnValue(simplexWriter));
// Pass the connection to the connection manager
oneOf(connectionManager).manageOutgoingConnection(contactId,
simplexId1, simplexWriter);
// Get the duplex plugins
oneOf(pluginManager).getDuplexPlugins();
will(returnValue(duplexPlugins));
// The first plugin supports polling
oneOf(duplexPlugin).shouldPoll();
will(returnValue(true));
// Check whether the contact is already connected
oneOf(duplexPlugin).getId();
will(returnValue(duplexId));
oneOf(connectionRegistry).isConnected(contactId, duplexId);
will(returnValue(false));
// Connect to the contact
oneOf(duplexPlugin).createConnection(contactId);
will(returnValue(duplexConnection));
// Pass the connection to the connection manager
oneOf(connectionManager).manageOutgoingConnection(contactId,
duplexId, duplexConnection);
// The second plugin doesn't support polling
oneOf(duplexPlugin1).shouldPoll();
will(returnValue(false));
}});
Poller p = new Poller(ioExecutor, scheduler, connectionManager,
connectionRegistry, pluginManager, random, clock);
p.eventOccurred(new ContactStatusChangedEvent(contactId, true));
context.assertIsSatisfied();
}
@Test
public void testRescheduleAndReconnectOnConnectionClosed()
throws Exception {
Mockery context = new Mockery();
context.setImposteriser(ClassImposteriser.INSTANCE);
final Executor ioExecutor = new ImmediateExecutor();
final ScheduledExecutorService scheduler =
context.mock(ScheduledExecutorService.class);
final ConnectionManager connectionManager =
context.mock(ConnectionManager.class);
final ConnectionRegistry connectionRegistry =
context.mock(ConnectionRegistry.class);
final PluginManager pluginManager = context.mock(PluginManager.class);
final SecureRandom random = context.mock(SecureRandom.class);
final Clock clock = context.mock(Clock.class);
final DuplexPlugin plugin = context.mock(DuplexPlugin.class);
final TransportId transportId = new TransportId("id");
final DuplexTransportConnection duplexConnection =
context.mock(DuplexTransportConnection.class);
context.checking(new Expectations() {{
allowing(plugin).getId();
will(returnValue(transportId));
// reschedule()
// Get the plugin
oneOf(pluginManager).getPlugin(transportId);
will(returnValue(plugin));
// The plugin supports polling
oneOf(plugin).shouldPoll();
will(returnValue(true));
// Get the plugin
oneOf(pluginManager).getPlugin(transportId);
will(returnValue(plugin));
// The plugin supports polling
oneOf(plugin).shouldPoll();
will(returnValue(true));
// Schedule the next poll
oneOf(plugin).getPollingInterval();
will(returnValue(pollingInterval));
oneOf(clock).currentTimeMillis();
will(returnValue(now));
oneOf(scheduler).schedule(with(any(Runnable.class)),
with((long) pollingInterval), with(MILLISECONDS));
// connectToContact()
// Check whether the contact is already connected
oneOf(connectionRegistry).isConnected(contactId, transportId);
will(returnValue(false));
// Connect to the contact
oneOf(plugin).createConnection(contactId);
will(returnValue(duplexConnection));
// Pass the connection to the connection manager
oneOf(connectionManager).manageOutgoingConnection(contactId,
transportId, duplexConnection);
}});
Poller p = new Poller(ioExecutor, scheduler, connectionManager,
connectionRegistry, pluginManager, random, clock);
p.eventOccurred(new ConnectionClosedEvent(contactId, transportId,
false));
context.assertIsSatisfied();
}
@Test
public void testRescheduleOnConnectionOpened() throws Exception {
Mockery context = new Mockery();
context.setImposteriser(ClassImposteriser.INSTANCE);
final Executor ioExecutor = new ImmediateExecutor();
final ScheduledExecutorService scheduler =
context.mock(ScheduledExecutorService.class);
final ConnectionManager connectionManager =
context.mock(ConnectionManager.class);
final ConnectionRegistry connectionRegistry =
context.mock(ConnectionRegistry.class);
final PluginManager pluginManager = context.mock(PluginManager.class);
final SecureRandom random = context.mock(SecureRandom.class);
final Clock clock = context.mock(Clock.class);
final DuplexPlugin plugin = context.mock(DuplexPlugin.class);
final TransportId transportId = new TransportId("id");
context.checking(new Expectations() {{
allowing(plugin).getId();
will(returnValue(transportId));
// Get the plugin
oneOf(pluginManager).getPlugin(transportId);
will(returnValue(plugin));
// The plugin supports polling
oneOf(plugin).shouldPoll();
will(returnValue(true));
// Schedule the next poll
oneOf(plugin).getPollingInterval();
will(returnValue(pollingInterval));
oneOf(clock).currentTimeMillis();
will(returnValue(now));
oneOf(scheduler).schedule(with(any(Runnable.class)),
with((long) pollingInterval), with(MILLISECONDS));
}});
Poller p = new Poller(ioExecutor, scheduler, connectionManager,
connectionRegistry, pluginManager, random, clock);
p.eventOccurred(new ConnectionOpenedEvent(contactId, transportId,
false));
context.assertIsSatisfied();
}
@Test
public void testRescheduleDoesNotReplaceEarlierTask() throws Exception {
Mockery context = new Mockery();
context.setImposteriser(ClassImposteriser.INSTANCE);
final Executor ioExecutor = new ImmediateExecutor();
final ScheduledExecutorService scheduler =
context.mock(ScheduledExecutorService.class);
final ConnectionManager connectionManager =
context.mock(ConnectionManager.class);
final ConnectionRegistry connectionRegistry =
context.mock(ConnectionRegistry.class);
final PluginManager pluginManager = context.mock(PluginManager.class);
final SecureRandom random = context.mock(SecureRandom.class);
final Clock clock = context.mock(Clock.class);
final DuplexPlugin plugin = context.mock(DuplexPlugin.class);
final TransportId transportId = new TransportId("id");
context.checking(new Expectations() {{
allowing(plugin).getId();
will(returnValue(transportId));
// First event
// Get the plugin
oneOf(pluginManager).getPlugin(transportId);
will(returnValue(plugin));
// The plugin supports polling
oneOf(plugin).shouldPoll();
will(returnValue(true));
// Schedule the next poll
oneOf(plugin).getPollingInterval();
will(returnValue(pollingInterval));
oneOf(clock).currentTimeMillis();
will(returnValue(now));
oneOf(scheduler).schedule(with(any(Runnable.class)),
with((long) pollingInterval), with(MILLISECONDS));
// Second event
// Get the plugin
oneOf(pluginManager).getPlugin(transportId);
will(returnValue(plugin));
// The plugin supports polling
oneOf(plugin).shouldPoll();
will(returnValue(true));
// Don't replace the previously scheduled task, due earlier
oneOf(plugin).getPollingInterval();
will(returnValue(pollingInterval));
oneOf(clock).currentTimeMillis();
will(returnValue(now + 1));
}});
Poller p = new Poller(ioExecutor, scheduler, connectionManager,
connectionRegistry, pluginManager, random, clock);
p.eventOccurred(new ConnectionOpenedEvent(contactId, transportId,
false));
p.eventOccurred(new ConnectionOpenedEvent(contactId, transportId,
false));
context.assertIsSatisfied();
}
@Test
public void testPollOnTransportEnabled() throws Exception {
Mockery context = new Mockery();
context.setImposteriser(ClassImposteriser.INSTANCE);
final Executor ioExecutor = new ImmediateExecutor();
final ScheduledExecutorService scheduler =
context.mock(ScheduledExecutorService.class);
final ConnectionManager connectionManager =
context.mock(ConnectionManager.class);
final ConnectionRegistry connectionRegistry =
context.mock(ConnectionRegistry.class);
final PluginManager pluginManager = context.mock(PluginManager.class);
final SecureRandom random = context.mock(SecureRandom.class);
final Clock clock = context.mock(Clock.class);
final Plugin plugin = context.mock(Plugin.class);
final TransportId transportId = new TransportId("id");
final List<ContactId> connected = Collections.singletonList(contactId);
context.checking(new Expectations() {{
allowing(plugin).getId();
will(returnValue(transportId));
// Get the plugin
oneOf(pluginManager).getPlugin(transportId);
will(returnValue(plugin));
// The plugin supports polling
oneOf(plugin).shouldPoll();
will(returnValue(true));
// Schedule a polling task immediately
oneOf(clock).currentTimeMillis();
will(returnValue(now));
oneOf(scheduler).schedule(with(any(Runnable.class)), with(0L),
with(MILLISECONDS));
will(new RunAction());
// Running the polling task schedules the next polling task
oneOf(plugin).getPollingInterval();
will(returnValue(pollingInterval));
oneOf(random).nextDouble();
will(returnValue(0.5));
oneOf(clock).currentTimeMillis();
will(returnValue(now));
oneOf(scheduler).schedule(with(any(Runnable.class)),
with((long) (pollingInterval * 0.5)), with(MILLISECONDS));
// Poll the plugin
oneOf(connectionRegistry).getConnectedContacts(transportId);
will(returnValue(connected));
oneOf(plugin).poll(connected);
}});
Poller p = new Poller(ioExecutor, scheduler, connectionManager,
connectionRegistry, pluginManager, random, clock);
p.eventOccurred(new TransportEnabledEvent(transportId));
context.assertIsSatisfied();
}
}