storage-ktx: add callbackFlow for upload/download progress

firebase-storage/ktx/api.txt

@@ -14,27 +14,25 @@ package com.google.firebase.storage.ktx {
     method @NonNull public static operator java.io.InputStream component3(@NonNull com.google.firebase.storage.StreamDownloadTask.TaskSnapshot);
     method @Nullable public static operator String component3(@NonNull com.google.firebase.storage.ListResult);
     method @Nullable public static operator android.net.Uri component4(@NonNull com.google.firebase.storage.UploadTask.TaskSnapshot);
-    method @NonNull public static kotlinx.coroutines.flow.Flow<com.google.firebase.storage.ktx.StorageProgress<com.google.firebase.storage.StreamDownloadTask.TaskSnapshot>> getProgress(@NonNull com.google.firebase.storage.StreamDownloadTask);
-    method @NonNull public static kotlinx.coroutines.flow.Flow<com.google.firebase.storage.ktx.StorageProgress<com.google.firebase.storage.FileDownloadTask.TaskSnapshot>> getProgress(@NonNull com.google.firebase.storage.FileDownloadTask);
-    method @NonNull public static kotlinx.coroutines.flow.Flow<com.google.firebase.storage.ktx.StorageProgress<com.google.firebase.storage.UploadTask.TaskSnapshot>> getProgress(@NonNull com.google.firebase.storage.UploadTask);
     method @NonNull public static com.google.firebase.storage.FirebaseStorage getStorage(@NonNull com.google.firebase.ktx.Firebase);
+    method @NonNull public static <T extends com.google.firebase.storage.StorageTask.SnapshotBase> kotlinx.coroutines.flow.Flow<com.google.firebase.storage.ktx.TaskState<T>> getTaskState(@NonNull com.google.firebase.storage.StorageTask<T>);
     method @NonNull public static com.google.firebase.storage.FirebaseStorage storage(@NonNull com.google.firebase.ktx.Firebase, @NonNull String url);
     method @NonNull public static com.google.firebase.storage.FirebaseStorage storage(@NonNull com.google.firebase.ktx.Firebase, @NonNull com.google.firebase.FirebaseApp app);
     method @NonNull public static com.google.firebase.storage.FirebaseStorage storage(@NonNull com.google.firebase.ktx.Firebase, @NonNull com.google.firebase.FirebaseApp app, @NonNull String url);
     method @NonNull public static com.google.firebase.storage.StorageMetadata storageMetadata(@NonNull kotlin.jvm.functions.Function1<? super com.google.firebase.storage.StorageMetadata.Builder,kotlin.Unit> init);
   }
 
-  public abstract class StorageProgress<T> {
+  public abstract class TaskState<T> {
   }
 
-  public static final class StorageProgress.InProgress<T> extends com.google.firebase.storage.ktx.StorageProgress<T> {
-    ctor public StorageProgress.InProgress(@Nullable T snapshot);
+  public static final class TaskState.InProgress<T> extends com.google.firebase.storage.ktx.TaskState<T> {
+    ctor public TaskState.InProgress(@Nullable T snapshot);
     method public T getSnapshot();
     property public final T snapshot;
   }
 
-  public static final class StorageProgress.Paused<T> extends com.google.firebase.storage.ktx.StorageProgress<T> {
-    ctor public StorageProgress.Paused(@Nullable T snapshot);
+  public static final class TaskState.Paused<T> extends com.google.firebase.storage.ktx.TaskState<T> {
+    ctor public TaskState.Paused(@Nullable T snapshot);
     method public T getSnapshot();
     property public final T snapshot;
   }

firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/Storage.kt

@@ -27,6 +27,7 @@ import com.google.firebase.storage.OnPausedListener
 import com.google.firebase.storage.OnProgressListener
 import com.google.firebase.storage.StorageMetadata
 import com.google.firebase.storage.StorageReference
+import com.google.firebase.storage.StorageTask
 import com.google.firebase.storage.StorageTaskScheduler
 import com.google.firebase.storage.StreamDownloadTask
 import com.google.firebase.storage.UploadTask
@@ -141,80 +142,22 @@ operator fun ListResult.component2(): List<StorageReference> = prefixes
 operator fun ListResult.component3(): String? = pageToken
 
 /**
- * Starts listening to this task's stream download progress and emits its values via a [Flow].
+ * Starts listening to this task's progress and emits its values via a [Flow].
  *
  * - When the returned flow starts being collected, it attaches the following listeners:
- * [OnPausedListener], [OnProgressListener].
+ * [OnProgressListener], [OnPausedListener].
  * - When the flow completes the listeners will be removed.
  */
-val StreamDownloadTask.progress: Flow<StorageProgress<StreamDownloadTask.TaskSnapshot>>
+val <T : StorageTask<T>.SnapshotBase> StorageTask<T>.taskState: Flow<TaskState<T>>
     get() = callbackFlow {
-        val progressListener = OnProgressListener<StreamDownloadTask.TaskSnapshot> { snapshot ->
+        val progressListener = OnProgressListener<T> { snapshot ->
             StorageTaskScheduler.getInstance().scheduleCallback {
-                trySendBlocking(StorageProgress.InProgress(snapshot))
+                trySendBlocking(TaskState.InProgress(snapshot))
             }
         }
-        val pauseListener = OnPausedListener<StreamDownloadTask.TaskSnapshot> { snapshot ->
+        val pauseListener = OnPausedListener<T> { snapshot ->
             StorageTaskScheduler.getInstance().scheduleCallback {
-                trySendBlocking(StorageProgress.Paused(snapshot))
-            }
-        }
-
-        addOnProgressListener(progressListener)
-        addOnPausedListener(pauseListener)
-
-        awaitClose {
-            removeOnProgressListener(progressListener)
-            removeOnPausedListener(pauseListener)
-        }
-    }
-
-/**
- * Starts listening to this task's file download progress and emits its values via a [Flow].
- *
- * - When the returned flow starts being collected, it attaches the following listeners:
- * [OnPausedListener], [OnProgressListener].
- * - When the flow completes the listeners will be removed.
- */
-val FileDownloadTask.progress: Flow<StorageProgress<FileDownloadTask.TaskSnapshot>>
-    get() = callbackFlow {
-        val progressListener = OnProgressListener<FileDownloadTask.TaskSnapshot> { snapshot ->
-            StorageTaskScheduler.getInstance().scheduleCallback {
-                trySendBlocking(StorageProgress.InProgress(snapshot))
-            }
-        }
-        val pauseListener = OnPausedListener<FileDownloadTask.TaskSnapshot> { snapshot ->
-            StorageTaskScheduler.getInstance().scheduleCallback {
-                trySendBlocking(StorageProgress.Paused(snapshot))
-            }
-        }
-
-        addOnProgressListener(progressListener)
-        addOnPausedListener(pauseListener)
-
-        awaitClose {
-            removeOnProgressListener(progressListener)
-            removeOnPausedListener(pauseListener)
-        }
-    }
-
-/**
- * Starts listening to this task's upload progress and emits its values via a [Flow].
- *
- * - When the returned flow starts being collected, it attaches the following listeners:
- * [OnPausedListener], [OnProgressListener].
- * - When the flow completes the listeners will be removed.
- */
-val UploadTask.progress: Flow<StorageProgress<UploadTask.TaskSnapshot>>
-    get() = callbackFlow {
-        val progressListener = OnProgressListener<UploadTask.TaskSnapshot> { snapshot ->
-            StorageTaskScheduler.getInstance().scheduleCallback {
-                trySendBlocking(StorageProgress.InProgress(snapshot))
-            }
-        }
-        val pauseListener = OnPausedListener<UploadTask.TaskSnapshot> { snapshot ->
-            StorageTaskScheduler.getInstance().scheduleCallback {
-                trySendBlocking(StorageProgress.Paused(snapshot))
+                trySendBlocking(TaskState.Paused(snapshot))
             }
         }
 

firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/StorageProgress.kt → firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/TaskState.kt

@@ -3,14 +3,14 @@ package com.google.firebase.storage.ktx
 /**
  * Used to emit events about the progress of storage tasks.
  */
-abstract class StorageProgress<T> private constructor() {
+abstract class TaskState<T> private constructor() {
     /**
      * Called periodically as data is transferred and can be used to populate an upload/download indicator.
      */
-    class InProgress<T>(val snapshot: T) : StorageProgress<T>()
+    class InProgress<T>(val snapshot: T) : TaskState<T>()
 
     /**
      * Called any time the upload/download is paused.
      */
-    class Paused<T>(val snapshot: T) : StorageProgress<T>()
+    class Paused<T>(val snapshot: T) : TaskState<T>()
 }