storage-ktx: add callbackFlow for upload/download progress

firebase-storage/ktx/api.txt

@@ -14,12 +14,30 @@ 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 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 static final class StorageProgress.InProgress<T> extends com.google.firebase.storage.ktx.StorageProgress<T> {
+    ctor public StorageProgress.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);
+    method public T getSnapshot();
+    property public final T snapshot;
+  }
+
 }
 

firebase-storage/ktx/ktx.gradle

@@ -17,6 +17,8 @@ plugins {
     id 'kotlin-android'
 }
 
+group = "com.google.firebase"
+
 firebaseLibrary {
     releaseWith project(':firebase-storage')
     publishJavadoc = true
@@ -61,6 +63,7 @@ dependencies {
     implementation project(':firebase-storage')
     implementation 'androidx.annotation:annotation:1.1.0'
     implementation 'com.google.android.gms:play-services-tasks:18.0.1'
+    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutinesVersion"
 
     androidTestImplementation 'junit:junit:4.12'
     androidTestImplementation "com.google.truth:truth:$googleTruthVersion"

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

@@ -23,10 +23,17 @@ import com.google.firebase.platforminfo.LibraryVersionComponent
 import com.google.firebase.storage.FileDownloadTask
 import com.google.firebase.storage.FirebaseStorage
 import com.google.firebase.storage.ListResult
+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.StorageTaskScheduler
 import com.google.firebase.storage.StreamDownloadTask
 import com.google.firebase.storage.UploadTask
+import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.channels.trySendBlocking
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.callbackFlow
 
 /** Returns the [FirebaseStorage] instance of the default [FirebaseApp]. */
 val Firebase.storage: FirebaseStorage
@@ -133,6 +140,93 @@ 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].
+ *
+ * - When the returned flow starts being collected, it attaches the following listeners:
+ * [OnPausedListener], [OnProgressListener].
+ * - When the flow completes the listeners will be removed.
+ */
+val StreamDownloadTask.progress: Flow<StorageProgress<StreamDownloadTask.TaskSnapshot>>
+    get() = callbackFlow {
+        val progressListener = OnProgressListener<StreamDownloadTask.TaskSnapshot> { snapshot ->
+            StorageTaskScheduler.getInstance().scheduleCallback {
+                trySendBlocking(StorageProgress.InProgress(snapshot))
+            }
+        }
+        val pauseListener = OnPausedListener<StreamDownloadTask.TaskSnapshot> { 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))
+            }
+        }
+
+        addOnProgressListener(progressListener)
+        addOnPausedListener(pauseListener)
+
+        awaitClose {
+            removeOnProgressListener(progressListener)
+            removeOnPausedListener(pauseListener)
+        }
+    }
+
 internal const val LIBRARY_NAME: String = "fire-stg-ktx"
 
 /** @suppress */

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

@@ -0,0 +1,16 @@
+package com.google.firebase.storage.ktx
+
+/**
+ * Used to emit events about the progress of storage tasks.
+ */
+abstract class StorageProgress<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>()
+
+    /**
+     * Called any time the upload/download is paused.
+     */
+    class Paused<T>(val snapshot: T) : StorageProgress<T>()
+}