Don't create files for upload while directly connected to contact.

bramble-core/src/main/java/org/briarproject/bramble/mailbox/MailboxUploadWorker.java

@@ -1,6 +1,7 @@
 package org.briarproject.bramble.mailbox;
 
 import org.briarproject.bramble.api.Cancellable;
+import org.briarproject.bramble.api.connection.ConnectionRegistry;
 import org.briarproject.bramble.api.contact.ContactId;
 import org.briarproject.bramble.api.db.DatabaseComponent;
 import org.briarproject.bramble.api.db.DbException;
@@ -12,6 +13,8 @@ import org.briarproject.bramble.api.lifecycle.IoExecutor;
 import org.briarproject.bramble.api.mailbox.MailboxFolderId;
 import org.briarproject.bramble.api.mailbox.MailboxProperties;
 import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
+import org.briarproject.bramble.api.plugin.event.ContactConnectedEvent;
+import org.briarproject.bramble.api.plugin.event.ContactDisconnectedEvent;
 import org.briarproject.bramble.api.sync.MessageId;
 import org.briarproject.bramble.api.sync.OutgoingSessionRecord;
 import org.briarproject.bramble.api.sync.event.GroupVisibilityUpdatedEvent;
@@ -32,6 +35,7 @@ import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
 
+import static java.lang.Boolean.TRUE;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.logging.Level.INFO;
@@ -58,9 +62,16 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 	 * <p>
 	 * If there's no data to send, the worker listens for events indicating
 	 * that new data may be ready to send.
+	 * <p>
+	 * Whenever we're directly connected to the contact, the worker doesn't
+	 * check for data to send or start connectivity checks until the contact
+	 * disconnects. However, if the worker has already started writing and
+	 * uploading a file when the contact connects, the worker will finish the
+	 * upload.
 	 */
 	private enum State {
 		CREATED,
+		CONNECTED_TO_CONTACT,
 		CHECKING_FOR_DATA,
 		WAITING_FOR_DATA,
 		CONNECTIVITY_CHECK,
@@ -95,6 +106,7 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 	private final Clock clock;
 	private final TaskScheduler taskScheduler;
 	private final EventBus eventBus;
+	private final ConnectionRegistry connectionRegistry;
 	private final ConnectivityChecker connectivityChecker;
 	private final MailboxApiCaller mailboxApiCaller;
 	private final MailboxApi mailboxApi;
@@ -121,6 +133,7 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 			Clock clock,
 			TaskScheduler taskScheduler,
 			EventBus eventBus,
+			ConnectionRegistry connectionRegistry,
 			ConnectivityChecker connectivityChecker,
 			MailboxApiCaller mailboxApiCaller,
 			MailboxApi mailboxApi,
@@ -133,6 +146,7 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 		this.clock = clock;
 		this.taskScheduler = taskScheduler;
 		this.eventBus = eventBus;
+		this.connectionRegistry = connectionRegistry;
 		this.connectivityChecker = connectivityChecker;
 		this.mailboxApiCaller = mailboxApiCaller;
 		this.mailboxApi = mailboxApi;
@@ -182,6 +196,12 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 		synchronized (lock) {
 			checkTask = null;
 			if (state != State.CHECKING_FOR_DATA) return;
+			// Check whether we're directly connected to the contact. Calling
+			// this while holding the lock isn't ideal, but it avoids races
+			if (connectionRegistry.isConnected(contactId)) {
+				state = State.CONNECTED_TO_CONTACT;
+				return;
+			}
 		}
 		LOG.info("Checking for data to send");
 		try {
@@ -364,8 +384,11 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 				onDataToSend();
 			}
 		} else if (e instanceof MessageSharedEvent) {
-			LOG.info("Message shared");
-			onDataToSend();
+			MessageSharedEvent m = (MessageSharedEvent) e;
+			if (m.getGroupVisibility().get(contactId) == TRUE) {
+				LOG.info("Message shared");
+				onDataToSend();
+			}
 		} else if (e instanceof GroupVisibilityUpdatedEvent) {
 			GroupVisibilityUpdatedEvent g = (GroupVisibilityUpdatedEvent) e;
 			if (g.getVisibility() == SHARED &&
@@ -373,6 +396,18 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 				LOG.info("Group shared");
 				onDataToSend();
 			}
+		} else if (e instanceof ContactConnectedEvent) {
+			ContactConnectedEvent c = (ContactConnectedEvent) e;
+			if (c.getContactId().equals(contactId)) {
+				LOG.info("Contact connected");
+				onContactConnected();
+			}
+		} else if (e instanceof ContactDisconnectedEvent) {
+			ContactDisconnectedEvent c = (ContactDisconnectedEvent) e;
+			if (c.getContactId().equals(contactId)) {
+				LOG.info("Contact disconnected");
+				onContactDisconnected();
+			}
 		}
 	}
 
@@ -391,4 +426,36 @@ class MailboxUploadWorker implements MailboxWorker, ConnectivityObserver,
 		// If we had scheduled a wakeup when data was due to be sent, cancel it
 		if (wakeupTask != null) wakeupTask.cancel();
 	}
+
+	@EventExecutor
+	private void onContactConnected() {
+		Cancellable wakeupTask = null, checkTask = null;
+		synchronized (lock) {
+			if (state == State.DESTROYED) return;
+			// If we're checking for data to send, waiting for data to send,
+			// or checking connectivity then wait until we disconnect from
+			// the contact before proceeding. If we're writing or uploading
+			// a file then continue
+			if (state == State.CHECKING_FOR_DATA ||
+					state == State.WAITING_FOR_DATA ||
+					state == State.CONNECTIVITY_CHECK) {
+				state = State.CONNECTED_TO_CONTACT;
+				wakeupTask = this.wakeupTask;
+				this.wakeupTask = null;
+				checkTask = this.checkTask;
+				this.checkTask = null;
+			}
+		}
+		if (wakeupTask != null) wakeupTask.cancel();
+		if (checkTask != null) checkTask.cancel();
+	}
+
+	@EventExecutor
+	private void onContactDisconnected() {
+		synchronized (lock) {
+			if (state != State.CONNECTED_TO_CONTACT) return;
+			state = State.CHECKING_FOR_DATA;
+		}
+		ioExecutor.execute(this::checkForDataToSend);
+	}
 }

bramble-core/src/main/java/org/briarproject/bramble/mailbox/MailboxWorkerFactoryImpl.java

@@ -1,5 +1,6 @@
 package org.briarproject.bramble.mailbox;
 
+import org.briarproject.bramble.api.connection.ConnectionRegistry;
 import org.briarproject.bramble.api.contact.ContactId;
 import org.briarproject.bramble.api.db.DatabaseComponent;
 import org.briarproject.bramble.api.event.EventBus;
@@ -25,6 +26,7 @@ class MailboxWorkerFactoryImpl implements MailboxWorkerFactory {
 	private final Clock clock;
 	private final TaskScheduler taskScheduler;
 	private final EventBus eventBus;
+	private final ConnectionRegistry connectionRegistry;
 	private final MailboxApiCaller mailboxApiCaller;
 	private final MailboxApi mailboxApi;
 	private final MailboxFileManager mailboxFileManager;
@@ -36,6 +38,7 @@ class MailboxWorkerFactoryImpl implements MailboxWorkerFactory {
 			Clock clock,
 			TaskScheduler taskScheduler,
 			EventBus eventBus,
+			ConnectionRegistry connectionRegistry,
 			MailboxApiCaller mailboxApiCaller,
 			MailboxApi mailboxApi,
 			MailboxFileManager mailboxFileManager,
@@ -45,6 +48,7 @@ class MailboxWorkerFactoryImpl implements MailboxWorkerFactory {
 		this.clock = clock;
 		this.taskScheduler = taskScheduler;
 		this.eventBus = eventBus;
+		this.connectionRegistry = connectionRegistry;
 		this.mailboxApiCaller = mailboxApiCaller;
 		this.mailboxApi = mailboxApi;
 		this.mailboxFileManager = mailboxFileManager;
@@ -57,9 +61,9 @@ class MailboxWorkerFactoryImpl implements MailboxWorkerFactory {
 			MailboxProperties properties, MailboxFolderId folderId,
 			ContactId contactId) {
 		MailboxUploadWorker worker = new MailboxUploadWorker(ioExecutor, db,
-				clock, taskScheduler, eventBus, connectivityChecker,
-				mailboxApiCaller, mailboxApi, mailboxFileManager,
-				properties, folderId, contactId);
+				clock, taskScheduler, eventBus, connectionRegistry,
+				connectivityChecker, mailboxApiCaller, mailboxApi,
+				mailboxFileManager, properties, folderId, contactId);
 		eventBus.addListener(worker);
 		return worker;
 	}

bramble-core/src/test/java/org/briarproject/bramble/mailbox/MailboxUploadWorkerTest.java

@@ -1,12 +1,16 @@
 package org.briarproject.bramble.mailbox;
 
 import org.briarproject.bramble.api.Cancellable;
+import org.briarproject.bramble.api.connection.ConnectionRegistry;
 import org.briarproject.bramble.api.contact.ContactId;
 import org.briarproject.bramble.api.db.DatabaseComponent;
 import org.briarproject.bramble.api.db.Transaction;
 import org.briarproject.bramble.api.event.EventBus;
 import org.briarproject.bramble.api.mailbox.MailboxFolderId;
 import org.briarproject.bramble.api.mailbox.MailboxProperties;
+import org.briarproject.bramble.api.plugin.event.ContactConnectedEvent;
+import org.briarproject.bramble.api.plugin.event.ContactDisconnectedEvent;
+import org.briarproject.bramble.api.sync.GroupId;
 import org.briarproject.bramble.api.sync.MessageId;
 import org.briarproject.bramble.api.sync.OutgoingSessionRecord;
 import org.briarproject.bramble.api.sync.event.MessageSharedEvent;
@@ -25,10 +29,12 @@ import org.junit.Test;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Map;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.briarproject.bramble.api.mailbox.MailboxConstants.CLIENT_SUPPORTS;
 import static org.briarproject.bramble.api.mailbox.MailboxConstants.MAX_LATENCY;
@@ -50,6 +56,8 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 	private final TaskScheduler taskScheduler =
 			context.mock(TaskScheduler.class);
 	private final EventBus eventBus = context.mock(EventBus.class);
+	private final ConnectionRegistry connectionRegistry =
+			context.mock(ConnectionRegistry.class);
 	private final ConnectivityChecker connectivityChecker =
 			context.mock(ConnectivityChecker.class);
 	private final MailboxApiCaller mailboxApiCaller =
@@ -72,6 +80,9 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 	private final MessageId ackedId = new MessageId(getRandomId());
 	private final MessageId sentId = new MessageId(getRandomId());
 	private final MessageId newMessageId = new MessageId(getRandomId());
+	private final GroupId groupId = new GroupId(getRandomId());
+	private final Map<ContactId, Boolean> groupVisibility =
+			singletonMap(contactId, true);
 
 	private File testDir, tempFile;
 	private MailboxUploadWorker worker;
@@ -81,8 +92,9 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		testDir = getTestDirectory();
 		tempFile = new File(testDir, "temp");
 		worker = new MailboxUploadWorker(ioExecutor, db, clock, taskScheduler,
-				eventBus, connectivityChecker, mailboxApiCaller, mailboxApi,
-				mailboxFileManager, mailboxProperties, folderId, contactId);
+				eventBus, connectionRegistry, connectivityChecker,
+				mailboxApiCaller, mailboxApi, mailboxFileManager,
+				mailboxProperties, folderId, contactId);
 	}
 
 	@After
@@ -93,8 +105,11 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 	@Test
 	public void testChecksForDataWhenStartedAndRemovesObserverWhenDestroyed()
 			throws Exception {
-		// When the worker is started it should check for data to send
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send
 		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendNoDataWaiting();
 
 		worker.start();
@@ -106,15 +121,59 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		worker.destroy();
 	}
 
+	@Test
+	public void testDoesNotCheckForDataWhenStartedIfConnectedToContact() {
+		// When the worker is started it should check the connection registry.
+		// We're connected to the contact, so the worker should not check for
+		// data to send
+		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(true);
+
+		worker.start();
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveObserverAndListener();
+
+		worker.destroy();
+	}
+
+	@Test
+	public void testChecksForDataWhenContactDisconnects() throws Exception {
+		// When the worker is started it should check the connection registry.
+		// We're connected to the contact, so the worker should not check for
+		// data to send
+		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(true);
+
+		worker.start();
+
+		// When the contact disconnects, the worker should start a task to
+		// check for data to send
+		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
+		expectCheckForDataToSendNoDataWaiting();
+
+		worker.eventOccurred(new ContactDisconnectedEvent(contactId));
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveObserverAndListener();
+
+		worker.destroy();
+	}
+
 	@Test
 	public void testChecksConnectivityWhenStartedIfDataIsReady()
 			throws Exception {
 		Transaction recordTxn = new Transaction(null, false);
 
-		// When the worker is started it should check for data to send. As
-		// there's data ready to send immediately, the worker should start a
-		// connectivity check
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send. As there's data ready to send immediately, the worker
+		// should start a connectivity check
 		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendAndStartConnectivityCheck();
 
 		worker.start();
@@ -149,7 +208,9 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		worker.onConnectivityCheckSucceeded();
 
 		// When the upload task runs, it should upload the file, record
-		// the acked/sent messages in the DB, and check for more data to send
+		// the acked/sent messages in the DB, and check the connection
+		// registry. We're not connected to the contact, so the worker should
+		// check for more data to send
 		context.checking(new DbExpectations() {{
 			oneOf(mailboxApi).addFile(mailboxProperties, folderId, tempFile);
 			oneOf(db).transaction(with(false), withDbRunnable(recordTxn));
@@ -157,6 +218,7 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 			oneOf(db).setMessagesSent(recordTxn, contactId,
 					singletonList(sentId), MAX_LATENCY);
 		}});
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendNoDataWaiting();
 
 		assertFalse(upload.get().callApi());
@@ -172,11 +234,41 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 	}
 
 	@Test
-	public void testCancelsApiCallWhenDestroyed() throws Exception {
-		// When the worker is started it should check for data to send. As
-		// there's data ready to send immediately, the worker should start a
-		// connectivity check
+	public void testDoesNotWriteFileIfContactConnectsDuringConnectivityCheck()
+			throws Exception {
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send. As there's data ready to send immediately, the worker
+		// should start a connectivity check
 		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
+		expectCheckForDataToSendAndStartConnectivityCheck();
+
+		worker.start();
+
+		// Before the connectivity check succeeds, we make a direct connection
+		// to the contact
+		worker.eventOccurred(new ContactConnectedEvent(contactId));
+
+		// When the connectivity check succeeds, the worker should not start
+		// writing and uploading a file
+		worker.onConnectivityCheckSucceeded();
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveObserverAndListener();
+
+		worker.destroy();
+	}
+
+	@Test
+	public void testCancelsApiCallWhenDestroyed() throws Exception {
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send. As there's data ready to send immediately, the worker
+		// should start a connectivity check
+		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendAndStartConnectivityCheck();
 
 		worker.start();
@@ -212,9 +304,7 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 
 		// When the worker is destroyed it should remove the connectivity
 		// observer and event listener and cancel the upload task
-		context.checking(new Expectations() {{
-			oneOf(apiCall).cancel();
-		}});
+		expectCancelTask(apiCall);
 		expectRemoveObserverAndListener();
 
 		worker.destroy();
@@ -230,16 +320,21 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 	@Test
 	public void testSchedulesWakeupWhenStartedIfDataIsNotReady()
 			throws Exception {
-		// When the worker is started it should check for data to send. As
-		// the data isn't ready to send immediately, the worker should
-		// schedule a wakeup
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send. As the data isn't ready to send immediately, the
+		// worker should schedule a wakeup
 		expectRunTaskOnIoExecutor();
 		AtomicReference<Runnable> wakeup = new AtomicReference<>();
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendAndScheduleWakeup(wakeup);
 
 		worker.start();
 
-		// When the wakeup task runs it should check for data to send
+		// When the wakeup task runs it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendNoDataWaiting();
 
 		wakeup.get().run();
@@ -252,21 +347,51 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 	}
 
 	@Test
-	public void testCancelsWakeupIfDestroyedBeforeWakingUp() throws Exception {
+	public void testCancelsWakeupIfContactConnectsBeforeWakingUp()
+			throws Exception {
 		// When the worker is started it should check for data to send. As
 		// the data isn't ready to send immediately, the worker should
 		// schedule a wakeup
 		expectRunTaskOnIoExecutor();
 		AtomicReference<Runnable> wakeup = new AtomicReference<>();
+		expectCheckConnectionRegistry(false);
+		expectCheckForDataToSendAndScheduleWakeup(wakeup);
+
+		worker.start();
+
+		// Before the wakeup task runs, we make a direct connection to the
+		// contact. The worker should cancel the wakeup task
+		expectCancelTask(wakeupTask);
+
+		worker.eventOccurred(new ContactConnectedEvent(contactId));
+
+		// If the wakeup task runs anyway (cancellation came too late), it
+		// should return without doing anything
+		wakeup.get().run();
+
+		// When the worker is destroyed it should remove the connectivity
+		// observer and event listener
+		expectRemoveObserverAndListener();
+
+		worker.destroy();
+	}
+
+	@Test
+	public void testCancelsWakeupIfDestroyedBeforeWakingUp() throws Exception {
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send. As the data isn't ready to send immediately, the
+		// worker should schedule a wakeup
+		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
+		AtomicReference<Runnable> wakeup = new AtomicReference<>();
 		expectCheckForDataToSendAndScheduleWakeup(wakeup);
 
 		worker.start();
 
 		// When the worker is destroyed it should cancel the wakeup and
 		// remove the connectivity observer and event listener
-		context.checking(new Expectations() {{
-			oneOf(wakeupTask).cancel();
-		}});
+		expectCancelTask(wakeupTask);
 		expectRemoveObserverAndListener();
 
 		worker.destroy();
@@ -279,10 +404,12 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 	@Test
 	public void testCancelsWakeupIfEventIsReceivedBeforeWakingUp()
 			throws Exception {
-		// When the worker is started it should check for data to send. As
-		// the data isn't ready to send immediately, the worker should
-		// schedule a wakeup
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send. As the data isn't ready to send immediately, the
+		// worker should schedule a wakeup
 		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
 		AtomicReference<Runnable> wakeup = new AtomicReference<>();
 		expectCheckForDataToSendAndScheduleWakeup(wakeup);
 
@@ -293,11 +420,10 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		// wakeup task and schedule a check for new data after a short delay
 		AtomicReference<Runnable> check = new AtomicReference<>();
 		expectScheduleCheck(check, CHECK_DELAY_MS);
-		context.checking(new Expectations() {{
-			oneOf(wakeupTask).cancel();
-		}});
+		expectCancelTask(wakeupTask);
 
-		worker.eventOccurred(new MessageSharedEvent(newMessageId));
+		worker.eventOccurred(new MessageSharedEvent(newMessageId, groupId,
+				groupVisibility));
 
 		// If the wakeup task runs anyway (cancellation came too late), it
 		// should return early when it finds the state has changed
@@ -306,9 +432,13 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		// Before the check task runs, the worker receives another event that
 		// indicates new data may be available. The event should be ignored,
 		// as a check for new data has already been scheduled
-		worker.eventOccurred(new MessageSharedEvent(newMessageId));
+		worker.eventOccurred(new MessageSharedEvent(newMessageId, groupId,
+				groupVisibility));
 
-		// When the check task runs, it should check for new data
+		// When the check task runs, it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// new data
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendNoDataWaiting();
 
 		check.get().run();
@@ -322,8 +452,11 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 
 	@Test
 	public void testCancelsCheckWhenDestroyed() throws Exception {
-		// When the worker is started it should check for data to send
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send
 		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendNoDataWaiting();
 
 		worker.start();
@@ -334,13 +467,12 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		AtomicReference<Runnable> check = new AtomicReference<>();
 		expectScheduleCheck(check, CHECK_DELAY_MS);
 
-		worker.eventOccurred(new MessageSharedEvent(newMessageId));
+		worker.eventOccurred(new MessageSharedEvent(newMessageId, groupId,
+				groupVisibility));
 
 		// When the worker is destroyed it should cancel the check and
 		// remove the connectivity observer and event listener
-		context.checking(new Expectations() {{
-			oneOf(checkTask).cancel();
-		}});
+		expectCancelTask(checkTask);
 		expectRemoveObserverAndListener();
 
 		worker.destroy();
@@ -350,13 +482,52 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		check.get().run();
 	}
 
+	@Test
+	public void testCancelsCheckIfContactConnects() throws Exception {
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send
+		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
+		expectCheckForDataToSendNoDataWaiting();
+
+		worker.start();
+
+		// The worker receives an event that indicates new data may be
+		// available. The worker should schedule a check for new data after
+		// a short delay
+		AtomicReference<Runnable> check = new AtomicReference<>();
+		expectScheduleCheck(check, CHECK_DELAY_MS);
+
+		worker.eventOccurred(new MessageSharedEvent(newMessageId, groupId,
+				groupVisibility));
+
+		// Before the check task runs, we make a direct connection to the
+		// contact. The worker should cancel the check
+		expectCancelTask(checkTask);
+
+		worker.eventOccurred(new ContactConnectedEvent(contactId));
+
+		// If the check runs anyway (cancellation came too late), it should
+		// return early when it finds the state has changed
+		check.get().run();
+
+		// When the worker is destroyed it should cancel the check and
+		// remove the connectivity observer and event listener
+		expectRemoveObserverAndListener();
+
+		worker.destroy();
+	}
+
 	@Test
 	public void testRetriesAfterDelayIfExceptionOccursWhileWritingFile()
 			throws Exception {
-		// When the worker is started it should check for data to send. As
-		// there's data ready to send immediately, the worker should start a
-		// connectivity check
+		// When the worker is started it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// data to send. As there's data ready to send immediately, the worker
+		// should start a connectivity check
 		expectRunTaskOnIoExecutor();
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendAndStartConnectivityCheck();
 
 		worker.start();
@@ -375,7 +546,10 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 
 		worker.onConnectivityCheckSucceeded();
 
-		// When the check task runs it should check for new data
+		// When the check task runs it should check the connection registry.
+		// We're not connected to the contact, so the worker should check for
+		// new data
+		expectCheckConnectionRegistry(false);
 		expectCheckForDataToSendNoDataWaiting();
 
 		check.get().run();
@@ -387,6 +561,13 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		worker.destroy();
 	}
 
+	private void expectCheckConnectionRegistry(boolean connected) {
+		context.checking(new Expectations() {{
+			oneOf(connectionRegistry).isConnected(contactId);
+			will(returnValue(connected));
+		}});
+	}
+
 	private void expectRunTaskOnIoExecutor() {
 		context.checking(new Expectations() {{
 			oneOf(ioExecutor).execute(with(any(Runnable.class)));
@@ -456,6 +637,12 @@ public class MailboxUploadWorkerTest extends BrambleMockTestCase {
 		}});
 	}
 
+	private void expectCancelTask(Cancellable task) {
+		context.checking(new Expectations() {{
+			oneOf(task).cancel();
+		}});
+	}
+
 	private void expectRemoveObserverAndListener() {
 		context.checking(new Expectations() {{
 			oneOf(connectivityChecker).removeObserver(worker);