Merge branch '2016-2017-2018-removable-drive-reader-writer' into '1802-sync-via-removable-storage'

Create removable drive manager and reader/writer tasks

See merge request briar/briar!1458

bramble-android/src/main/java/org/briarproject/bramble/plugin/file/AndroidRemovableDrivePlugin.java

@@ -12,7 +12,7 @@ import java.io.OutputStream;
 
 import javax.annotation.concurrent.Immutable;
 
-import static org.briarproject.bramble.api.plugin.RemovableDriveConstants.PROP_URI;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.PROP_URI;
 import static org.briarproject.bramble.util.StringUtils.isNullOrEmpty;
 
 @Immutable

bramble-android/src/main/java/org/briarproject/bramble/plugin/file/AndroidRemovableDrivePluginFactory.java

@@ -13,7 +13,7 @@ import javax.annotation.concurrent.Immutable;
 import javax.inject.Inject;
 
 import static java.util.concurrent.TimeUnit.DAYS;
-import static org.briarproject.bramble.api.plugin.RemovableDriveConstants.ID;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
 
 @Immutable
 @NotNullByDefault

bramble-api/src/main/java/org/briarproject/bramble/api/Consumer.java

@@ -0,0 +1,9 @@
+package org.briarproject.bramble.api;
+
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+
+@NotNullByDefault
+public interface Consumer<T> {
+
+	void accept(T t);
+}

bramble-api/src/main/java/org/briarproject/bramble/api/plugin/file/FileConstants.java

@@ -1,4 +1,4 @@
-package org.briarproject.bramble.api.plugin;
+package org.briarproject.bramble.api.plugin.file;
 
 public interface FileConstants {
 

bramble-api/src/main/java/org/briarproject/bramble/api/plugin/file/RemovableDriveConstants.java

@@ -1,4 +1,6 @@
-package org.briarproject.bramble.api.plugin;
+package org.briarproject.bramble.api.plugin.file;
+
+import org.briarproject.bramble.api.plugin.TransportId;
 
 public interface RemovableDriveConstants {
 

bramble-api/src/main/java/org/briarproject/bramble/api/plugin/file/RemovableDriveManager.java

@@ -0,0 +1,41 @@
+package org.briarproject.bramble.api.plugin.file;
+
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import javax.annotation.Nullable;
+
+@NotNullByDefault
+public interface RemovableDriveManager {
+
+	/**
+	 * Returns the currently running reader task for the given contact,
+	 * or null if no task is running.
+	 */
+	@Nullable
+	RemovableDriveTask getCurrentReaderTask(ContactId c);
+
+	/**
+	 * Returns the currently running writer task for the given contact,
+	 * or null if no task is running.
+	 */
+	@Nullable
+	RemovableDriveTask getCurrentWriterTask(ContactId c);
+
+	/**
+	 * Starts and returns a reader task for the given contact, reading from
+	 * a stream described by the given transport properties. If a reader task
+	 * for the contact is already running, it will be returned and the
+	 * transport properties will be ignored.
+	 */
+	RemovableDriveTask startReaderTask(ContactId c, TransportProperties p);
+
+	/**
+	 * Starts and returns a writer task for the given contact, writing to
+	 * a stream described by the given transport properties. If a writer task
+	 * for the contact is already running, it will be returned and the
+	 * transport properties will be ignored.
+	 */
+	RemovableDriveTask startWriterTask(ContactId c, TransportProperties p);
+}

bramble-api/src/main/java/org/briarproject/bramble/api/plugin/file/RemovableDriveTask.java

@@ -0,0 +1,65 @@
+package org.briarproject.bramble.api.plugin.file;
+
+import org.briarproject.bramble.api.Consumer;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+@NotNullByDefault
+public interface RemovableDriveTask extends Runnable {
+
+	/**
+	 * Returns the {@link TransportProperties} that were used for creating
+	 * this task.
+	 */
+	TransportProperties getTransportProperties();
+
+	/**
+	 * Adds an observer to the task. The observer will be notified of state
+	 * changes on the event thread. If the task has already finished, the
+	 * observer will be notified of its final state.
+	 */
+	void addObserver(Consumer<State> observer);
+
+	/**
+	 * Removes an observer from the task.
+	 */
+	void removeObserver(Consumer<State> observer);
+
+	class State {
+
+		private final long done, total;
+		private final boolean finished, success;
+
+		public State(long done, long total, boolean finished, boolean success) {
+			this.done = done;
+			this.total = total;
+			this.finished = finished;
+			this.success = success;
+		}
+
+		/**
+		 * Returns the total length in bytes of the messages read or written
+		 * so far.
+		 */
+		public long getDone() {
+			return done;
+		}
+
+		/**
+		 * Returns the total length in bytes of the messages that will have
+		 * been read or written when the task is complete, or zero if the
+		 * total is unknown.
+		 */
+		public long getTotal() {
+			return total;
+		}
+
+		public boolean isFinished() {
+			return finished;
+		}
+
+		public boolean isSuccess() {
+			return success;
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/AbstractRemovableDrivePlugin.java

@@ -20,7 +20,7 @@ import javax.annotation.concurrent.Immutable;
 import static java.util.logging.Level.WARNING;
 import static java.util.logging.Logger.getLogger;
 import static org.briarproject.bramble.api.plugin.Plugin.State.ACTIVE;
-import static org.briarproject.bramble.api.plugin.RemovableDriveConstants.ID;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
 import static org.briarproject.bramble.util.LogUtils.logException;
 
 @Immutable

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/FilePlugin.java

@@ -15,8 +15,8 @@ import java.util.logging.Logger;
 
 import static java.util.logging.Level.WARNING;
 import static java.util.logging.Logger.getLogger;
-import static org.briarproject.bramble.api.plugin.FileConstants.PROP_PATH;
 import static org.briarproject.bramble.api.plugin.Plugin.State.ACTIVE;
+import static org.briarproject.bramble.api.plugin.file.FileConstants.PROP_PATH;
 import static org.briarproject.bramble.util.LogUtils.logException;
 import static org.briarproject.bramble.util.StringUtils.isNullOrEmpty;
 

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveManagerImpl.java

@@ -0,0 +1,83 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.lifecycle.IoExecutor;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveManager;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+import javax.inject.Inject;
+
+@ThreadSafe
+@NotNullByDefault
+class RemovableDriveManagerImpl
+		implements RemovableDriveManager, RemovableDriveTaskRegistry {
+
+	private final Executor ioExecutor;
+	private final RemovableDriveTaskFactory taskFactory;
+	private final ConcurrentHashMap<ContactId, RemovableDriveTask>
+			readers = new ConcurrentHashMap<>();
+	private final ConcurrentHashMap<ContactId, RemovableDriveTask>
+			writers = new ConcurrentHashMap<>();
+
+	@Inject
+	RemovableDriveManagerImpl(@IoExecutor Executor ioExecutor,
+			RemovableDriveTaskFactory taskFactory) {
+		this.ioExecutor = ioExecutor;
+		this.taskFactory = taskFactory;
+	}
+
+	@Nullable
+	@Override
+	public RemovableDriveTask getCurrentReaderTask(ContactId c) {
+		return readers.get(c);
+	}
+
+	@Nullable
+	@Override
+	public RemovableDriveTask getCurrentWriterTask(ContactId c) {
+		return writers.get(c);
+	}
+
+	@Override
+	public RemovableDriveTask startReaderTask(ContactId c,
+			TransportProperties p) {
+		RemovableDriveTask task = taskFactory.createReader(this, c, p);
+		RemovableDriveTask old = readers.putIfAbsent(c, task);
+		if (old == null) {
+			ioExecutor.execute(task);
+			return task;
+		} else {
+			return old;
+		}
+	}
+
+	@Override
+	public RemovableDriveTask startWriterTask(ContactId c,
+			TransportProperties p) {
+		RemovableDriveTask task = taskFactory.createWriter(this, c, p);
+		RemovableDriveTask old = writers.putIfAbsent(c, task);
+		if (old == null) {
+			ioExecutor.execute(task);
+			return task;
+		} else {
+			return old;
+		}
+	}
+
+	@Override
+	public void removeReader(ContactId c, RemovableDriveTask task) {
+		readers.remove(c, task);
+	}
+
+	@Override
+	public void removeWriter(ContactId c, RemovableDriveTask task) {
+		writers.remove(c, task);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveModule.java

@@ -0,0 +1,25 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.plugin.file.RemovableDriveManager;
+
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+
+@Module
+public class RemovableDriveModule {
+
+	@Provides
+	@Singleton
+	RemovableDriveManager provideRemovableDriveManager(
+			RemovableDriveManagerImpl removableDriveManager) {
+		return removableDriveManager;
+	}
+
+	@Provides
+	RemovableDriveTaskFactory provideTaskFactory(
+			RemovableDriveTaskFactoryImpl taskFactory) {
+		return taskFactory;
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDrivePlugin.java

@@ -11,7 +11,7 @@ import java.io.OutputStream;
 
 import javax.annotation.concurrent.Immutable;
 
-import static org.briarproject.bramble.api.plugin.RemovableDriveConstants.PROP_PATH;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.PROP_PATH;
 import static org.briarproject.bramble.util.StringUtils.isNullOrEmpty;
 
 @Immutable

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDrivePluginFactory.java

@@ -11,7 +11,7 @@ import javax.annotation.concurrent.Immutable;
 import javax.inject.Inject;
 
 import static java.util.concurrent.TimeUnit.DAYS;
-import static org.briarproject.bramble.api.plugin.RemovableDriveConstants.ID;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
 
 @Immutable
 @NotNullByDefault

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveReaderTask.java

@@ -0,0 +1,88 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.event.Event;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.event.EventListener;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.TransportConnectionReader;
+import org.briarproject.bramble.api.properties.TransportProperties;
+import org.briarproject.bramble.api.sync.event.MessageAddedEvent;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.Executor;
+import java.util.logging.Logger;
+
+import static java.util.logging.Logger.getLogger;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
+
+@NotNullByDefault
+class RemovableDriveReaderTask extends RemovableDriveTaskImpl
+		implements EventListener {
+
+	private final static Logger LOG =
+			getLogger(RemovableDriveReaderTask.class.getName());
+
+	RemovableDriveReaderTask(
+			Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus,
+			RemovableDriveTaskRegistry registry,
+			ContactId contactId,
+			TransportProperties transportProperties) {
+		super(eventExecutor, pluginManager, connectionManager, eventBus,
+				registry, contactId, transportProperties);
+	}
+
+	@Override
+	public void run() {
+		TransportConnectionReader r =
+				getPlugin().createReader(transportProperties);
+		if (r == null) {
+			LOG.warning("Failed to create reader");
+			registry.removeReader(contactId, this);
+			setSuccess(false);
+			return;
+		}
+		eventBus.addListener(this);
+		connectionManager.manageIncomingConnection(ID, new DecoratedReader(r));
+	}
+
+	@Override
+	public void eventOccurred(Event e) {
+		if (e instanceof MessageAddedEvent) {
+			MessageAddedEvent m = (MessageAddedEvent) e;
+			if (contactId.equals(m.getContactId())) {
+				LOG.info("Message received");
+				addDone(m.getMessage().getRawLength());
+			}
+		}
+	}
+
+	private class DecoratedReader implements TransportConnectionReader {
+
+		private final TransportConnectionReader delegate;
+
+		private DecoratedReader(TransportConnectionReader delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public InputStream getInputStream() throws IOException {
+			return delegate.getInputStream();
+		}
+
+		@Override
+		public void dispose(boolean exception, boolean recognised)
+				throws IOException {
+			delegate.dispose(exception, recognised);
+			registry.removeReader(contactId, RemovableDriveReaderTask.this);
+			eventBus.removeListener(RemovableDriveReaderTask.this);
+			setSuccess(!exception && recognised);
+		}
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskFactory.java

@@ -0,0 +1,16 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+@NotNullByDefault
+interface RemovableDriveTaskFactory {
+
+	RemovableDriveTask createReader(RemovableDriveTaskRegistry registry,
+			ContactId c, TransportProperties p);
+
+	RemovableDriveTask createWriter(RemovableDriveTaskRegistry registry,
+			ContactId c, TransportProperties p);
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskFactoryImpl.java

@@ -0,0 +1,55 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.event.EventExecutor;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.util.concurrent.Executor;
+
+import javax.annotation.concurrent.Immutable;
+import javax.inject.Inject;
+
+@Immutable
+@NotNullByDefault
+class RemovableDriveTaskFactoryImpl implements RemovableDriveTaskFactory {
+
+	private final DatabaseComponent db;
+	private final Executor eventExecutor;
+	private final PluginManager pluginManager;
+	private final ConnectionManager connectionManager;
+	private final EventBus eventBus;
+
+	@Inject
+	RemovableDriveTaskFactoryImpl(
+			DatabaseComponent db,
+			@EventExecutor Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus) {
+		this.db = db;
+		this.eventExecutor = eventExecutor;
+		this.pluginManager = pluginManager;
+		this.connectionManager = connectionManager;
+		this.eventBus = eventBus;
+	}
+
+	@Override
+	public RemovableDriveTask createReader(RemovableDriveTaskRegistry registry,
+			ContactId c, TransportProperties p) {
+		return new RemovableDriveReaderTask(eventExecutor, pluginManager,
+				connectionManager, eventBus, registry, c, p);
+	}
+
+	@Override
+	public RemovableDriveTask createWriter(RemovableDriveTaskRegistry registry,
+			ContactId c, TransportProperties p) {
+		return new RemovableDriveWriterTask(db, eventExecutor, pluginManager,
+				connectionManager, eventBus, registry, c, p);
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskImpl.java

@@ -0,0 +1,120 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.Consumer;
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
+import org.briarproject.bramble.api.properties.TransportProperties;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import javax.annotation.concurrent.GuardedBy;
+import javax.annotation.concurrent.ThreadSafe;
+
+import static java.lang.Math.min;
+import static org.briarproject.bramble.api.nullsafety.NullSafety.requireNonNull;
+import static org.briarproject.bramble.api.plugin.file.RemovableDriveConstants.ID;
+
+@ThreadSafe
+@NotNullByDefault
+abstract class RemovableDriveTaskImpl implements RemovableDriveTask {
+
+	private final Executor eventExecutor;
+	private final PluginManager pluginManager;
+	final ConnectionManager connectionManager;
+	final EventBus eventBus;
+	final RemovableDriveTaskRegistry registry;
+	final ContactId contactId;
+	final TransportProperties transportProperties;
+
+	private final Object lock = new Object();
+	@GuardedBy("lock")
+	private final List<Consumer<State>> observers = new ArrayList<>();
+	@GuardedBy("lock")
+	private State state = new State(0, 0, false, false);
+
+	RemovableDriveTaskImpl(
+			Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus,
+			RemovableDriveTaskRegistry registry,
+			ContactId contactId,
+			TransportProperties transportProperties) {
+		this.eventExecutor = eventExecutor;
+		this.pluginManager = pluginManager;
+		this.connectionManager = connectionManager;
+		this.eventBus = eventBus;
+		this.registry = registry;
+		this.contactId = contactId;
+		this.transportProperties = transportProperties;
+	}
+
+	@Override
+	public TransportProperties getTransportProperties() {
+		return transportProperties;
+	}
+
+	@Override
+	public void addObserver(Consumer<State> o) {
+		State state;
+		synchronized (lock) {
+			observers.add(o);
+			state = this.state;
+		}
+		if (state.isFinished()) {
+			eventExecutor.execute(() -> o.accept(state));
+		}
+	}
+
+	@Override
+	public void removeObserver(Consumer<State> o) {
+		synchronized (lock) {
+			observers.remove(o);
+		}
+	}
+
+	SimplexPlugin getPlugin() {
+		return (SimplexPlugin) requireNonNull(pluginManager.getPlugin(ID));
+	}
+
+	void setTotal(long total) {
+		synchronized (lock) {
+			state = new State(state.getDone(), total, state.isFinished(),
+					state.isSuccess());
+			notifyObservers();
+		}
+	}
+
+	void addDone(long done) {
+		synchronized (lock) {
+			// Done and total come from different sources; make them consistent
+			done = min(state.getDone() + done, state.getTotal());
+			state = new State(done, state.getTotal(), state.isFinished(),
+					state.isSuccess());
+		}
+		notifyObservers();
+	}
+
+	void setSuccess(boolean success) {
+		synchronized (lock) {
+			state = new State(state.getDone(), state.getTotal(), true, success);
+		}
+		notifyObservers();
+	}
+
+	@GuardedBy("lock")
+	private void notifyObservers() {
+		List<Consumer<State>> observers = new ArrayList<>(this.observers);
+		State state = this.state;
+		eventExecutor.execute(() -> {
+			for (Consumer<State> o : observers) o.accept(state);
+		});
+	}
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveTaskRegistry.java

@@ -0,0 +1,13 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+
+@NotNullByDefault
+interface RemovableDriveTaskRegistry {
+
+	void removeReader(ContactId c, RemovableDriveTask task);
+
+	void removeWriter(ContactId c, RemovableDriveTask task);
+}

bramble-core/src/main/java/org/briarproject/bramble/plugin/file/RemovableDriveWriterTask.java

@@ -0,0 +1,120 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.connection.ConnectionManager;
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.db.DatabaseComponent;
+import org.briarproject.bramble.api.db.DbException;
+import org.briarproject.bramble.api.event.Event;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.event.EventListener;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginManager;
+import org.briarproject.bramble.api.plugin.TransportConnectionWriter;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPlugin;
+import org.briarproject.bramble.api.properties.TransportProperties;
+import org.briarproject.bramble.api.sync.event.MessagesSentEvent;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.Executor;
+import java.util.logging.Logger;
+
+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.api.plugin.file.RemovableDriveConstants.ID;
+import static org.briarproject.bramble.util.LogUtils.logException;
+
+@NotNullByDefault
+class RemovableDriveWriterTask extends RemovableDriveTaskImpl
+		implements EventListener {
+
+	private static final Logger LOG =
+			getLogger(RemovableDriveWriterTask.class.getName());
+
+	private final DatabaseComponent db;
+
+	RemovableDriveWriterTask(
+			DatabaseComponent db,
+			Executor eventExecutor,
+			PluginManager pluginManager,
+			ConnectionManager connectionManager,
+			EventBus eventBus,
+			RemovableDriveTaskRegistry registry,
+			ContactId contactId,
+			TransportProperties transportProperties) {
+		super(eventExecutor, pluginManager, connectionManager, eventBus,
+				registry, contactId, transportProperties);
+		this.db = db;
+	}
+
+	@Override
+	public void run() {
+		SimplexPlugin plugin = getPlugin();
+		TransportConnectionWriter w = plugin.createWriter(transportProperties);
+		if (w == null) {
+			LOG.warning("Failed to create writer");
+			registry.removeWriter(contactId, this);
+			setSuccess(false);
+			return;
+		}
+		int maxLatency = plugin.getMaxLatency();
+		try {
+			setTotal(db.transactionWithResult(true, txn ->
+					db.getMessageBytesToSend(txn, contactId, maxLatency)));
+		} catch (DbException e) {
+			logException(LOG, WARNING, e);
+			registry.removeWriter(contactId, this);
+			setSuccess(false);
+			return;
+		}
+		eventBus.addListener(this);
+		connectionManager.manageOutgoingConnection(contactId, ID,
+				new DecoratedWriter(w));
+	}
+
+	@Override
+	public void eventOccurred(Event e) {
+		if (e instanceof MessagesSentEvent) {
+			MessagesSentEvent m = (MessagesSentEvent) e;
+			if (contactId.equals(m.getContactId())) {
+				if (LOG.isLoggable(INFO)) {
+					LOG.info(m.getMessageIds().size() + " messages sent");
+				}
+				addDone(m.getTotalLength());
+			}
+		}
+	}
+
+	private class DecoratedWriter implements TransportConnectionWriter {
+
+		private final TransportConnectionWriter delegate;
+
+		private DecoratedWriter(TransportConnectionWriter delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public int getMaxLatency() {
+			return delegate.getMaxLatency();
+		}
+
+		@Override
+		public int getMaxIdleTime() {
+			return delegate.getMaxIdleTime();
+		}
+
+		@Override
+		public OutputStream getOutputStream() throws IOException {
+			return delegate.getOutputStream();
+		}
+
+		@Override
+		public void dispose(boolean exception) throws IOException {
+			delegate.dispose(exception);
+			registry.removeWriter(contactId, RemovableDriveWriterTask.this);
+			eventBus.removeListener(RemovableDriveWriterTask.this);
+			setSuccess(!exception);
+		}
+	}
+}

bramble-core/src/test/java/org/briarproject/bramble/plugin/file/RemovableDriveIntegrationTest.java

@@ -0,0 +1,168 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.contact.ContactId;
+import org.briarproject.bramble.api.contact.ContactManager;
+import org.briarproject.bramble.api.crypto.SecretKey;
+import org.briarproject.bramble.api.event.Event;
+import org.briarproject.bramble.api.event.EventListener;
+import org.briarproject.bramble.api.identity.Author;
+import org.briarproject.bramble.api.identity.Identity;
+import org.briarproject.bramble.api.identity.IdentityManager;
+import org.briarproject.bramble.api.lifecycle.LifecycleManager;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveTask;
+import org.briarproject.bramble.api.properties.TransportProperties;
+import org.briarproject.bramble.api.sync.event.MessageStateChangedEvent;
+import org.briarproject.bramble.test.BrambleTestCase;
+import org.briarproject.bramble.test.TestDatabaseConfigModule;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.concurrent.CountDownLatch;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.briarproject.bramble.api.plugin.file.FileConstants.PROP_PATH;
+import static org.briarproject.bramble.api.sync.validation.MessageState.DELIVERED;
+import static org.briarproject.bramble.test.TestUtils.deleteTestDirectory;
+import static org.briarproject.bramble.test.TestUtils.getSecretKey;
+import static org.briarproject.bramble.test.TestUtils.getTestDirectory;
+import static org.junit.Assert.assertTrue;
+
+public class RemovableDriveIntegrationTest extends BrambleTestCase {
+
+	private static final int TIMEOUT_MS = 5_000;
+
+	private final File testDir = getTestDirectory();
+	private final File aliceDir = new File(testDir, "alice");
+	private final File bobDir = new File(testDir, "bob");
+
+	private final SecretKey rootKey = getSecretKey();
+	private final long timestamp = System.currentTimeMillis();
+
+	private RemovableDriveIntegrationTestComponent alice, bob;
+
+	@Before
+	public void setUp() {
+		assertTrue(testDir.mkdirs());
+		alice = DaggerRemovableDriveIntegrationTestComponent.builder()
+				.testDatabaseConfigModule(
+						new TestDatabaseConfigModule(aliceDir)).build();
+		RemovableDriveIntegrationTestComponent.Helper
+				.injectEagerSingletons(alice);
+		bob = DaggerRemovableDriveIntegrationTestComponent.builder()
+				.testDatabaseConfigModule(
+						new TestDatabaseConfigModule(bobDir)).build();
+		RemovableDriveIntegrationTestComponent.Helper
+				.injectEagerSingletons(bob);
+	}
+
+	@Test
+	public void testWriteAndRead() throws Exception {
+		// Create the identities
+		Identity aliceIdentity =
+				alice.getIdentityManager().createIdentity("Alice");
+		Identity bobIdentity = bob.getIdentityManager().createIdentity("Bob");
+		// Set up the devices and get the contact IDs
+		ContactId bobId = setUp(alice, aliceIdentity,
+				bobIdentity.getLocalAuthor(), true);
+		ContactId aliceId = setUp(bob, bobIdentity,
+				aliceIdentity.getLocalAuthor(), false);
+		// Sync Alice's client versions and transport properties
+		read(bob, aliceId, write(alice, bobId), 2);
+		// Sync Bob's client versions and transport properties
+		read(alice, bobId, write(bob, aliceId), 2);
+	}
+
+	private ContactId setUp(RemovableDriveIntegrationTestComponent device,
+			Identity local, Author remote, boolean alice) throws Exception {
+		// Add an identity for the user
+		IdentityManager identityManager = device.getIdentityManager();
+		identityManager.registerIdentity(local);
+		// Start the lifecycle manager
+		LifecycleManager lifecycleManager = device.getLifecycleManager();
+		lifecycleManager.startServices(getSecretKey());
+		lifecycleManager.waitForStartup();
+		// Add the other user as a contact
+		ContactManager contactManager = device.getContactManager();
+		return contactManager.addContact(remote, local.getId(), rootKey,
+				timestamp, alice, true, true);
+	}
+
+	@SuppressWarnings("SameParameterValue")
+	private void read(RemovableDriveIntegrationTestComponent device,
+			ContactId contactId, File file, int deliveries) throws Exception {
+		// Listen for message deliveries
+		MessageDeliveryListener listener =
+				new MessageDeliveryListener(deliveries);
+		device.getEventBus().addListener(listener);
+		// Read the incoming stream
+		TransportProperties p = new TransportProperties();
+		p.put(PROP_PATH, file.getAbsolutePath());
+		RemovableDriveTask reader = device.getRemovableDriveManager()
+				.startReaderTask(contactId, p);
+		CountDownLatch disposedLatch = new CountDownLatch(1);
+		reader.addObserver(state -> {
+			if (state.isFinished()) disposedLatch.countDown();
+		});
+		// Wait for the messages to be delivered
+		assertTrue(listener.delivered.await(TIMEOUT_MS, MILLISECONDS));
+		// Clean up the listener
+		device.getEventBus().removeListener(listener);
+		// Wait for the reader to be disposed
+		disposedLatch.await(TIMEOUT_MS, MILLISECONDS);
+	}
+
+	private File write(RemovableDriveIntegrationTestComponent device,
+			ContactId contactId) throws Exception {
+		// Write the outgoing stream to a file
+		File file = File.createTempFile("sync", ".tmp", testDir);
+		TransportProperties p = new TransportProperties();
+		p.put(PROP_PATH, file.getAbsolutePath());
+		RemovableDriveTask writer = device.getRemovableDriveManager()
+				.startWriterTask(contactId, p);
+		CountDownLatch disposedLatch = new CountDownLatch(1);
+		writer.addObserver(state -> {
+			if (state.isFinished()) disposedLatch.countDown();
+		});
+		// Wait for the writer to be disposed
+		disposedLatch.await(TIMEOUT_MS, MILLISECONDS);
+		// Return the file containing the stream
+		return file;
+	}
+
+	private void tearDown(RemovableDriveIntegrationTestComponent device)
+			throws Exception {
+		// Stop the lifecycle manager
+		LifecycleManager lifecycleManager = device.getLifecycleManager();
+		lifecycleManager.stopServices();
+		lifecycleManager.waitForShutdown();
+	}
+
+	@After
+	public void tearDown() throws Exception {
+		// Tear down the devices
+		tearDown(alice);
+		tearDown(bob);
+		deleteTestDirectory(testDir);
+	}
+
+	@NotNullByDefault
+	private static class MessageDeliveryListener implements EventListener {
+
+		private final CountDownLatch delivered;
+
+		private MessageDeliveryListener(int deliveries) {
+			delivered = new CountDownLatch(deliveries);
+		}
+
+		@Override
+		public void eventOccurred(Event e) {
+			if (e instanceof MessageStateChangedEvent) {
+				MessageStateChangedEvent m = (MessageStateChangedEvent) e;
+				if (m.getState().equals(DELIVERED)) delivered.countDown();
+			}
+		}
+	}
+}

bramble-core/src/test/java/org/briarproject/bramble/plugin/file/RemovableDriveIntegrationTestComponent.java

@@ -0,0 +1,53 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.BrambleCoreEagerSingletons;
+import org.briarproject.bramble.BrambleCoreModule;
+import org.briarproject.bramble.api.contact.ContactManager;
+import org.briarproject.bramble.api.event.EventBus;
+import org.briarproject.bramble.api.identity.IdentityManager;
+import org.briarproject.bramble.api.lifecycle.LifecycleManager;
+import org.briarproject.bramble.api.plugin.file.RemovableDriveManager;
+import org.briarproject.bramble.battery.DefaultBatteryManagerModule;
+import org.briarproject.bramble.event.DefaultEventExecutorModule;
+import org.briarproject.bramble.system.DefaultWakefulIoExecutorModule;
+import org.briarproject.bramble.system.TimeTravelModule;
+import org.briarproject.bramble.test.TestDatabaseConfigModule;
+import org.briarproject.bramble.test.TestSecureRandomModule;
+
+import javax.inject.Singleton;
+
+import dagger.Component;
+
+@Singleton
+@Component(modules = {
+		BrambleCoreModule.class,
+		DefaultBatteryManagerModule.class,
+		DefaultEventExecutorModule.class,
+		DefaultWakefulIoExecutorModule.class,
+		TestDatabaseConfigModule.class,
+		RemovableDriveIntegrationTestModule.class,
+		RemovableDriveModule.class,
+		TestSecureRandomModule.class,
+		TimeTravelModule.class
+})
+interface RemovableDriveIntegrationTestComponent
+		extends BrambleCoreEagerSingletons {
+
+	ContactManager getContactManager();
+
+	EventBus getEventBus();
+
+	IdentityManager getIdentityManager();
+
+	LifecycleManager getLifecycleManager();
+
+	RemovableDriveManager getRemovableDriveManager();
+
+	class Helper {
+
+		public static void injectEagerSingletons(
+				RemovableDriveIntegrationTestComponent c) {
+			BrambleCoreEagerSingletons.Helper.injectEagerSingletons(c);
+		}
+	}
+}

bramble-core/src/test/java/org/briarproject/bramble/plugin/file/RemovableDriveIntegrationTestModule.java

@@ -0,0 +1,81 @@
+package org.briarproject.bramble.plugin.file;
+
+import org.briarproject.bramble.api.FeatureFlags;
+import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.PluginConfig;
+import org.briarproject.bramble.api.plugin.TransportId;
+import org.briarproject.bramble.api.plugin.duplex.DuplexPluginFactory;
+import org.briarproject.bramble.api.plugin.simplex.SimplexPluginFactory;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import javax.inject.Singleton;
+
+import dagger.Module;
+import dagger.Provides;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.singletonList;
+
+@Module
+class RemovableDriveIntegrationTestModule {
+
+	@Provides
+	@Singleton
+	PluginConfig providePluginConfig(RemovableDrivePluginFactory drive) {
+		@NotNullByDefault
+		PluginConfig pluginConfig = new PluginConfig() {
+
+			@Override
+			public Collection<DuplexPluginFactory> getDuplexFactories() {
+				return emptyList();
+			}
+
+			@Override
+			public Collection<SimplexPluginFactory> getSimplexFactories() {
+				return singletonList(drive);
+			}
+
+			@Override
+			public boolean shouldPoll() {
+				return false;
+			}
+
+			@Override
+			public Map<TransportId, List<TransportId>> getTransportPreferences() {
+				return emptyMap();
+			}
+
+		};
+		return pluginConfig;
+	}
+
+	@Provides
+	FeatureFlags provideFeatureFlags() {
+		return new FeatureFlags() {
+
+			@Override
+			public boolean shouldEnableImageAttachments() {
+				return true;
+			}
+
+			@Override
+			public boolean shouldEnableProfilePictures() {
+				return true;
+			}
+
+			@Override
+			public boolean shouldEnableDisappearingMessages() {
+				return true;
+			}
+
+			@Override
+			public boolean shouldEnableConnectViaBluetooth() {
+				return true;
+			}
+		};
+	}
+}

briar-android/src/main/java/org/briarproject/briar/android/AndroidComponent.java

@@ -28,6 +28,7 @@ import org.briarproject.bramble.api.system.AndroidExecutor;
 import org.briarproject.bramble.api.system.AndroidWakeLockManager;
 import org.briarproject.bramble.api.system.Clock;
 import org.briarproject.bramble.api.system.LocationUtils;
+import org.briarproject.bramble.plugin.file.RemovableDriveModule;
 import org.briarproject.bramble.plugin.tor.CircumventionProvider;
 import org.briarproject.bramble.system.ClockModule;
 import org.briarproject.briar.BriarCoreEagerSingletons;
@@ -83,7 +84,8 @@ import dagger.Component;
 		AppModule.class,
 		AttachmentModule.class,
 		ClockModule.class,
-		MediaModule.class
+		MediaModule.class,
+		RemovableDriveModule.class
 })
 public interface AndroidComponent
 		extends BrambleCoreEagerSingletons, BrambleAndroidEagerSingletons,