From cdff29befd06b0206935f4698d12330d3649e344 Mon Sep 17 00:00:00 2001 From: Guillaume Bernos Date: Wed, 21 Oct 2026 13:49:29 +0200 Subject: [PATCH 1/2] fix(storage): keep tasks alive when snapshotEvents listeners cancel, and scope Android task registries per engine - Dart: one internal subscription now drives `snapshot` and `onComplete`, and users get a separate broadcast stream. Cancelling the last `snapshotEvents` listener no longer cancels the native event stream (which cancelled the upload on Android) and `await task` no longer hangs afterwards. `onComplete` also fails instead of hanging when the native stream closes without a final state or the task fails to start. - Android: cancelling a task's event stream only stops sending events, as on iOS. In-progress tasks and event channels are tracked per plugin instance, so detaching a background engine no longer cancels the main engine's tasks or tears down its streams, and same-numbered handles from different engines no longer collide. --- .github/workflows/android_unit_tests.yaml | 2 +- .../firebase_storage/android/build.gradle | 5 + .../storage/FlutterFirebaseStoragePlugin.kt | 51 ++- .../storage/FlutterFirebaseStorageTask.kt | 70 ++-- .../FlutterFirebaseStorageTaskRegistry.kt | 35 ++ .../storage/TaskStateChannelStreamHandler.kt | 30 +- .../FlutterFirebaseStorageTaskRegistryTest.kt | 65 ++++ .../firebasestorageexample/MainActivity.kt | 25 +- .../example/integration_test/task_e2e.dart | 43 +++ .../method_channel/method_channel_task.dart | 72 ++-- .../method_channel_task_test.dart | 326 +++++++++++++++--- 11 files changed, 582 insertions(+), 142 deletions(-) create mode 100644 packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistry.kt create mode 100644 packages/firebase_storage/firebase_storage/android/src/test/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistryTest.kt diff --git a/.github/workflows/android_unit_tests.yaml b/.github/workflows/android_unit_tests.yaml index 2ec8d17726a1..ba14a07af892 100644 --- a/.github/workflows/android_unit_tests.yaml +++ b/.github/workflows/android_unit_tests.yaml @@ -43,7 +43,7 @@ jobs: # Tasks are listed per package rather than using the root testDebugUnitTest # task, which would also build the Dart sources of the aggregate test app. # Add a task here when a package gains an android/src/test directory. - run: cd tests/android && ./gradlew :firebase_crashlytics:testDebugUnitTest :cloud_functions:testDebugUnitTest :firebase_messaging:testDebugUnitTest + run: cd tests/android && ./gradlew :firebase_crashlytics:testDebugUnitTest :cloud_functions:testDebugUnitTest :firebase_messaging:testDebugUnitTest :firebase_storage:testDebugUnitTest - name: 'Upload test reports' if: failure() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a diff --git a/packages/firebase_storage/firebase_storage/android/build.gradle b/packages/firebase_storage/firebase_storage/android/build.gradle index 9baa3cb74bfd..aa35340b0eaa 100755 --- a/packages/firebase_storage/firebase_storage/android/build.gradle +++ b/packages/firebase_storage/firebase_storage/android/build.gradle @@ -85,6 +85,11 @@ android { implementation platform("com.google.firebase:firebase-bom:${getRootProjectExtOrCoreProperty("FirebaseSDKVersion", firebaseCoreProject)}") implementation 'com.google.firebase:firebase-storage' implementation 'androidx.annotation:annotation:1.7.0' + testImplementation 'junit:junit:4.13.2' + } + + testOptions { + unitTests.returnDefaultValues = true } } diff --git a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStoragePlugin.kt b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStoragePlugin.kt index 8a632ceb75f2..6a778a961ccc 100644 --- a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStoragePlugin.kt +++ b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStoragePlugin.kt @@ -30,12 +30,17 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb private var channel: MethodChannel? = null private var messenger: BinaryMessenger? = null + // Per instance, as each engine attaches its own plugin instance and numbers its tasks from 0. + private val inProgressTasks = FlutterFirebaseStorageTaskRegistry() + private val eventChannels: MutableMap = HashMap() + private val streamHandlers: MutableMap = HashMap() + override fun onAttachedToEngine(binding: FlutterPlugin.FlutterPluginBinding) { initInstance(binding.binaryMessenger) } override fun onDetachedFromEngine(binding: FlutterPlugin.FlutterPluginBinding) { - FlutterFirebaseStorageTask.cancelInProgressTasks() + cancelInProgressTasks() channel?.setMethodCallHandler(null) checkNotNull(messenger) FirebaseStorageHostApi.setUp(messenger!!, null) @@ -51,6 +56,11 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb this.messenger = messenger } + private fun cancelInProgressTasks() { + inProgressTasks.removeAll().forEach { it.destroy() } + } + + @Synchronized private fun registerEventChannel( prefix: String, identifier: String, @@ -64,6 +74,12 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb return identifier } + @Synchronized + private fun removeEventChannel(identifier: String) { + eventChannels.remove(identifier)?.setStreamHandler(null) + streamHandlers.remove(identifier) + } + @Synchronized private fun removeEventListeners() { val eventChannelKeys: List = ArrayList(eventChannels.keys) @@ -284,10 +300,11 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb val androidMetaData = getMetaDataFromPigeon(settableMetaData) val storageTask = FlutterFirebaseStorageTask.uploadBytes( - handle.toInt(), androidReference, data, androidMetaData) + handle.toInt(), androidReference, data, androidMetaData, inProgressTasks) try { val identifier = UUID.randomUUID().toString().lowercase(Locale.US) - val handler = storageTask.startTaskWithMethodChannel(channel!!, identifier) + val handler = + storageTask.startTaskWithMethodChannel(channel!!, identifier, ::removeEventChannel) callback( Result.success( registerEventChannel( @@ -311,10 +328,11 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb val bytes = stringToByteData(data, format.toInt()) val storageTask = FlutterFirebaseStorageTask.uploadBytes( - handle.toInt(), androidReference, bytes!!, androidMetaData) + handle.toInt(), androidReference, bytes!!, androidMetaData, inProgressTasks) try { val identifier = UUID.randomUUID().toString().lowercase(Locale.US) - val handler = storageTask.startTaskWithMethodChannel(channel!!, identifier) + val handler = + storageTask.startTaskWithMethodChannel(channel!!, identifier, ::removeEventChannel) callback( Result.success( registerEventChannel( @@ -338,10 +356,12 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb handle.toInt(), androidReference, Uri.fromFile(File(filePath)), - settableMetaData?.let { getMetaDataFromPigeon(it) }) + settableMetaData?.let { getMetaDataFromPigeon(it) }, + inProgressTasks) try { val identifier = UUID.randomUUID().toString().lowercase(Locale.US) - val handler = storageTask.startTaskWithMethodChannel(channel!!, identifier) + val handler = + storageTask.startTaskWithMethodChannel(channel!!, identifier, ::removeEventChannel) callback( Result.success( registerEventChannel( @@ -360,10 +380,12 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb ) { val androidReference = getReferenceFromPigeon(app, reference) val storageTask = - FlutterFirebaseStorageTask.downloadFile(handle.toInt(), androidReference, File(filePath)) + FlutterFirebaseStorageTask.downloadFile( + handle.toInt(), androidReference, File(filePath), inProgressTasks) try { val identifier = UUID.randomUUID().toString().lowercase(Locale.US) - val handler = storageTask.startTaskWithMethodChannel(channel!!, identifier) + val handler = + storageTask.startTaskWithMethodChannel(channel!!, identifier, ::removeEventChannel) callback( Result.success( registerEventChannel( @@ -378,7 +400,7 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb handle: Long, callback: (Result>) -> Unit ) { - val storageTask = FlutterFirebaseStorageTask.getInProgressTaskForHandle(handle.toInt()) + val storageTask = inProgressTasks.get(handle.toInt()) if (storageTask == null) { val statusMap = HashMap() statusMap["status"] = false @@ -407,7 +429,7 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb handle: Long, callback: (Result>) -> Unit ) { - val storageTask = FlutterFirebaseStorageTask.getInProgressTaskForHandle(handle.toInt()) + val storageTask = inProgressTasks.get(handle.toInt()) if (storageTask == null) { val statusMap = HashMap() statusMap["status"] = false @@ -436,7 +458,7 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb handle: Long, callback: (Result>) -> Unit ) { - val storageTask = FlutterFirebaseStorageTask.getInProgressTaskForHandle(handle.toInt()) + val storageTask = inProgressTasks.get(handle.toInt()) if (storageTask == null) { val statusMap = HashMap() statusMap["status"] = false @@ -501,7 +523,7 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb override fun didReinitializeFirebaseCore(): Task { val taskCompletionSource = TaskCompletionSource() cachedThreadPool.execute { - FlutterFirebaseStorageTask.cancelInProgressTasks() + cancelInProgressTasks() taskCompletionSource.setResult(null) removeEventListeners() } @@ -513,9 +535,6 @@ class FlutterFirebaseStoragePlugin : FlutterFirebasePlugin, FlutterPlugin, Fireb const val STORAGE_TASK_EVENT_NAME = "taskEvent" const val DEFAULT_ERROR_CODE = "firebase_storage" - val eventChannels: MutableMap = HashMap() - val streamHandlers: MutableMap = HashMap() - fun getExceptionDetails(exception: Exception): Map { val storageException = FlutterFirebaseStorageException.parserExceptionToFlutter(exception) val details: MutableMap = HashMap() diff --git a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTask.kt b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTask.kt index f8cd77d739a8..53bf26ed8988 100644 --- a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTask.kt +++ b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTask.kt @@ -6,7 +6,6 @@ package io.flutter.plugins.firebase.storage import android.net.Uri -import android.util.SparseArray import androidx.annotation.NonNull import com.google.firebase.storage.FileDownloadTask import com.google.firebase.storage.StorageMetadata @@ -24,7 +23,8 @@ private constructor( private val reference: StorageReference, private val bytes: ByteArray?, private val fileUri: Uri?, - private val metadata: StorageMetadata? + private val metadata: StorageMetadata?, + private val inProgressTasks: FlutterFirebaseStorageTaskRegistry ) { private val pauseSyncObject = Object() private val resumeSyncObject = Object() @@ -33,12 +33,13 @@ private constructor( private var destroyed: Boolean = false init { - synchronized(inProgressTasks) { inProgressTasks.put(handle, this) } + inProgressTasks.put(handle, this) } fun startTaskWithMethodChannel( @NonNull channel: MethodChannel, - @NonNull identifier: String + @NonNull identifier: String, + onCancelled: (String) -> Unit ): TaskStateChannelStreamHandler { storageTask = when (type) { @@ -51,7 +52,8 @@ private constructor( FlutterFirebaseStorageTaskType.DOWNLOAD -> reference.getFile(fileUri!!) } - return TaskStateChannelStreamHandler(this, reference.storage, storageTask as Any, identifier) + return TaskStateChannelStreamHandler( + this, reference.storage, storageTask as Any, identifier, onCancelled) } fun getSnapshot(): Any = storageTask.snapshot @@ -88,12 +90,10 @@ private constructor( if (destroyed) return destroyed = true - synchronized(inProgressTasks) { - if (storageTask.isInProgress || storageTask.isPaused) { - storageTask.cancel() - } - inProgressTasks.remove(handle) + if (storageTask.isInProgress || storageTask.isPaused) { + storageTask.cancel() } + inProgressTasks.remove(handle, this) synchronized(cancelSyncObject) { cancelSyncObject.notifyAll() } synchronized(pauseSyncObject) { pauseSyncObject.notifyAll() } @@ -101,37 +101,22 @@ private constructor( } companion object { - val inProgressTasks: SparseArray = SparseArray() - - @JvmStatic - fun getInProgressTaskForHandle(handle: Int): FlutterFirebaseStorageTask? { - synchronized(inProgressTasks) { - return inProgressTasks.get(handle) - } - } - - @JvmStatic - fun cancelInProgressTasks() { - synchronized(inProgressTasks) { - val tasks = ArrayList(inProgressTasks.size()) - for (i in 0 until inProgressTasks.size()) { - val task: FlutterFirebaseStorageTask? = inProgressTasks.valueAt(i) - task?.let { tasks.add(it) } - } - inProgressTasks.clear() - tasks.forEach { it.destroy() } - } - } - @JvmStatic fun uploadBytes( handle: Int, reference: StorageReference, data: ByteArray, - metadata: StorageMetadata? + metadata: StorageMetadata?, + inProgressTasks: FlutterFirebaseStorageTaskRegistry ): FlutterFirebaseStorageTask { return FlutterFirebaseStorageTask( - FlutterFirebaseStorageTaskType.BYTES, handle, reference, data, null, metadata) + FlutterFirebaseStorageTaskType.BYTES, + handle, + reference, + data, + null, + metadata, + inProgressTasks) } @JvmStatic @@ -139,17 +124,25 @@ private constructor( handle: Int, reference: StorageReference, fileUri: Uri, - metadata: StorageMetadata? + metadata: StorageMetadata?, + inProgressTasks: FlutterFirebaseStorageTaskRegistry ): FlutterFirebaseStorageTask { return FlutterFirebaseStorageTask( - FlutterFirebaseStorageTaskType.FILE, handle, reference, null, fileUri, metadata) + FlutterFirebaseStorageTaskType.FILE, + handle, + reference, + null, + fileUri, + metadata, + inProgressTasks) } @JvmStatic fun downloadFile( handle: Int, reference: StorageReference, - file: File + file: File, + inProgressTasks: FlutterFirebaseStorageTaskRegistry ): FlutterFirebaseStorageTask { return FlutterFirebaseStorageTask( FlutterFirebaseStorageTaskType.DOWNLOAD, @@ -157,7 +150,8 @@ private constructor( reference, null, Uri.fromFile(file), - null) + null, + inProgressTasks) } @JvmStatic diff --git a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistry.kt b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistry.kt new file mode 100644 index 000000000000..31ecbeec3721 --- /dev/null +++ b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistry.kt @@ -0,0 +1,35 @@ +// Copyright 2026, the Chromium project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. +package io.flutter.plugins.firebase.storage + +/** + * The in-progress tasks of one plugin instance, keyed by their Dart handle. + * + * Handles come from a per-isolate counter, so each Flutter engine needs its own registry: with a + * shared one, engines overwrite each other's tasks, and detaching one engine cancels them all. + */ +internal class FlutterFirebaseStorageTaskRegistry { + private val tasks = HashMap() + + @Synchronized + fun put(handle: Int, task: T) { + tasks[handle] = task + } + + @Synchronized fun get(handle: Int): T? = tasks[handle] + + /** Removes [task], unless [handle] now belongs to another task. */ + @Synchronized + fun remove(handle: Int, task: T) { + if (tasks[handle] === task) tasks.remove(handle) + } + + /** Removes and returns all tasks. */ + @Synchronized + fun removeAll(): List { + val all = ArrayList(tasks.values) + tasks.clear() + return all + } +} diff --git a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt index 396b525d116b..b27dbc8888f1 100644 --- a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt +++ b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt @@ -16,22 +16,26 @@ internal class TaskStateChannelStreamHandler( private val flutterTask: FlutterFirebaseStorageTask, private val androidStorage: FirebaseStorage, task: Any, - private val identifier: String + private val identifier: String, + private val onCancelled: (String) -> Unit ) : StreamHandler { private val androidTask: StorageTask<*> = task as StorageTask<*> + @Volatile private var events: EventSink? = null + private val TASK_STATE_NAME = "taskState" private val TASK_APP_NAME = "appName" private val TASK_SNAPSHOT = "snapshot" private val TASK_ERROR = "error" override fun onListen(arguments: Any?, events: EventSink) { + this.events = events androidTask.addOnProgressListener { taskSnapshot -> if (flutterTask.isDestroyed()) return@addOnProgressListener val event = getTaskEventMap(taskSnapshot, null) event[TASK_STATE_NAME] = InternalStorageTaskState.RUNNING.raw - events.success(event) + this.events?.success(event) flutterTask.notifyResumeObjects() } @@ -39,7 +43,7 @@ internal class TaskStateChannelStreamHandler( if (flutterTask.isDestroyed()) return@addOnPausedListener val event = getTaskEventMap(taskSnapshot, null) event[TASK_STATE_NAME] = InternalStorageTaskState.PAUSED.raw - events.success(event) + this.events?.success(event) flutterTask.notifyPauseObjects() } @@ -47,7 +51,7 @@ internal class TaskStateChannelStreamHandler( if (flutterTask.isDestroyed()) return@addOnSuccessListener val event = getTaskEventMap(taskSnapshot, null) event[TASK_STATE_NAME] = InternalStorageTaskState.SUCCESS.raw - events.success(event) + this.events?.success(event) flutterTask.destroy() } @@ -61,7 +65,7 @@ internal class TaskStateChannelStreamHandler( syntheticException["message"] = FlutterFirebaseStorageException.getMessage(StorageException.ERROR_CANCELED) event[TASK_ERROR] = syntheticException - events.success(event) + this.events?.success(event) flutterTask.notifyCancelObjects() flutterTask.destroy() } @@ -70,22 +74,16 @@ internal class TaskStateChannelStreamHandler( if (flutterTask.isDestroyed()) return@addOnFailureListener val event = getTaskEventMap(null, exception) event[TASK_STATE_NAME] = InternalStorageTaskState.ERROR.raw - events.success(event) + this.events?.success(event) flutterTask.destroy() } } override fun onCancel(arguments: Any?) { - if (!androidTask.isCanceled) androidTask.cancel() - if (!flutterTask.isDestroyed()) flutterTask.destroy() - val eventChannel = FlutterFirebaseStoragePlugin.eventChannels[identifier] - if (eventChannel != null) { - eventChannel.setStreamHandler(null) - FlutterFirebaseStoragePlugin.eventChannels.remove(identifier) - } - if (FlutterFirebaseStoragePlugin.streamHandlers[identifier] != null) { - FlutterFirebaseStoragePlugin.streamHandlers.remove(identifier) - } + // Only stop sending events, as on iOS. The task keeps running, can still be paused, resumed or + // cancelled by its handle, and the listeners above still release it once it ends. + events = null + onCancelled(identifier) } private fun getTaskEventMap( diff --git a/packages/firebase_storage/firebase_storage/android/src/test/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistryTest.kt b/packages/firebase_storage/firebase_storage/android/src/test/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistryTest.kt new file mode 100644 index 000000000000..61698b75f840 --- /dev/null +++ b/packages/firebase_storage/firebase_storage/android/src/test/kotlin/io/flutter/plugins/firebase/storage/FlutterFirebaseStorageTaskRegistryTest.kt @@ -0,0 +1,65 @@ +// Copyright 2026, the Chromium project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. +package io.flutter.plugins.firebase.storage + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.junit.runners.JUnit4 + +@RunWith(JUnit4::class) +class FlutterFirebaseStorageTaskRegistryTest { + private class FakeTask(val name: String) + + @Test + fun sameHandleInTwoEnginesDoesNotCollide() { + // Every engine's Dart isolate numbers its tasks from 0. + val mainEngine = FlutterFirebaseStorageTaskRegistry() + val backgroundEngine = FlutterFirebaseStorageTaskRegistry() + val mainTask = FakeTask("main") + val backgroundTask = FakeTask("background") + + mainEngine.put(0, mainTask) + backgroundEngine.put(0, backgroundTask) + + assertSame(mainTask, mainEngine.get(0)) + assertSame(backgroundTask, backgroundEngine.get(0)) + } + + @Test + fun removeAllOnlyAffectsItsOwnEngine() { + val mainEngine = FlutterFirebaseStorageTaskRegistry() + val backgroundEngine = FlutterFirebaseStorageTaskRegistry() + val mainTask = FakeTask("main") + val backgroundTask = FakeTask("background") + mainEngine.put(0, mainTask) + backgroundEngine.put(0, backgroundTask) + + // What onDetachedFromEngine does for the detaching engine. + assertEquals(listOf(backgroundTask), backgroundEngine.removeAll()) + + assertNull(backgroundEngine.get(0)) + assertSame(mainTask, mainEngine.get(0)) + assertTrue(backgroundEngine.removeAll().isEmpty()) + } + + @Test + fun removeKeepsANewerTaskWithTheSameHandle() { + val registry = FlutterFirebaseStorageTaskRegistry() + val oldTask = FakeTask("old") + val newTask = FakeTask("new") + registry.put(0, oldTask) + // A hot restart resets the Dart handle counter. + registry.put(0, newTask) + + registry.remove(0, oldTask) + assertSame(newTask, registry.get(0)) + + registry.remove(0, newTask) + assertNull(registry.get(0)) + } +} diff --git a/packages/firebase_storage/firebase_storage/example/android/app/src/main/kotlin/io/flutter/plugins/firebasestorageexample/MainActivity.kt b/packages/firebase_storage/firebase_storage/example/android/app/src/main/kotlin/io/flutter/plugins/firebasestorageexample/MainActivity.kt index 4e41f73c96ae..a6c75ae2c426 100644 --- a/packages/firebase_storage/firebase_storage/example/android/app/src/main/kotlin/io/flutter/plugins/firebasestorageexample/MainActivity.kt +++ b/packages/firebase_storage/firebase_storage/example/android/app/src/main/kotlin/io/flutter/plugins/firebasestorageexample/MainActivity.kt @@ -1,5 +1,28 @@ package io.flutter.plugins.firebasestorageexample import io.flutter.embedding.android.FlutterActivity +import io.flutter.embedding.engine.FlutterEngine +import io.flutter.plugin.common.MethodChannel +import io.flutter.plugins.firebase.storage.FlutterFirebaseStoragePlugin -class MainActivity : FlutterActivity() +class MainActivity : FlutterActivity() { + override fun configureFlutterEngine(flutterEngine: FlutterEngine) { + super.configureFlutterEngine(flutterEngine) + + // Test-only channel used by the integration tests. + MethodChannel(flutterEngine.dartExecutor.binaryMessenger, "tests/engines") + .setMethodCallHandler { call, result -> + when (call.method) { + // Attaches firebase_storage to a second engine, then destroys that + // engine, as a background engine does when its work is done. + "attachAndDetachSecondEngine" -> { + val engine = FlutterEngine(applicationContext, null, false) + engine.plugins.add(FlutterFirebaseStoragePlugin()) + engine.destroy() + result.success(null) + } + else -> result.notImplemented() + } + } + } +} diff --git a/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart b/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart index 203f9fd355e8..f7b1d5a49d42 100644 --- a/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart +++ b/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart @@ -10,6 +10,7 @@ import 'package:firebase_core/firebase_core.dart'; import 'package:firebase_core_platform_interface/firebase_core_platform_interface.dart'; import 'package:firebase_storage/firebase_storage.dart'; import 'package:flutter/foundation.dart'; +import 'package:flutter/services.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:firebase_storage_example/firebase_options.dart'; @@ -465,6 +466,48 @@ void setupTaskTests() { expect(streamError, isA()); }, ); + + test('completes the task after its only `snapshotEvents` listener is ' + 'cancelled', () async { + const size = 5000000; + final ref = storage.ref('flutter-tests/listener-cancel.txt'); + final task = ref.putString('A' * size); + + // For example, a progress dialog that is closed mid-upload, before + // the app awaits the task. + await task.snapshotEvents.first; + await Future.delayed(const Duration(milliseconds: 100)); + + final snapshot = await task.timeout(_completerTimeout); + expect(snapshot.state, TaskState.success); + expect(task.snapshot.state, TaskState.success); + expect((await ref.getMetadata()).size, size); + }); }); + + test('detaching firebase_storage from another engine does not affect tasks ' + 'of this engine', () async { + final task = storage + .ref('flutter-tests/engine-detach.txt') + .putString('A' * 20000000); + final started = Completer(); + final subscription = task.snapshotEvents.listen((_) { + if (!started.isCompleted) started.complete(); + }, onError: (_) {}); + + try { + await started.future.timeout(_completerTimeout); + // Background engines (e.g. workmanager) attach their own plugin + // instances and detach them when their work is done. + await const MethodChannel( + 'tests/engines', + ).invokeMethod('attachAndDetachSecondEngine'); + + final snapshot = await task.timeout(_completerTimeout); + expect(snapshot.state, TaskState.success); + } finally { + await subscription.cancel(); + } + }, skip: kIsWeb || defaultTargetPlatform != TargetPlatform.android); }); } diff --git a/packages/firebase_storage/firebase_storage_platform_interface/lib/src/method_channel/method_channel_task.dart b/packages/firebase_storage/firebase_storage_platform_interface/lib/src/method_channel/method_channel_task.dart index 23bb0d5a872b..98bc64803fea 100644 --- a/packages/firebase_storage/firebase_storage_platform_interface/lib/src/method_channel/method_channel_task.dart +++ b/packages/firebase_storage/firebase_storage_platform_interface/lib/src/method_channel/method_channel_task.dart @@ -23,7 +23,9 @@ abstract class MethodChannelTask extends TaskPlatform { MethodChannelTask(this._handle, this.storage, String path, this._initialTask) : super() { Stream mapNativeStream() async* { - final observerId = await _initialTask; + final observerId = await _initialTask.catchError( + convertPlatformException, + ); // Raw metadata of the last native snapshot, reused by the error snapshot. Object? lastMetadata; @@ -66,10 +68,7 @@ abstract class MethodChannelTask extends TaskPlatform { } _exception = exception; _completer?.completeError(exception); - if (_userListening) { - // If the user is listening to the stream, yield the error. Otherwise, it results in an unhandled exception. - yield* Stream.error(exception); - } + yield* Stream.error(exception); break; } @@ -115,9 +114,10 @@ abstract class MethodChannelTask extends TaskPlatform { } } - _stream = mapNativeStream().asBroadcastStream( - onListen: (sub) => sub.resume(), - onCancel: (sub) => sub.cancel(), + _nativeStream = mapNativeStream(); + _controller = StreamController.broadcast( + sync: true, + onListen: _listenToNativeStream, ); // Keep reference to whether the initial "start" task has completed. @@ -128,8 +128,6 @@ abstract class MethodChannelTask extends TaskPlatform { }); } - bool _userListening = false; - /// FirebaseApp pigeon instance static InternalStorageFirebaseApp pigeonFirebaseApp( FirebaseStoragePlatform storage, @@ -162,7 +160,11 @@ abstract class MethodChannelTask extends TaskPlatform { Completer? _completer; - late Stream _stream; + late final Stream _nativeStream; + + bool _listeningToNativeStream = false; + + late final StreamController _controller; Future _initialTask; @@ -173,12 +175,44 @@ abstract class MethodChannelTask extends TaskPlatform { late TaskSnapshotPlatform _snapshot; - @override - Stream get snapshotEvents { - _userListening = true; - return _stream; + // A single internal subscription drives [snapshot] and [onComplete], so + // cancelling a user's [snapshotEvents] subscription does not cancel the + // native event stream (which also cancels the task on Android). + void _listenToNativeStream() { + if (_listeningToNativeStream) return; + _listeningToNativeStream = true; + _nativeStream.listen( + _controller.add, + onError: (Object error, StackTrace stackTrace) { + if (!_didComplete) { + _didComplete = true; + _exception = error; + _completer?.completeError(error, stackTrace); + } + _controller.addError(error, stackTrace); + }, + onDone: () { + if (!_didComplete) { + // The native stream closed without a final state. Fail instead of + // leaving `await task` pending forever. + final exception = FirebaseException( + plugin: 'firebase_storage', + code: 'unknown', + message: 'The task event stream closed before the task finished.', + ); + _didComplete = true; + _exception = exception; + _completer?.completeError(exception); + _controller.addError(exception); + } + _controller.close(); + }, + ); } + @override + Stream get snapshotEvents => _controller.stream; + @override TaskSnapshotPlatform get snapshot => _snapshot; @@ -189,14 +223,8 @@ abstract class MethodChannelTask extends TaskPlatform { } else if (_didComplete && _exception != null) { return catchFuturePlatformException(_exception!, StackTrace.current); } else { - // Call _stream.last to trigger the stream initialization, in case it hasn't been. - // ignore: unawaited_futures - _stream.last.catchError((_) { - // We ignore the exception here, stream exceptions are propagated in the stream handler above - // This causes unhandled exceptions when task.listen & onComplete are used together. - return Future.value(snapshot); - }); _completer ??= Completer(); + _listenToNativeStream(); return _completer!.future; } } diff --git a/packages/firebase_storage/firebase_storage_platform_interface/test/method_channel_tests/method_channel_task_test.dart b/packages/firebase_storage/firebase_storage_platform_interface/test/method_channel_tests/method_channel_task_test.dart index 9d7230359ff7..09b70d7dd8b7 100644 --- a/packages/firebase_storage/firebase_storage_platform_interface/test/method_channel_tests/method_channel_task_test.dart +++ b/packages/firebase_storage/firebase_storage_platform_interface/test/method_channel_tests/method_channel_task_test.dart @@ -2,20 +2,45 @@ // for details. All rights reserved. Use of this source code is governed by a // BSD-style license that can be found in the LICENSE file. -import 'dart:typed_data'; +import 'dart:async'; import 'package:firebase_core/firebase_core.dart'; import 'package:firebase_storage_platform_interface/firebase_storage_platform_interface.dart'; import 'package:firebase_storage_platform_interface/src/method_channel/method_channel_firebase_storage.dart'; import 'package:firebase_storage_platform_interface/src/method_channel/method_channel_task.dart'; import 'package:firebase_storage_platform_interface/src/pigeon/messages.pigeon.dart'; +import 'package:flutter/services.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:mockito/mockito.dart'; import '../mock.dart'; import '../pigeon/test_api.dart'; -const _observerId = 'put-data-observer'; +const String _observerId = 'task-observer'; +const Duration _hangTimeout = Duration(seconds: 2); + +class _TestTask extends MethodChannelTask { + _TestTask(FirebaseStoragePlatform storage, [Future? initialTask]) + : super(0, storage, '/foo', initialTask ?? Future.value(_observerId)); +} + +Map _event(TaskState state, int bytesTransferred) => { + 'taskState': state.index, + 'appName': '[DEFAULT]', + 'snapshot': { + 'path': '/foo', + 'bytesTransferred': bytesTransferred, + 'totalBytes': 10, + }, +}; + +Map _errorEvent(String code) => { + 'taskState': TaskState.error.index, + 'appName': '[DEFAULT]', + 'error': {'code': code, 'message': 'error: $code'}, +}; + +const _putDataObserverId = 'put-data-observer'; class _MockFirebaseStorageHostApi extends Mock implements TestFirebaseStorageHostApi { @@ -27,7 +52,7 @@ class _MockFirebaseStorageHostApi extends Mock InternalSettableMetadata settableMetaData, int handle, ) async { - return _observerId; + return _putDataObserverId; } } @@ -35,69 +60,274 @@ void main() { setupFirebaseStorageMocks(); late FirebaseStoragePlatform storage; + late MockStreamHandlerEventSink nativeSink; + late int nativeListenCount; + late int nativeCancelCount; setUpAll(() async { final app = await Firebase.initializeApp(); - storage = MethodChannelFirebaseStorage(app: app, bucket: 'bucket-test'); + storage = MethodChannelFirebaseStorage(app: app, bucket: 'bucket'); }); setUp(() { - TestFirebaseStorageHostApi.setUp(_MockFirebaseStorageHostApi()); - }); - - tearDown(() { - TestFirebaseStorageHostApi.setUp(null); - TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger - .setMockStreamHandler( - MethodChannelFirebaseStorage.storageTaskChannel(_observerId), - null, - ); - }); - - test('keeps the last metadata when the task fails', () async { + nativeListenCount = 0; + nativeCancelCount = 0; TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger .setMockStreamHandler( MethodChannelFirebaseStorage.storageTaskChannel(_observerId), MockStreamHandler.inline( onListen: (arguments, events) { - events.success({ - 'taskState': TaskState.running.index, - 'snapshot': { - 'path': 'foo/bar', - 'bytesTransferred': 1, - 'totalBytes': 2, - 'metadata': {'contentType': 'text/plain'}, - }, - }); - events.success({ - 'taskState': TaskState.error.index, - 'error': { - 'code': 'unauthorized', - 'message': 'User is not authorized.', - }, - }); + nativeListenCount++; + nativeSink = events; + events.success(_event(TaskState.running, 1)); }, + // On Android, cancelling the native stream used to cancel the task. + onCancel: (arguments) => nativeCancelCount++, + ), + ); + }); + + group('$MethodChannelTask', () { + test( + 'cancelling the only snapshotEvents listener does not cancel the native ' + 'task stream', + () async { + final task = _TestTask(storage); + + final first = await task.snapshotEvents.first; + expect(first.state, TaskState.running); + await pumpEventQueue(); + + expect(nativeListenCount, 1); + expect(nativeCancelCount, 0); + }, + ); + + test( + 'onComplete resolves after the snapshotEvents listener was cancelled', + () async { + final task = _TestTask(storage); + await task.snapshotEvents.first; + await pumpEventQueue(); + + final onComplete = task.onComplete; + nativeSink.success(_event(TaskState.success, 10)); + + final snapshot = await onComplete.timeout(_hangTimeout); + expect(snapshot.state, TaskState.success); + expect(snapshot.bytesTransferred, 10); + expect(task.snapshot.state, TaskState.success); + expect(nativeListenCount, 1); + }, + ); + + test('a new snapshotEvents listener receives events after an earlier one ' + 'was cancelled', () async { + final task = _TestTask(storage); + await task.snapshotEvents.first; + await pumpEventQueue(); + + final events = task.snapshotEvents.toList(); + nativeSink.success(_event(TaskState.running, 5)); + nativeSink.success(_event(TaskState.success, 10)); + + final states = (await events.timeout( + _hangTimeout, + )).map((s) => s.state).toList(); + expect(states, [TaskState.running, TaskState.success]); + }); + + test('onComplete completes with an error if the native stream closes ' + 'before a final state', () async { + final task = _TestTask(storage); + final onComplete = task.onComplete; + await pumpEventQueue(); + + nativeSink.endOfStream(); + + await expectLater( + onComplete.timeout(_hangTimeout), + throwsA( + isA().having( + (e) => e.plugin, + 'plugin', + 'firebase_storage', ), + ), + ); + }); + + test('snapshotEvents emits an error and closes if the native stream closes ' + 'before a final state', () async { + final task = _TestTask(storage); + final errors = []; + final done = Completer(); + task.snapshotEvents.listen( + (_) {}, + onError: errors.add, + onDone: done.complete, + ); + await pumpEventQueue(); + + nativeSink.endOfStream(); + + await done.future.timeout(_hangTimeout); + expect(errors, [isA()]); + }); + + test( + 'onComplete completes with an error if the task fails to start', + () async { + final initialTask = Completer(); + final task = _TestTask(storage, initialTask.future); + final onComplete = task.onComplete; + + initialTask.completeError( + PlatformException(code: 'unknown', message: 'failed'), ); - final task = MethodChannelPutTask( - nextMockHandleId, - storage, - 'foo/bar', - Uint8List.fromList([1, 2]), - null, + await expectLater( + onComplete.timeout(_hangTimeout), + throwsA( + isA() + .having((e) => e.code, 'code', 'unknown') + .having((e) => e.message, 'message', 'failed'), + ), + ); + expect(nativeListenCount, 0); + }, ); - await expectLater( - task.snapshotEvents, - emitsInOrder([ - isA(), - emitsError(isA()), - ]), + test('an error is delivered to snapshotEvents and onComplete', () async { + final task = _TestTask(storage); + final streamErrors = []; + final done = Completer(); + task.snapshotEvents.listen( + (_) {}, + onError: streamErrors.add, + onDone: done.complete, + ); + final onComplete = task.onComplete; + await pumpEventQueue(); + + nativeSink.success(_errorEvent('unauthorized')); + + await expectLater( + onComplete, + throwsA( + isA().having( + (e) => e.code, + 'code', + 'unauthorized', + ), + ), + ); + await done.future.timeout(_hangTimeout); + expect(streamErrors, [ + isA().having((e) => e.code, 'code', 'unauthorized'), + ]); + expect(task.snapshot.state, TaskState.error); + await expectLater( + task.onComplete, + throwsA( + isA().having( + (e) => e.code, + 'code', + 'unauthorized', + ), + ), + ); + }); + + test( + 'an error without a snapshotEvents listener only fails onComplete', + () async { + final task = _TestTask(storage); + final onComplete = task.onComplete; + await pumpEventQueue(); + + nativeSink.success(_errorEvent('canceled')); + + await expectLater( + onComplete, + throwsA( + isA().having((e) => e.code, 'code', 'canceled'), + ), + ); + }, ); - expect(task.snapshot.state, TaskState.error); - expect(task.snapshot.bytesTransferred, 1); - expect(task.snapshot.metadata?.contentType, 'text/plain'); + test('a listener added after completion receives done', () async { + final task = _TestTask(storage); + final onComplete = task.onComplete; + await pumpEventQueue(); + nativeSink.success(_event(TaskState.success, 10)); + await onComplete; + + await expectLater(task.snapshotEvents, emitsDone); + expect((await task.onComplete).state, TaskState.success); + }); + }); + + group('$MethodChannelPutTask', () { + setUp(() { + TestFirebaseStorageHostApi.setUp(_MockFirebaseStorageHostApi()); + }); + + tearDown(() { + TestFirebaseStorageHostApi.setUp(null); + TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger + .setMockStreamHandler( + MethodChannelFirebaseStorage.storageTaskChannel(_putDataObserverId), + null, + ); + }); + + test('keeps the last metadata when the task fails', () async { + TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger + .setMockStreamHandler( + MethodChannelFirebaseStorage.storageTaskChannel(_putDataObserverId), + MockStreamHandler.inline( + onListen: (arguments, events) { + events.success({ + 'taskState': TaskState.running.index, + 'snapshot': { + 'path': 'foo/bar', + 'bytesTransferred': 1, + 'totalBytes': 2, + 'metadata': {'contentType': 'text/plain'}, + }, + }); + events.success({ + 'taskState': TaskState.error.index, + 'error': { + 'code': 'unauthorized', + 'message': 'User is not authorized.', + }, + }); + }, + ), + ); + + final task = MethodChannelPutTask( + nextMockHandleId, + storage, + 'foo/bar', + Uint8List.fromList([1, 2]), + null, + ); + + await expectLater( + task.snapshotEvents, + emitsInOrder([ + isA(), + emitsError(isA()), + ]), + ); + + expect(task.snapshot.state, TaskState.error); + expect(task.snapshot.bytesTransferred, 1); + expect(task.snapshot.metadata?.contentType, 'text/plain'); + }); }); } From 001a1776700f8a5227db712ff8947cc4f7a3cac7 Mon Sep 17 00:00:00 2001 From: Guillaume Bernos Date: Wed, 21 Oct 2026 15:41:09 +0200 Subject: [PATCH 2/2] test(storage): bound every wait in the new task e2e tests, and halve the engine-detach upload - The listener-cancel e2e test now times out its `snapshotEvents.first` wait, like the rest of the file. - The engine-detach e2e test uploads 10 MB instead of 20 MB. On the Android emulator it is still uploading about 2 s after the second engine is destroyed, and it still fails on the old Android code. - The Android `onCancel` comment no longer claims parity with iOS. Android behaviour change (from this PR): disposing a `StreamBuilder`, a rebuild, or `await for ... break` no longer cancels the upload or download. The task keeps running until it ends or is cancelled by `task.cancel()`, as on iOS and web. A paused task the app forgets is kept for the life of the process. --- .../firebase/storage/TaskStateChannelStreamHandler.kt | 4 ++-- .../firebase_storage/example/integration_test/task_e2e.dart | 6 ++++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt index b27dbc8888f1..5923cbea4dfa 100644 --- a/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt +++ b/packages/firebase_storage/firebase_storage/android/src/main/kotlin/io/flutter/plugins/firebase/storage/TaskStateChannelStreamHandler.kt @@ -80,8 +80,8 @@ internal class TaskStateChannelStreamHandler( } override fun onCancel(arguments: Any?) { - // Only stop sending events, as on iOS. The task keeps running, can still be paused, resumed or - // cancelled by its handle, and the listeners above still release it once it ends. + // Only stop sending events: the task keeps running, can still be paused, resumed or cancelled + // by its handle, and the listeners above still release it once it ends. events = null onCancelled(identifier) } diff --git a/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart b/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart index f7b1d5a49d42..7d2f468a3fd7 100644 --- a/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart +++ b/packages/firebase_storage/firebase_storage/example/integration_test/task_e2e.dart @@ -475,7 +475,7 @@ void setupTaskTests() { // For example, a progress dialog that is closed mid-upload, before // the app awaits the task. - await task.snapshotEvents.first; + await task.snapshotEvents.first.timeout(_completerTimeout); await Future.delayed(const Duration(milliseconds: 100)); final snapshot = await task.timeout(_completerTimeout); @@ -487,9 +487,11 @@ void setupTaskTests() { test('detaching firebase_storage from another engine does not affect tasks ' 'of this engine', () async { + // Big enough to still be uploading when the other engine is + // destroyed, small enough to finish well within the timeout. final task = storage .ref('flutter-tests/engine-detach.txt') - .putString('A' * 20000000); + .putString('A' * 10000000); final started = Completer(); final subscription = task.snapshotEvents.listen((_) { if (!started.isCompleted) started.complete();