diff --git a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/InstancesDataService.kt b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/InstancesDataService.kt index 318db463542..f796d54bded 100644 --- a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/InstancesDataService.kt +++ b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/InstancesDataService.kt @@ -1,6 +1,8 @@ package org.odk.collect.android.instancemanagement +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.isActive import org.odk.collect.analytics.Analytics import org.odk.collect.android.analytics.AnalyticsEvents import org.odk.collect.android.application.Collect @@ -213,32 +215,34 @@ class InstancesDataService( } } - fun sendInstances( + suspend fun sendInstances( projectId: String, instances: List, referrer: String, overrideURL: String?, - cancelAfterAuthException: Boolean, externalDeleteAfterUpload: Boolean?, defaultSuccessMessage: String, - ensureActive: () -> Unit, onProgress: (current: Int, total: Int) -> Unit = { _, _ -> } ): List { val projectDependencyModule = projectDependencyModuleFactory.create(projectId) + val coroutineContext = currentCoroutineContext() - return projectDependencyModule.instancesLock.withLock { acquiredLock: Boolean -> + return projectDependencyModule.instancesLock.withLockSuspend { acquiredLock: Boolean -> if (acquiredLock) { - instanceSubmitter.submitInstances( + val result = instanceSubmitter.submitInstances( projectId, instances, referrer, overrideURL, - cancelAfterAuthException, - externalDeleteAfterUpload, - defaultSuccessMessage, - ensureActive, - onProgress + cancelAfterAuthException = true, + externalDeleteAfterUpload = externalDeleteAfterUpload, + defaultSuccessMessage = defaultSuccessMessage, + isCancelled = { !coroutineContext.isActive }, + onProgress = onProgress ) + update(projectId) + + result } else { emptyList() } diff --git a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/FormUploadException.kt b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/FormUploadException.kt index cb13e263bab..a0dd2dcbc66 100644 --- a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/FormUploadException.kt +++ b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/FormUploadException.kt @@ -47,3 +47,10 @@ class FormUploadAuthRequestedException( message: String, val authRequestingServer: Uri ) : FormUploadException(message) + +/** + * Thrown to indicate that an upload was interrupted because the submission attempt was cancelled. + * Unlike other [FormUploadException]s this should not be reported to the user as an error - it + * simply stops the current submission attempt. + */ +class FormUploadInterruptedException : FormUploadException("Upload interrupted") diff --git a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceSubmitter.kt b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceSubmitter.kt index 99b2fa2f7a1..d2c16794cae 100644 --- a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceSubmitter.kt +++ b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceSubmitter.kt @@ -27,7 +27,7 @@ class InstanceSubmitter( cancelAfterAuthException: Boolean = false, externalDeleteAfterUpload: Boolean? = null, defaultSuccessMessage: String? = null, - ensureActive: () -> Unit = {}, + isCancelled: () -> Boolean = { false }, onProgress: (current: Int, total: Int) -> Unit = { _, _ -> } ): List { val projectDependencyModule = projectDependencyFactory.create(projectId) @@ -40,14 +40,14 @@ class InstanceSubmitter( val sortedInstances = toUpload.sortedBy { it.finalizationDate } for ((index, instance) in sortedInstances.withIndex()) { - ensureActive() - onProgress( index + 1, sortedInstances.size) + onProgress(index + 1, sortedInstances.size) try { - val resultMessage = instanceUploader.uploadOneSubmission(projectId, instance, deviceId, overrideURL, referrer) + val resultMessage = instanceUploader.uploadOneSubmission(projectId, instance, deviceId, overrideURL, referrer, isCancelled) uploadResults.add(InstanceUploadResult.Success(instance, resultMessage ?: defaultSuccessMessage)) - deleteInstance(instance, formsRepository, instancesRepository, generalSettings, externalDeleteAfterUpload) + } catch (_: FormUploadInterruptedException) { + break } catch (e: FormUploadException) { Timber.d(e) uploadResults.add(InstanceUploadResult.Error(instance, e)) diff --git a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploader.kt b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploader.kt index 2b5577dd919..0156e43a956 100644 --- a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploader.kt +++ b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploader.kt @@ -4,5 +4,12 @@ import org.odk.collect.forms.instances.Instance interface InstanceUploader { @Throws(FormUploadException::class) - fun uploadOneSubmission(projectId: String, instance: Instance, deviceId: String?, overrideURL: String?, referrer: String): String? + fun uploadOneSubmission( + projectId: String, + instance: Instance, + deviceId: String?, + overrideURL: String?, + referrer: String, + isCancelled: () -> Boolean + ): String? } diff --git a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploaderViewModel.kt b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploaderViewModel.kt index 5c4ec76dcd1..318437c55b8 100644 --- a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploaderViewModel.kt +++ b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/InstanceUploaderViewModel.kt @@ -6,7 +6,6 @@ import androidx.lifecycle.MutableLiveData import androidx.lifecycle.ViewModel import androidx.lifecycle.viewModelScope import kotlinx.coroutines.Job -import kotlinx.coroutines.ensureActive import kotlinx.coroutines.launch import org.odk.collect.android.instancemanagement.InstancesDataService import org.odk.collect.android.utilities.WebCredentialsUtils @@ -52,10 +51,8 @@ class InstanceUploadViewModel( instancesToUpload, referrer, externalUrl, - true, externalDeleteAfterUpload, defaultSuccessMessage, - { coroutineContext.ensureActive() } ) { current, total -> _state.postValue(UploadState.Progress(current, total)) } @@ -71,7 +68,6 @@ class InstanceUploadViewModel( } clearTemporaryCredentials() - instancesDataService.update(projectId) _state.postValue(UploadState.Completed(uploadResults)) } } diff --git a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/OpenRosaServerInstanceUploader.kt b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/OpenRosaServerInstanceUploader.kt index f69402f09d6..2f33f7239f2 100644 --- a/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/OpenRosaServerInstanceUploader.kt +++ b/collect_app/src/main/java/org/odk/collect/android/instancemanagement/send/OpenRosaServerInstanceUploader.kt @@ -44,8 +44,13 @@ class OpenRosaServerInstanceUploader( instance: Instance, deviceId: String?, overrideURL: String?, - referrer: String + referrer: String, + isCancelled: () -> Boolean ): String? { + if (isCancelled()) { + throw FormUploadInterruptedException() + } + val projectDependencyModule = projectDependencyFactory.create(projectId) val unprotectedSettings = projectDependencyModule.generalSettings val instancesRepository = projectDependencyModule.instancesRepository @@ -170,6 +175,10 @@ class OpenRosaServerInstanceUploader( val messageParser = ResponseMessageParser() + if (isCancelled()) { + throw FormUploadInterruptedException() + } + try { val uri = URI.create(submissionUri.toString()) val postResult = httpInterface.uploadSubmissionAndFiles( @@ -178,7 +187,7 @@ class OpenRosaServerInstanceUploader( uri, webCredentialsUtils.getCredentials(uri), contentLength - ) + ) { isCancelled() } val responseCode = postResult.responseCode messageParser.setMessageResponse(postResult.httpResponse) @@ -203,6 +212,9 @@ class OpenRosaServerInstanceUploader( } } catch (e: Exception) { + if (isCancelled()) { + throw FormUploadInterruptedException() + } throw FormUploadException(e.message ?: e.toString()) } diff --git a/collect_app/src/test/java/org/odk/collect/android/instancemanagement/send/InstanceUploadViewModelTest.kt b/collect_app/src/test/java/org/odk/collect/android/instancemanagement/send/InstanceUploadViewModelTest.kt index 57ae5efde6e..f9eaf5d15fc 100644 --- a/collect_app/src/test/java/org/odk/collect/android/instancemanagement/send/InstanceUploadViewModelTest.kt +++ b/collect_app/src/test/java/org/odk/collect/android/instancemanagement/send/InstanceUploadViewModelTest.kt @@ -26,7 +26,102 @@ class InstanceUploadViewModelTest { private lateinit var viewModel: InstanceUploadViewModel @Test - fun `submitted instance is deleted even when upload is cancelled`() { + fun `instance is not submitted when upload is canceled`() { + val form = FormFixtures.form("1") + val formsRepository = InMemFormsRepository().apply { + save(form) + } + + val instance = Instance.Builder() + .dbId(1) + .formId(form.formId) + .formVersion(form.version) + .status(Instance.STATUS_COMPLETE) + .finalizationDate(1) + .build() + + val instancesRepository = InMemInstancesRepository().apply { + save(instance) + } + + val projectsDependencyModuleFactory = CachingProjectDependencyModuleFactory { projectId -> + ProjectDependencyModule( + projectId, + { InMemSettings() }, + { formsRepository }, + { instancesRepository }, + mock(), + { ChangeLocks(BooleanChangeLock(), BooleanChangeLock()) }, + mock(), + mock(), + mock(), + mock(), + mock() + ) + } + + val submittedInstances = mutableListOf() + + val instanceUploader = object : InstanceUploader { + override fun uploadOneSubmission( + projectId: String, + instance: Instance, + deviceId: String?, + overrideURL: String?, + referrer: String, + isCancelled: () -> Boolean + ): String { + viewModel.cancel() + if (isCancelled()) { + throw FormUploadInterruptedException() + } + + submittedInstances.add(instance.dbId) + instancesRepository.save( + Instance.Builder(instance) + .status(Instance.STATUS_SUBMITTED) + .build() + ) + return "Success" + } + } + + val instancesSubmitter = InstanceSubmitter( + instanceUploader, + projectsDependencyModuleFactory, + mock() + ) + val instancesDataService = InstancesDataService( + AppState(), + mock(), + projectsDependencyModuleFactory, + mock(), + instancesSubmitter + ) {} + + val dispatcherProvider = TestDispatcherProvider() + viewModel = InstanceUploadViewModel( + dispatcherProvider, + mock(), + instancesRepository, + instancesDataService, + "projectId", + "", + null, + null, + null, + null, + "Success", + "Waiting" + ) + viewModel.upload(listOf(instance.dbId)) + dispatcherProvider.runBackground() + + assertThat(submittedInstances.isEmpty(), equalTo(true)) + } + + @Test + fun `instances uploaded before cancellation get deleted`() { val form = FormFixtures.form("1") val formsRepository = InMemFormsRepository().apply { save(form) @@ -75,21 +170,29 @@ class InstanceUploadViewModelTest { val submittedInstances = mutableListOf() + var progress = 0; val instanceUploader = object : InstanceUploader { override fun uploadOneSubmission( projectId: String, instance: Instance, deviceId: String?, overrideURL: String?, - referrer: String + referrer: String, + isCancelled: () -> Boolean ): String { + if (++progress == 2) { + viewModel.cancel() + } + if (isCancelled()) { + throw FormUploadInterruptedException() + } + submittedInstances.add(instance.dbId) instancesRepository.save( Instance.Builder(instance) .status(Instance.STATUS_SUBMITTED) .build() ) - viewModel.cancel() return "Success" } } diff --git a/open-rosa/src/main/java/org/odk/collect/openrosa/http/OpenRosaHttpInterface.java b/open-rosa/src/main/java/org/odk/collect/openrosa/http/OpenRosaHttpInterface.java index 07f7db1c360..ceb5bf29e00 100644 --- a/open-rosa/src/main/java/org/odk/collect/openrosa/http/OpenRosaHttpInterface.java +++ b/open-rosa/src/main/java/org/odk/collect/openrosa/http/OpenRosaHttpInterface.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.net.URI; import java.util.List; +import java.util.function.Supplier; public interface OpenRosaHttpInterface { @@ -66,6 +67,21 @@ HttpPostResult uploadSubmissionAndFiles(@NonNull File submissionFile, @NonNull HttpCredentialsInterface credentials, @NonNull long contentLength) throws Exception; + /** + * Variant of {@link #uploadSubmissionAndFiles} that aborts an in-progress upload when + * {@code isCancelled} starts returning {@code true}. The default delegates to the + * non-cancellable variant so existing implementations keep working. + */ + @NonNull + default HttpPostResult uploadSubmissionAndFiles(@NonNull File submissionFile, + @NonNull List fileList, + @NonNull URI uri, + @NonNull HttpCredentialsInterface credentials, + @NonNull long contentLength, + @NonNull Supplier isCancelled) throws Exception { + return uploadSubmissionAndFiles(submissionFile, fileList, uri, credentials, contentLength); + } + interface FileToContentTypeMapper { @NonNull diff --git a/open-rosa/src/main/java/org/odk/collect/openrosa/http/okhttp/OkHttpConnection.java b/open-rosa/src/main/java/org/odk/collect/openrosa/http/okhttp/OkHttpConnection.java index aabae7c8f0f..adf1e8e150d 100644 --- a/open-rosa/src/main/java/org/odk/collect/openrosa/http/okhttp/OkHttpConnection.java +++ b/open-rosa/src/main/java/org/odk/collect/openrosa/http/okhttp/OkHttpConnection.java @@ -15,7 +15,10 @@ import java.io.ByteArrayInputStream; import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; import java.io.InputStream; +import java.io.InterruptedIOException; import java.net.HttpURLConnection; import java.net.URI; import java.util.Date; @@ -23,6 +26,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.function.Supplier; import okhttp3.Headers; import okhttp3.MediaType; @@ -31,6 +35,7 @@ import okhttp3.RequestBody; import okhttp3.Response; import okhttp3.ResponseBody; +import okio.BufferedSink; import timber.log.Timber; public class OkHttpConnection implements OpenRosaHttpInterface { @@ -138,17 +143,27 @@ public HttpHeadResult executeHeadRequest(@NonNull URI uri, @Nullable HttpCredent @NonNull @Override public HttpPostResult uploadSubmissionAndFiles(@NonNull File submissionFile, @NonNull List fileList, @NonNull URI uri, @Nullable HttpCredentialsInterface credentials, @NonNull long contentLength) throws Exception { + return uploadSubmissionAndFiles(submissionFile, fileList, uri, credentials, contentLength, () -> false); + } + + @NonNull + @Override + public HttpPostResult uploadSubmissionAndFiles(@NonNull File submissionFile, @NonNull List fileList, @NonNull URI uri, @Nullable HttpCredentialsInterface credentials, @NonNull long contentLength, @NonNull Supplier isCancelled) throws Exception { HttpPostResult postResult = null; boolean first = true; int fileIndex = 0; int lastFileIndex; while (fileIndex < fileList.size() || first) { + if (isCancelled.get()) { + throw new InterruptedIOException("Upload canceled"); + } + lastFileIndex = fileIndex; first = false; long byteCount = 0L; - RequestBody requestBody = RequestBody.create(MediaType.parse(HTTP_CONTENT_TYPE_TEXT_XML), submissionFile); + RequestBody requestBody = cancellableRequestBody(MediaType.parse(HTTP_CONTENT_TYPE_TEXT_XML), submissionFile, isCancelled); MultipartBody.Builder multipartBuilder = new MultipartBody.Builder() .setType(MultipartBody.FORM) @@ -162,7 +177,7 @@ public HttpPostResult uploadSubmissionAndFiles(@NonNull File submissionFile, @No String contentType = fileToContentTypeMapper.map(file.getName()); - RequestBody fileRequestBody = RequestBody.create(MediaType.parse(contentType), file); + RequestBody fileRequestBody = cancellableRequestBody(MediaType.parse(contentType), file, isCancelled); multipartBuilder.addPart(MultipartBody.Part.createFormData(file.getName(), file.getName(), fileRequestBody)); byteCount += file.length(); @@ -218,6 +233,40 @@ private HttpPostResult executePostRequest(@NonNull URI uri, @Nullable HttpCreden return postResult; } + /** + * Wraps a file in a {@link RequestBody} that streams it in chunks and aborts the upload (by + * throwing {@link InterruptedIOException}) as soon as {@code isCancelled} starts returning + * {@code true}. This mirrors the cancellable download in {@code FileUtils.interuptablyWriteFile} + * so an in-progress submission can be stopped mid-transfer. + */ + private static RequestBody cancellableRequestBody(MediaType contentType, File file, @NonNull Supplier isCancelled) { + return new RequestBody() { + @Override + public MediaType contentType() { + return contentType; + } + + @Override + public long contentLength() { + return file.length(); + } + + @Override + public void writeTo(@NonNull BufferedSink sink) throws IOException { + byte[] buffer = new byte[IOUtils.DEFAULT_BUFFER_SIZE]; + try (InputStream inputStream = new FileInputStream(file)) { + int read; + while ((read = inputStream.read(buffer)) != -1) { + if (isCancelled.get()) { + throw new InterruptedIOException("Upload canceled"); + } + sink.write(buffer, 0, read); + } + } + } + }; + } + /** * Utility to ensure that the entity stream of a response is drained of * bytes. diff --git a/open-rosa/src/test/java/org/odk/collect/openrosa/http/OpenRosaPostRequestTest.java b/open-rosa/src/test/java/org/odk/collect/openrosa/http/OpenRosaPostRequestTest.java index 311166228f3..53d54b82f41 100644 --- a/open-rosa/src/test/java/org/odk/collect/openrosa/http/OpenRosaPostRequestTest.java +++ b/open-rosa/src/test/java/org/odk/collect/openrosa/http/OpenRosaPostRequestTest.java @@ -7,6 +7,7 @@ import static org.hamcrest.Matchers.isA; import static org.hamcrest.Matchers.isEmptyString; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.fail; import static java.util.Arrays.asList; @@ -22,10 +23,12 @@ import java.io.File; import java.io.FileWriter; import java.io.IOException; +import java.io.InterruptedIOException; import java.net.URI; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.zip.GZIPOutputStream; @@ -243,6 +246,47 @@ public void whenMoreThanOneAttachment_andRequestIsLargerThanMaxContentLength_and assertThat(response.getResponseCode(), equalTo(500)); } + @Test + public void whenCanceledBeforeUpload_doesNotMakeRequest() { + URI uri = mockWebServer.url("/blah").uri(); + + assertThrows( + InterruptedIOException.class, + () -> subject.uploadSubmissionAndFiles( + createTempFile("content"), + new ArrayList<>(), + uri, + null, + 0, + () -> true + ) + ); + + assertThat(mockWebServer.getRequestCount(), equalTo(0)); + } + + @Test + public void whenCanceledDuringUpload_abortsRequest() { + mockWebServer.enqueue(new MockResponse().setResponseCode(201)); + + // Not canceled for the initial loop check, then canceled once the body starts streaming. + AtomicInteger checks = new AtomicInteger(0); + + URI uri = mockWebServer.url("/blah").uri(); + + assertThrows( + InterruptedIOException.class, + () -> subject.uploadSubmissionAndFiles( + createTempFile("content"), + new ArrayList<>(), + uri, + null, + 0, + () -> checks.getAndIncrement() > 0 + ) + ); + } + private File createTempFile(String content) throws Exception { return createTempFile(content, ".tmp"); } diff --git a/shared/src/main/java/org/odk/collect/shared/locks/ChangeLock.kt b/shared/src/main/java/org/odk/collect/shared/locks/ChangeLock.kt index 15c6a0c0462..9af1896a151 100644 --- a/shared/src/main/java/org/odk/collect/shared/locks/ChangeLock.kt +++ b/shared/src/main/java/org/odk/collect/shared/locks/ChangeLock.kt @@ -20,6 +20,18 @@ interface ChangeLock { } } + suspend fun withLockSuspend(function: suspend (Boolean) -> T): T { + val acquired = tryLock(DEFAULT_TOKEN) + + return try { + function(acquired) + } finally { + if (acquired) { + unlock(DEFAULT_TOKEN) + } + } + } + fun tryLock(token: Any): Boolean fun unlock(token: Any)