dataconnect(change): Added internal logic: DataConnectGrpcRPCs.connect() to open a bidi streaming connection for realtime query updates

firebase-dataconnect/CHANGELOG.md

@@ -4,6 +4,8 @@
   ([#8081](https://github.com/firebase/firebase-android-sdk/pull/8081))
 - [changed] Fixed wasteful computation that is only for debug logging.
   ([#8126](https://github.com/firebase/firebase-android-sdk/pull/8126))
+- [changed] Internal implementation of backend connection for realtime query results.
+  ([#8141](https://github.com/firebase/firebase-android-sdk/pull/8141))
 
 # 17.2.2
 

firebase-dataconnect/androidTestutil/androidTestutil.gradle.kts

@@ -67,12 +67,10 @@ dependencies {
   implementation(libs.androidx.test.core)
   implementation(libs.androidx.test.junit)
   implementation(libs.auth0.jwt)
-  implementation(libs.grpc.api)
   implementation(libs.kotest.assertions)
   implementation(libs.kotest.property)
   implementation(libs.kotlinx.coroutines.core)
   implementation(libs.kotlinx.serialization.core)
   implementation(libs.kotlinx.serialization.json)
   implementation(libs.truth)
-  implementation(libs.turbine)
 }

firebase-dataconnect/androidTestutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/DataConnectIntegrationTestBase.kt

@@ -70,6 +70,12 @@ abstract class DataConnectIntegrationTestBase {
     }
   }
 
+  /**
+   * Convenience extension function on [Arb] that gets a non-edge-case value using [rs] for the
+   * randomness source.
+   */
+  fun <T> Arb<T>.sample(): T = sample(rs).value
+
   companion object {
     val testConnectorConfig: ConnectorConfig
       get() =

firebase-dataconnect/firebase-dataconnect.gradle.kts

@@ -134,6 +134,7 @@ dependencies {
   testImplementation(libs.mockk)
   testImplementation(libs.testonly.three.ten.abp)
   testImplementation(libs.robolectric)
+  testImplementation(libs.turbine)
 
   androidTestImplementation(project(":firebase-dataconnect:androidTestutil"))
   androidTestImplementation(project(":firebase-dataconnect:connectors"))

firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/ConnectRPCIntegrationTest.kt

@@ -505,12 +505,6 @@ class ConnectRPCIntegrationTest : DataConnectIntegrationTestBase() {
     val backend = DataConnectBackend.Custom("localhost:${inProcessServer.port}", sslEnabled = false)
     return connect(backend)
   }
-
-  /**
-   * Convenience extension function on [Arb] that gets a non-edge-case value using the [rs] property
-   * of the [DataConnectIntegrationTestBase] superclass for the randomness source.
-   */
-  private fun <T> Arb<T>.sample(): T = sample(rs).value
 }
 
 @OptIn(ExperimentalKotest::class)

firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/DataConnectGrpcRPCsConnectIntegrationTest.kt

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.firebase.dataconnect
+
+import app.cash.turbine.test
+import com.google.firebase.dataconnect.FirebaseDataConnect.CallerSdkType
+import com.google.firebase.dataconnect.core.DataConnectBidiConnectStream
+import com.google.firebase.dataconnect.core.DataConnectBidiConnectStream.ExecuteResponse
+import com.google.firebase.dataconnect.core.DataConnectGrpcRPCs
+import com.google.firebase.dataconnect.testutil.DataConnectIntegrationTestBase
+import com.google.firebase.dataconnect.testutil.property.arbitrary.dataConnect
+import com.google.firebase.dataconnect.testutil.property.arbitrary.pair
+import com.google.firebase.dataconnect.testutil.registerDataConnectKotestPrinters
+import com.google.firebase.dataconnect.testutil.schemas.RealtimeConnector
+import com.google.firebase.dataconnect.testutil.schemas.RealtimeConnector.GetStringByKeyQuery
+import com.google.firebase.dataconnect.util.ProtoUtil.decodeFromStruct
+import com.google.firebase.dataconnect.util.ProtoUtil.encodeToStruct
+import com.google.protobuf.Struct
+import google.firebase.dataconnect.proto.GraphqlError as GraphqlErrorProto
+import io.kotest.assertions.assertSoftly
+import io.kotest.assertions.withClue
+import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder
+import io.kotest.matchers.nulls.shouldBeNull
+import io.kotest.matchers.nulls.shouldNotBeNull
+import io.kotest.matchers.shouldBe
+import io.kotest.property.Arb
+import io.kotest.property.arbitrary.Codepoint
+import io.kotest.property.arbitrary.az
+import io.kotest.property.arbitrary.enum
+import io.kotest.property.arbitrary.map
+import io.kotest.property.arbitrary.string
+import io.kotest.property.arbitrary.uuid
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.test.runTest
+import kotlinx.serialization.serializer
+import org.junit.Before
+import org.junit.Test
+
+/** Integration tests for the [DataConnectGrpcRPCs.connect] method and its return value. */
+class DataConnectGrpcRPCsConnectIntegrationTest : DataConnectIntegrationTestBase() {
+
+  private val requestIdArb = Arb.dataConnect.requestId()
+  private val streamIdArb = Arb.dataConnect.streamId()
+  private val callerSdkTypeArb = Arb.enum<CallerSdkType>()
+  private val keyArb = Arb.uuid().map(RealtimeConnector::Key)
+  private val nameArb = Arb.string(size = 4, Codepoint.az()).map { "name_$it" }
+
+  @Before
+  fun registerPrinters() {
+    registerDataConnectKotestPrinters()
+  }
+
+  @Test
+  fun subscribeEmitsInitialResult() = runTest {
+    val connector = RealtimeConnector.getInstance(dataConnectFactory)
+    val dataConnectGrpcRPCs = connector.dataConnectGrpcRPCs
+    val key = keyArb.sample()
+
+    val stream = dataConnectGrpcRPCs.connect()
+    val executeResponseFlow: Flow<ExecuteResponse> =
+      stream.subscribe(
+        requestId = requestIdArb.sample(),
+        operationName = GetStringByKeyQuery.OPERATION_NAME,
+        variables = key.encodeToGetStringByKeyQueryVariables(),
+      )
+
+    executeResponseFlow.test {
+      awaitItem().shouldBeGetStringByKeyQueryResponse(expectedName = null)
+    }
+  }
+
+  @Test
+  fun subscribeEmitsUpdatedResult() = runTest {
+    val connector = RealtimeConnector.getInstance(dataConnectFactory)
+    val dataConnectGrpcRPCs = connector.dataConnectGrpcRPCs
+    val (name1, name2) = nameArb.pair().sample()
+    val key = connector.insertString(name = name1)
+
+    val stream = dataConnectGrpcRPCs.connect()
+    val executeResponseFlow: Flow<ExecuteResponse> =
+      stream.subscribe(
+        requestId = requestIdArb.sample(),
+        operationName = GetStringByKeyQuery.OPERATION_NAME,
+        variables = key.encodeToGetStringByKeyQueryVariables(),
+      )
+
+    executeResponseFlow.test {
+      awaitItem().shouldBeGetStringByKeyQueryResponse(expectedName = name1)
+      connector.updateString(key, name = name2)
+      awaitItem().shouldBeGetStringByKeyQueryResponse(expectedName = name2)
+      connector.deleteString(key)
+      awaitItem().shouldBeGetStringByKeyQueryResponse(expectedName = null)
+    }
+  }
+
+  private suspend fun DataConnectGrpcRPCs.connect(): DataConnectBidiConnectStream =
+    connect(
+      streamId = streamIdArb.sample(),
+      callerSdkType = callerSdkTypeArb.sample(),
+      authToken = null,
+      appCheckToken = null,
+    )
+}
+
+private fun GetStringByKeyQuery.Variables.encodeToStruct(): Struct =
+  encodeToStruct(this, serializer(), null)
+
+private fun RealtimeConnector.Key.encodeToGetStringByKeyQueryVariables(): Struct =
+  GetStringByKeyQuery.Variables(this).encodeToStruct()
+
+private fun Struct.decodeAsGetStringByKeyQueryData(): GetStringByKeyQuery.Data =
+  decodeFromStruct(this, serializer(), null)
+
+private fun ExecuteResponse.shouldBeGetStringByKeyQueryResponse(
+  expectedName: String?,
+  expectedErrors: List<GraphqlErrorProto> = emptyList(),
+) {
+  assertSoftly {
+    withClue("data") { data.shouldBeGetStringByKeyQueryData(expectedName) }
+    withClue("errors") { errors shouldContainExactlyInAnyOrder expectedErrors }
+  }
+}
+
+private fun Struct.shouldBeGetStringByKeyQueryData(expectedName: String?) {
+  val data: GetStringByKeyQuery.Data = decodeAsGetStringByKeyQueryData()
+  if (expectedName === null) {
+    data.item.shouldBeNull()
+  } else {
+    data.item.shouldNotBeNull() shouldBe GetStringByKeyQuery.Data.Item(expectedName)
+  }
+}

firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/testutil/FirebaseDataConnectInternalExts.kt

@@ -24,3 +24,6 @@ suspend fun FirebaseDataConnect.awaitAuthReady() =
 
 suspend fun FirebaseDataConnect.awaitAppCheckReady() =
   (this as FirebaseDataConnectInternal).awaitAppCheckReady()
+
+internal val FirebaseDataConnect.dataConnectGrpcRPCs
+  get() = (this as FirebaseDataConnectInternal).grpcRPCs

firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/testutil/schemas/RealtimeConnector.kt

@@ -18,6 +18,7 @@ package com.google.firebase.dataconnect.testutil.schemas
 
 import com.google.firebase.dataconnect.ConnectorConfig
 import com.google.firebase.dataconnect.FirebaseDataConnect
+import com.google.firebase.dataconnect.core.DataConnectGrpcRPCs
 import com.google.firebase.dataconnect.core.FirebaseDataConnectInternal
 import com.google.firebase.dataconnect.serializers.UUIDSerializer
 import com.google.firebase.dataconnect.testutil.DataConnectBackend
@@ -30,6 +31,9 @@ class RealtimeConnector private constructor(dataConnectInternal: FirebaseDataCon
 
   val dataConnect: FirebaseDataConnect = dataConnectInternal
 
+  internal val dataConnectGrpcRPCs: DataConnectGrpcRPCs by
+    lazy(LazyThreadSafetyMode.PUBLICATION) { dataConnectInternal.grpcRPCs }
+
   val resourceName: String = dataConnectInternal.connectorResourceName
 
   val getStringByKey = GetStringByKeyQuery(this)
@@ -57,7 +61,10 @@ class RealtimeConnector private constructor(dataConnectInternal: FirebaseDataCon
     fun queryRef(variables: Variables) =
       connector.dataConnect.query(OPERATION_NAME, variables, serializer<Data>(), serializer())
 
-    @Serializable data class Variables(val key: Key)
+    @Serializable
+    data class Variables(val key: Key) {
+      constructor(id: UUID) : this(Key(id))
+    }
 
     @Serializable
     data class Data(val item: Item?) {