fix(functions): use notifyError() instead of throwing

firebase-functions/src/androidTest/java/com/google/firebase/functions/StreamTests.kt

@@ -143,6 +143,26 @@ class StreamTests {
     assertThat(subscriber.throwable).isInstanceOf(FirebaseFunctionsException::class.java)
   }
 
+  @Test
+  fun nonExistentFunction_receivesError() = runBlocking {
+    val function =
+      functions.getHttpsCallable("nonexistentFunction").withTimeout(2000, TimeUnit.MILLISECONDS)
+    val subscriber = StreamSubscriber()
+
+    function.stream().subscribe(subscriber)
+
+    withTimeout(2000) {
+      while (subscriber.throwable == null) {
+        delay(100)
+      }
+    }
+
+    assertThat(subscriber.throwable).isNotNull()
+    assertThat(subscriber.throwable).isInstanceOf(FirebaseFunctionsException::class.java)
+    assertThat((subscriber.throwable as FirebaseFunctionsException).code)
+      .isEqualTo(FirebaseFunctionsException.Code.NOT_FOUND)
+  }
+
   @Test
   fun genStreamWeather_receivesWeatherForecasts() = runBlocking {
     val inputData = listOf(mapOf("name" to "Toronto"), mapOf("name" to "London"))

firebase-functions/src/main/java/com/google/firebase/functions/PublisherStream.kt

@@ -300,10 +300,12 @@ internal class PublisherStream(
     val errorMessage: String
     if (response.code() == 404 && response.header("Content-Type") == htmlContentType) {
       errorMessage = """URL not found. Raw response: ${response.body()?.string()}""".trimMargin()
-      throw FirebaseFunctionsException(
-        errorMessage,
-        FirebaseFunctionsException.Code.fromHttpStatus(response.code()),
-        null
+      notifyError(
+        FirebaseFunctionsException(
+          errorMessage,
+          FirebaseFunctionsException.Code.fromHttpStatus(response.code()),
+          null
+        )
       )
     }
 
@@ -313,16 +315,17 @@ internal class PublisherStream(
       val json = JSONObject(text)
       error = serializer.decode(json.opt("error"))
     } catch (e: Throwable) {
-      throw FirebaseFunctionsException(
-        "${e.message} Unexpected Response:\n$text ",
-        FirebaseFunctionsException.Code.INTERNAL,
-        e
+      notifyError(
+        FirebaseFunctionsException(
+          "${e.message} Unexpected Response:\n$text ",
+          FirebaseFunctionsException.Code.INTERNAL,
+          e
+        )
       )
+      return
     }
-    throw FirebaseFunctionsException(
-      error.toString(),
-      FirebaseFunctionsException.Code.INTERNAL,
-      error
+    notifyError(
+      FirebaseFunctionsException(error.toString(), FirebaseFunctionsException.Code.INTERNAL, error)
     )
   }
 }