Couple of improvements on streaming

- Better code organization to make it easier to know which headers are
being set
- Process events from the buffer in case there's no empty line at the end

Author: rlazo

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

@@ -136,12 +136,15 @@ internal class PublisherStream(
           MediaType.parse("application/json"),
           JSONObject(mapOf("data" to serializer.encode(data))).toString()
         )
-      val requestBuilder =
-        Request.Builder().url(url).post(requestBody).header("Accept", "text/event-stream")
-      context?.authToken?.let { requestBuilder.header("Authorization", "Bearer $it") }
-      context?.instanceIdToken?.let { requestBuilder.header("Firebase-Instance-ID-Token", it) }
-      context?.appCheckToken?.let { requestBuilder.header("X-Firebase-AppCheck", it) }
-      val request = requestBuilder.build()
+      val request = Request.Builder().url(url).post(requestBody).apply {
+        header("Accept", "text/event-stream")
+        header("Content-Type", "application/json")
+        context?.apply {
+          authToken?.let { header("Authorization", "Bearer $it") }
+          instanceIdToken?.let { header("Firebase-Instance-ID-Token", it) }
+          appCheckToken?.let { header("X-Firebase-AppCheck", it) }
+        }
+      }.build()
       val call = configuredClient.newCall(request)
       activeCall = call
 
@@ -206,6 +209,9 @@ internal class PublisherStream(
             eventBuffer.append(dataChunk.trim()).append("\n")
           }
         }
+        if (eventBuffer.isNotEmpty()) {
+          processEvent(eventBuffer.toString())
+        }
       } catch (e: Exception) {
         notifyError(
           FirebaseFunctionsException(