diff --git a/firebase-storage/ktx/api.txt b/firebase-storage/ktx/api.txt
index 104fa68d25b..20f84128b45 100644
--- a/firebase-storage/ktx/api.txt
+++ b/firebase-storage/ktx/api.txt
@@ -15,11 +15,27 @@ package com.google.firebase.storage.ktx {
     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 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 TaskState<T> {
+  }
+
+  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 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;
+  }
+
 }
 
diff --git a/firebase-storage/ktx/ktx.gradle b/firebase-storage/ktx/ktx.gradle
index e45840aadf5..5669f37f715 100644
--- a/firebase-storage/ktx/ktx.gradle
+++ b/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"
diff --git a/firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/Storage.kt b/firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/Storage.kt
index d560b1232ad..915f5d6f85e 100644
--- a/firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/Storage.kt
+++ b/firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/Storage.kt
@@ -15,6 +15,7 @@
 package com.google.firebase.storage.ktx
 
 import androidx.annotation.Keep
+import com.google.android.gms.tasks.OnCompleteListener
 import com.google.firebase.FirebaseApp
 import com.google.firebase.components.Component
 import com.google.firebase.components.ComponentRegistrar
@@ -23,10 +24,19 @@ 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.StorageTask
+import com.google.firebase.storage.StorageTaskScheduler
 import com.google.firebase.storage.StreamDownloadTask
 import com.google.firebase.storage.UploadTask
+import kotlinx.coroutines.cancel
+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 +143,47 @@ operator fun ListResult.component2(): List<StorageReference> = prefixes
  */
 operator fun ListResult.component3(): String? = pageToken
 
+/**
+ * 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:
+ * [OnProgressListener], [OnPausedListener], [OnCompleteListener].
+ * - When the flow completes the listeners will be removed.
+ */
+val <T : StorageTask<T>.SnapshotBase> StorageTask<T>.taskState: Flow<TaskState<T>>
+    get() = callbackFlow {
+        val progressListener = OnProgressListener<T> { snapshot ->
+            StorageTaskScheduler.getInstance().scheduleCallback {
+                trySendBlocking(TaskState.InProgress(snapshot))
+            }
+        }
+        val pauseListener = OnPausedListener<T> { snapshot ->
+            StorageTaskScheduler.getInstance().scheduleCallback {
+                trySendBlocking(TaskState.Paused(snapshot))
+            }
+        }
+
+        // Only used to close or cancel the Flows, doesn't send any values
+        val completionListener = OnCompleteListener<T> { task ->
+            if (task.isSuccessful) {
+                close()
+            } else {
+                val exception = task.exception
+                cancel("Error getting the TaskState", exception)
+            }
+        }
+
+        addOnProgressListener(progressListener)
+        addOnPausedListener(pauseListener)
+        addOnCompleteListener(completionListener)
+
+        awaitClose {
+            removeOnProgressListener(progressListener)
+            removeOnPausedListener(pauseListener)
+            removeOnCompleteListener(completionListener)
+        }
+    }
+
 internal const val LIBRARY_NAME: String = "fire-stg-ktx"
 
 /** @suppress */
diff --git a/firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/TaskState.kt b/firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/TaskState.kt
new file mode 100644
index 00000000000..8b8454c8efb
--- /dev/null
+++ b/firebase-storage/ktx/src/main/kotlin/com/google/firebase/storage/ktx/TaskState.kt
@@ -0,0 +1,16 @@
+package com.google.firebase.storage.ktx
+
+/**
+ * Used to emit events about the progress of storage tasks.
+ */
+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) : TaskState<T>()
+
+    /**
+     * Called any time the upload/download is paused.
+     */
+    class Paused<T>(val snapshot: T) : TaskState<T>()
+}