Skip to content

Commit 087d386

Browse files
wu-huiwu-hui
committed
Hide realtime pipeline
1 parent b6152cd commit 087d386

File tree

19 files changed

+103
-50
lines changed

19 files changed

+103
-50
lines changed

‎firebase-firestore/api.txt‎

Lines changed: 0 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -207,7 +207,6 @@ package com.google.firebase.firestore {
207207
method public com.google.firebase.firestore.LoadBundleTask loadBundle(java.io.InputStream);
208208
method public com.google.firebase.firestore.LoadBundleTask loadBundle(java.nio.ByteBuffer);
209209
method public com.google.firebase.firestore.PipelineSource pipeline();
210-
method public com.google.firebase.firestore.RealtimePipelineSource realtimePipeline();
211210
method public com.google.android.gms.tasks.Task<java.lang.Void!> runBatch(com.google.firebase.firestore.WriteBatch.Function);
212211
method public <TResult> com.google.android.gms.tasks.Task<TResult!> runTransaction(com.google.firebase.firestore.Transaction.Function<TResult!>);
213212
method public <TResult> com.google.android.gms.tasks.Task<TResult!> runTransaction(com.google.firebase.firestore.TransactionOptions, com.google.firebase.firestore.Transaction.Function<TResult!>);
@@ -613,18 +612,6 @@ package com.google.firebase.firestore {
613612
method public <T> java.util.List<T!> toObjects(Class<T!>, com.google.firebase.firestore.DocumentSnapshot.ServerTimestampBehavior);
614613
}
615614

616-
public final class RealtimePipeline {
617-
method public com.google.firebase.firestore.ListenerRegistration addSnapshotListener(com.google.firebase.firestore.EventListener<com.google.firebase.firestore.RealtimePipelineSnapshot> listener);
618-
method public com.google.firebase.firestore.ListenerRegistration addSnapshotListener(com.google.firebase.firestore.RealtimePipelineOptions options, com.google.firebase.firestore.EventListener<com.google.firebase.firestore.RealtimePipelineSnapshot> listener);
619-
method public com.google.firebase.firestore.ListenerRegistration addSnapshotListener(java.util.concurrent.Executor executor, com.google.firebase.firestore.EventListener<com.google.firebase.firestore.RealtimePipelineSnapshot> listener);
620-
method public com.google.firebase.firestore.ListenerRegistration addSnapshotListener(java.util.concurrent.Executor executor, com.google.firebase.firestore.RealtimePipelineOptions options, com.google.firebase.firestore.EventListener<com.google.firebase.firestore.RealtimePipelineSnapshot> listener);
621-
method public com.google.firebase.firestore.RealtimePipeline limit(int limit);
622-
method public kotlinx.coroutines.flow.Flow<com.google.firebase.firestore.RealtimePipelineSnapshot> snapshots();
623-
method public kotlinx.coroutines.flow.Flow<com.google.firebase.firestore.RealtimePipelineSnapshot> snapshots(com.google.firebase.firestore.RealtimePipelineOptions options);
624-
method public com.google.firebase.firestore.RealtimePipeline sort(com.google.firebase.firestore.pipeline.Ordering order, com.google.firebase.firestore.pipeline.Ordering... additionalOrders);
625-
method public com.google.firebase.firestore.RealtimePipeline where(com.google.firebase.firestore.pipeline.BooleanExpression condition);
626-
}
627-
628615
public final class RealtimePipelineOptions {
629616
ctor public RealtimePipelineOptions();
630617
method public com.google.firebase.firestore.RealtimePipelineOptions withMetadataChanges(com.google.firebase.firestore.MetadataChanges metadataChanges);
@@ -645,13 +632,6 @@ package com.google.firebase.firestore {
645632
property public final java.util.List<com.google.firebase.firestore.PipelineResult> results;
646633
}
647634

648-
public final class RealtimePipelineSource {
649-
method public com.google.firebase.firestore.RealtimePipeline collection(com.google.firebase.firestore.CollectionReference ref);
650-
method public com.google.firebase.firestore.RealtimePipeline collection(String path);
651-
method public com.google.firebase.firestore.RealtimePipeline collectionGroup(String collectionId);
652-
method public com.google.firebase.firestore.RealtimePipeline convertFrom(com.google.firebase.firestore.Query query);
653-
}
654-
655635
@java.lang.annotation.Retention(java.lang.annotation.RetentionPolicy.RUNTIME) @java.lang.annotation.Target({java.lang.annotation.ElementType.METHOD, java.lang.annotation.ElementType.FIELD}) public @interface ServerTimestamp {
656636
}
657637

‎firebase-firestore/gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
1-
version=26.0.3
1+
version=99.0.0-pipeline.preview.1
22
latestReleasedVersion=26.0.2

‎firebase-firestore/src/androidTest/java/com/google/firebase/firestore/RealtimePipelineTest.kt‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,60 @@ class RealtimePipelineTest {
338338
job.cancel()
339339
}
340340

341+
@Test
342+
fun testExpressionEvaluateToErrorInWhere() = runBlocking {
343+
val pipeline =
344+
db
345+
.realtimePipeline()
346+
.collection(collRef.path)
347+
.where(field("published").add(Long.MAX_VALUE).lessThan(2000))
348+
val options =
349+
RealtimePipelineOptions()
350+
.withMetadataChanges(MetadataChanges.INCLUDE)
351+
.withSource(ListenSource.CACHE)
352+
val channel = Channel<RealtimePipelineSnapshot>(Channel.UNLIMITED)
353+
val job = launch { pipeline.snapshots(options).collect { snapshot -> channel.send(snapshot) } }
354+
355+
val firstSnapshot = channel.receive()
356+
assertThat(firstSnapshot.metadata.isConsistentBetweenListeners).isFalse()
357+
assertThat(firstSnapshot.results).hasSize(0)
358+
359+
waitFor(db.disableNetwork())
360+
waitFor(db.enableNetwork())
361+
362+
val nextSnapshot = withTimeoutOrNull(500) { channel.receive() }
363+
assertThat(nextSnapshot).isNull()
364+
365+
job.cancel()
366+
}
367+
368+
@Test
369+
fun testExpressionEvaluateToErrorInSort() = runBlocking {
370+
val pipeline =
371+
db
372+
.realtimePipeline()
373+
.collection(collRef.path)
374+
.sort(field("published").add(Long.MAX_VALUE - 1950).descending())
375+
val options =
376+
RealtimePipelineOptions()
377+
.withMetadataChanges(MetadataChanges.INCLUDE)
378+
.withSource(ListenSource.CACHE)
379+
val channel = Channel<RealtimePipelineSnapshot>(Channel.UNLIMITED)
380+
val job = launch { pipeline.snapshots(options).collect { snapshot -> channel.send(snapshot) } }
381+
382+
val firstSnapshot = channel.receive()
383+
assertThat(firstSnapshot.metadata.isConsistentBetweenListeners).isFalse()
384+
assertThat(firstSnapshot.results).hasSize(0)
385+
386+
waitFor(db.disableNetwork())
387+
waitFor(db.enableNetwork())
388+
389+
val nextSnapshot = withTimeoutOrNull(500) { channel.receive() }
390+
assertThat(nextSnapshot).isNull()
391+
392+
job.cancel()
393+
}
394+
341395
@Test
342396
fun testCanListenToMetadataOnlyChanges() = runBlocking {
343397
val pipeline =

‎firebase-firestore/src/androidTest/java/com/google/firebase/firestore/testutil/IntegrationTestUtil.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ public enum TargetBackend {
100100

101101
// Set this to the desired enum value to change the target backend when running tests locally.
102102
// Note: DO NOT change this variable except for local testing.
103-
private static final TargetBackend backendForLocalTesting = TargetBackend.EMULATOR;
103+
private static final TargetBackend backendForLocalTesting = TargetBackend.PROD;
104104

105105
private static final TargetBackend backend = getTargetBackend();
106106
private static final String EMULATOR_HOST = "10.0.2.2";

‎firebase-firestore/src/main/java/com/google/firebase/firestore/FirebaseFirestore.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -931,7 +931,7 @@ public PipelineSource pipeline() {
931931
* @return {@code RealtimePipelineSource} for this Firestore instance.
932932
*/
933933
@NonNull
934-
public RealtimePipelineSource realtimePipeline() {
934+
RealtimePipelineSource realtimePipeline() {
935935
clientProvider.ensureConfigured();
936936
return new RealtimePipelineSource(this);
937937
}

‎firebase-firestore/src/main/java/com/google/firebase/firestore/RealtimePipeline.kt‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,8 @@ import kotlinx.coroutines.channels.awaitClose
4949
import kotlinx.coroutines.flow.Flow
5050
import kotlinx.coroutines.flow.callbackFlow
5151

52-
class RealtimePipelineSource internal constructor(private val firestore: FirebaseFirestore) {
52+
internal class RealtimePipelineSource
53+
internal constructor(private val firestore: FirebaseFirestore) {
5354
/**
5455
* Convert the given Query into an equivalent Pipeline.
5556
*
@@ -128,7 +129,7 @@ class RealtimePipelineSource internal constructor(private val firestore: Firebas
128129
)
129130
}
130131

131-
class RealtimePipeline
132+
internal class RealtimePipeline
132133
internal constructor(
133134
// This is nullable because RealtimePipeline is also created from deserialization from persistent
134135
// cache. In that case, it is only used to facilitate remote store requests, and this field is

‎firebase-firestore/src/main/java/com/google/firebase/firestore/core/PipelineUtil.kt‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,8 @@ import com.google.firebase.firestore.util.Assert.hardAssert
3232

3333
/** A class that wraps either a Query or a RealtimePipeline. */
3434
sealed class QueryOrPipeline {
35-
data class QueryWrapper(val query: Query) : QueryOrPipeline()
36-
data class PipelineWrapper(val pipeline: RealtimePipeline) : QueryOrPipeline()
35+
internal data class QueryWrapper(internal val query: Query) : QueryOrPipeline()
36+
internal data class PipelineWrapper(internal val pipeline: RealtimePipeline) : QueryOrPipeline()
3737

3838
val isQuery: Boolean
3939
get() = this is QueryWrapper
@@ -45,7 +45,7 @@ sealed class QueryOrPipeline {
4545
return (this as QueryWrapper).query
4646
}
4747

48-
fun pipeline(): RealtimePipeline {
48+
internal fun pipeline(): RealtimePipeline {
4949
return (this as PipelineWrapper).pipeline
5050
}
5151

@@ -102,7 +102,7 @@ sealed class QueryOrPipeline {
102102
/** A class that wraps either a Target or a RealtimePipeline. */
103103
sealed class TargetOrPipeline {
104104
data class TargetWrapper(val target: Target) : TargetOrPipeline()
105-
data class PipelineWrapper(val pipeline: RealtimePipeline) : TargetOrPipeline()
105+
internal data class PipelineWrapper(val pipeline: RealtimePipeline) : TargetOrPipeline()
106106

107107
val isTarget: Boolean
108108
get() = this is TargetWrapper
@@ -114,7 +114,7 @@ sealed class TargetOrPipeline {
114114
return (this as TargetWrapper).target
115115
}
116116

117-
fun pipeline(): RealtimePipeline {
117+
internal fun pipeline(): RealtimePipeline {
118118
return (this as PipelineWrapper).pipeline
119119
}
120120

@@ -178,15 +178,15 @@ enum class PipelineSourceType {
178178
}
179179

180180
// Determines the flavor of the given pipeline based on its stages.
181-
fun getPipelineFlavor(pipeline: RealtimePipeline): PipelineFlavor {
181+
internal fun getPipelineFlavor(pipeline: RealtimePipeline): PipelineFlavor {
182182
// For now, it is only possible to construct RealtimePipeline that is kExact.
183183
// PORTING NOTE: the typescript implementation support other flavors already,
184184
// despite not being used. We can port that later.
185185
return PipelineFlavor.EXACT
186186
}
187187

188188
// Determines the source type of the given pipeline based on its first stage.
189-
fun getPipelineSourceType(pipeline: RealtimePipeline): PipelineSourceType {
189+
internal fun getPipelineSourceType(pipeline: RealtimePipeline): PipelineSourceType {
190190
hardAssert(
191191
!pipeline.stages.isEmpty(),
192192
"Pipeline must have at least one stage to determine its source.",
@@ -202,7 +202,7 @@ fun getPipelineSourceType(pipeline: RealtimePipeline): PipelineSourceType {
202202

203203
// Retrieves the collection group ID if the pipeline's source is a collection
204204
// group.
205-
fun getPipelineCollectionGroup(pipeline: RealtimePipeline): String? {
205+
internal fun getPipelineCollectionGroup(pipeline: RealtimePipeline): String? {
206206
if (getPipelineSourceType(pipeline) == PipelineSourceType.COLLECTION_GROUP) {
207207
hardAssert(
208208
!pipeline.stages.isEmpty(),
@@ -217,7 +217,7 @@ fun getPipelineCollectionGroup(pipeline: RealtimePipeline): String? {
217217
}
218218

219219
// Retrieves the collection path if the pipeline's source is a collection.
220-
fun getPipelineCollection(pipeline: RealtimePipeline): String? {
220+
internal fun getPipelineCollection(pipeline: RealtimePipeline): String? {
221221
if (getPipelineSourceType(pipeline) == PipelineSourceType.COLLECTION) {
222222
hardAssert(
223223
!pipeline.stages.isEmpty(),
@@ -232,7 +232,7 @@ fun getPipelineCollection(pipeline: RealtimePipeline): String? {
232232
}
233233

234234
// Retrieves the document pathes if the pipeline's source is a document source.
235-
fun getPipelineDocuments(pipeline: RealtimePipeline): Array<out String>? {
235+
internal fun getPipelineDocuments(pipeline: RealtimePipeline): Array<out String>? {
236236
if (getPipelineSourceType(pipeline) == PipelineSourceType.DOCUMENTS) {
237237
hardAssert(
238238
!pipeline.stages.isEmpty(),
@@ -248,7 +248,7 @@ fun getPipelineDocuments(pipeline: RealtimePipeline): Array<out String>? {
248248

249249
// Creates a new pipeline by replacing CollectionGroupSource stages with
250250
// CollectionSource stages using the provided path.
251-
fun asCollectionPipelineAtPath(
251+
internal fun asCollectionPipelineAtPath(
252252
pipeline: RealtimePipeline,
253253
path: ResourcePath,
254254
): RealtimePipeline {
@@ -271,7 +271,7 @@ fun asCollectionPipelineAtPath(
271271
)
272272
}
273273

274-
fun getLastEffectiveLimit(pipeline: RealtimePipeline): Int? {
274+
internal fun getLastEffectiveLimit(pipeline: RealtimePipeline): Int? {
275275
for (stagePtr in pipeline.rewrittenStages.asReversed()) {
276276
// Check if the stage is a LimitStage
277277
if (stagePtr is LimitStage) {

‎firebase-firestore/src/main/java/com/google/firebase/firestore/core/QueryListener.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,9 @@ public QueryListener(
5656
if (query.isPipeline()) {
5757
this.query =
5858
new QueryOrPipeline.PipelineWrapper(
59-
query.pipeline().withListenOptions$com_google_firebase_firebase_firestore(options));
59+
query
60+
.pipeline$com_google_firebase_firebase_firestore()
61+
.withListenOptions$com_google_firebase_firebase_firestore(options));
6062
} else {
6163
this.query = query;
6264
}

‎firebase-firestore/src/main/java/com/google/firebase/firestore/core/View.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -245,7 +245,9 @@ public DocumentChanges computeDocChanges(
245245
candidates.add((MutableDocument) doc);
246246
}
247247
List<MutableDocument> results =
248-
this.query.pipeline().evaluate$com_google_firebase_firebase_firestore(candidates);
248+
this.query
249+
.pipeline$com_google_firebase_firebase_firestore()
250+
.evaluate$com_google_firebase_firebase_firestore(candidates);
249251
DocumentSet newResults = DocumentSet.emptySet(query.comparator());
250252
for (MutableDocument doc : results) {
251253
newResults = newResults.add(doc);
@@ -505,7 +507,8 @@ private static int changeTypeOrder(DocumentViewChange change) {
505507
@Nullable
506508
private static Long getLimit(QueryOrPipeline query) {
507509
if (query.isPipeline()) {
508-
Integer limit = getLastEffectiveLimit(query.pipeline());
510+
Integer limit =
511+
getLastEffectiveLimit(query.pipeline$com_google_firebase_firebase_firestore());
509512
if (limit == null) {
510513
return null;
511514
}

‎firebase-firestore/src/main/java/com/google/firebase/firestore/local/LocalDocumentsView.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -400,7 +400,7 @@ private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingCollection
400400

401401
private ImmutableSortedMap<DocumentKey, Document> getDocumentsMatchingPipeline(
402402
QueryOrPipeline queryOrPipeline, IndexOffset offset, @Nullable QueryContext context) {
403-
RealtimePipeline pipeline = queryOrPipeline.pipeline();
403+
RealtimePipeline pipeline = queryOrPipeline.pipeline$com_google_firebase_firebase_firestore();
404404
if (getPipelineSourceType(pipeline) == PipelineSourceType.COLLECTION_GROUP) {
405405
String collectionGroup = getPipelineCollectionGroup(pipeline);
406406
hardAssert(

0 commit comments

Comments
 (0)