update

Author: VinayGuthal

firebase-vertexai/api.txt

@@ -627,7 +627,7 @@ package com.google.firebase.vertexai.type {
 
   public final class LiveSession {
     method public suspend Object? close(kotlin.coroutines.Continuation<? super kotlin.Unit>);
-    method public suspend Object? receive(java.util.List<com.google.firebase.vertexai.type.ContentModality> outputModalities, kotlin.coroutines.Continuation<? super kotlinx.coroutines.flow.Flow<? extends com.google.firebase.vertexai.type.LiveContentResponse>>);
+    method public kotlinx.coroutines.flow.Flow<com.google.firebase.vertexai.type.LiveContentResponse> receive(java.util.List<com.google.firebase.vertexai.type.ContentModality> outputModalities);
     method public kotlinx.coroutines.flow.Flow<java.util.List<com.google.firebase.vertexai.type.FunctionCallPart>> receiveAudioConversationFunctionCalls();
     method public suspend Object? send(com.google.firebase.vertexai.type.Content content, kotlin.coroutines.Continuation<? super kotlin.Unit>);
     method public suspend Object? send(String text, kotlin.coroutines.Continuation<? super kotlin.Unit>);

firebase-vertexai/src/main/kotlin/com/google/firebase/vertexai/LiveGenerativeModel.kt

@@ -81,7 +81,7 @@ internal constructor(
 
   /**
    * Returns a LiveSession object using which you could send/receive messages from the server
-   * @return LiveSession object created. Returns null if the object cannot be created.
+   * @return LiveSession object created
    * @throws [BidiServerHandshakeFailed] if the handshake with the server failed.
    */
   public suspend fun connect(): LiveSession {

firebase-vertexai/src/main/kotlin/com/google/firebase/vertexai/java/LiveSessionFutures.kt

@@ -31,9 +31,9 @@ import kotlinx.coroutines.reactive.asPublisher
 import org.reactivestreams.Publisher
 
 /**
- * Wrapper class providing Java compatible methods for [GenerativeModel].
+ * Wrapper class providing Java compatible methods for [LiveSession].
  *
- * @see [GenerativeModel]
+ * @see [LiveSession]
  */
 public abstract class LiveSessionFutures internal constructor() {
 
@@ -94,7 +94,7 @@ public abstract class LiveSessionFutures internal constructor() {
    */
   public abstract fun receive(
     outputModalities: List<ContentModality>
-  ): ListenableFuture<Publisher<LiveContentResponse>>
+  ): Publisher<LiveContentResponse>
 
   /**
    * Receives all function call responses from the server for the audio conversation feature..
@@ -106,10 +106,8 @@ public abstract class LiveSessionFutures internal constructor() {
 
   private class FuturesImpl(private val session: LiveSession) : LiveSessionFutures() {
 
-    override fun receive(
-      outputModalities: List<ContentModality>
-    ): ListenableFuture<Publisher<LiveContentResponse>> =
-      SuspendToFutureAdapter.launchFuture { session.receive(outputModalities).asPublisher() }
+    override fun receive(outputModalities: List<ContentModality>): Publisher<LiveContentResponse> =
+      session.receive(outputModalities).asPublisher()
 
     override fun receiveAudioConversationFunctionCalls(): Publisher<List<FunctionCallPart>> =
       session.receiveAudioConversationFunctionCalls().asPublisher()

firebase-vertexai/src/main/kotlin/com/google/firebase/vertexai/type/Exceptions.kt

@@ -175,7 +175,7 @@ public class QuotaExceededException
 internal constructor(message: String, cause: Throwable? = null) :
   FirebaseVertexAIException(message, cause)
 
-/** Bidi session already receiving. */
+/** Streaming session already receiving. */
 public class SessionAlreadyReceivingException :
   FirebaseVertexAIException(
     "This session is already receiving. Please call stopReceiving() before calling this again."

firebase-vertexai/src/main/kotlin/com/google/firebase/vertexai/type/LiveSession.kt

@@ -27,7 +27,6 @@ import io.ktor.websocket.readBytes
 import java.util.concurrent.ConcurrentLinkedQueue
 import kotlin.coroutines.CoroutineContext
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
@@ -156,7 +155,7 @@ internal constructor(
       receivedAudio.copyInto(audioBuffer, offset)
       offset += receivedAudio.size
       if (offset >= MIN_BUFFER_SIZE) {
-        sendMediaStream(listOf(MediaData( audioBuffer, "audio/pcm")))
+        sendMediaStream(listOf(MediaData(audioBuffer, "audio/pcm")))
         audioBuffer.fill(0)
         offset = 0
       }
@@ -248,13 +247,13 @@ internal constructor(
    *
    * @throws [SessionAlreadyReceivingException] when the session is already receiving.
    */
-  public suspend fun receive(outputModalities: List<ContentModality>): Flow<LiveContentResponse> {
+  public fun receive(outputModalities: List<ContentModality>): Flow<LiveContentResponse> {
     if (startedReceiving) {
       throw SessionAlreadyReceivingException()
     }
 
     val flowReceive = session!!.incoming.receiveAsFlow()
-    CoroutineScope(Dispatchers.IO).launch { flowReceive.collect { receiveChannel.send(it) } }
+    CoroutineScope(backgroundDispatcher).launch { flowReceive.collect { receiveChannel.send(it) } }
     return flow {
       startedReceiving = true
       while (true) {