Skip to content

Commit edf508f

Browse files
wu-huiwu-hui
committed
Hide realtime pipeline
1 parent 545ac38 commit edf508f

File tree

18 files changed

+94
-49
lines changed

18 files changed

+94
-49
lines changed

‎firebase-firestore/api.txt‎

Lines changed: 0 additions & 21 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,19 +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 String canonicalId();
622-
method public com.google.firebase.firestore.RealtimePipeline limit(int limit);
623-
method public kotlinx.coroutines.flow.Flow<com.google.firebase.firestore.RealtimePipelineSnapshot> snapshots();
624-
method public kotlinx.coroutines.flow.Flow<com.google.firebase.firestore.RealtimePipelineSnapshot> snapshots(com.google.firebase.firestore.RealtimePipelineOptions options);
625-
method public com.google.firebase.firestore.RealtimePipeline sort(com.google.firebase.firestore.pipeline.Ordering order, com.google.firebase.firestore.pipeline.Ordering... additionalOrders);
626-
method public com.google.firebase.firestore.RealtimePipeline where(com.google.firebase.firestore.pipeline.BooleanExpression condition);
627-
}
628-
629615
public final class RealtimePipelineOptions {
630616
ctor public RealtimePipelineOptions();
631617
method public com.google.firebase.firestore.RealtimePipelineOptions withMetadataChanges(com.google.firebase.firestore.MetadataChanges metadataChanges);
@@ -646,13 +632,6 @@ package com.google.firebase.firestore {
646632
property public final java.util.List<com.google.firebase.firestore.PipelineResult> results;
647633
}
648634

649-
public final class RealtimePipelineSource {
650-
method public com.google.firebase.firestore.RealtimePipeline collection(com.google.firebase.firestore.CollectionReference ref);
651-
method public com.google.firebase.firestore.RealtimePipeline collection(String path);
652-
method public com.google.firebase.firestore.RealtimePipeline collectionGroup(String collectionId);
653-
method public com.google.firebase.firestore.RealtimePipeline convertFrom(com.google.firebase.firestore.Query query);
654-
}
655-
656635
@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 {
657636
}
658637

‎firebase-firestore/gradle.properties‎

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

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

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

341+
@Test
342+
fun testExpressionEvaluateToErrorInWhere() = runBlocking {
343+
val pipeline =
344+
db.realtimePipeline().collection(collRef.path).where(field("published").add(Long.MAX_VALUE).lessThan(2000))
345+
val options =
346+
RealtimePipelineOptions()
347+
.withMetadataChanges(MetadataChanges.INCLUDE)
348+
.withSource(ListenSource.CACHE)
349+
val channel = Channel<RealtimePipelineSnapshot>(Channel.UNLIMITED)
350+
val job = launch { pipeline.snapshots(options).collect { snapshot -> channel.send(snapshot) } }
351+
352+
val firstSnapshot = channel.receive()
353+
assertThat(firstSnapshot.metadata.isConsistentBetweenListeners).isFalse()
354+
assertThat(firstSnapshot.results).hasSize(0)
355+
356+
waitFor(db.disableNetwork())
357+
waitFor(db.enableNetwork())
358+
359+
val nextSnapshot = withTimeoutOrNull(500) { channel.receive() }
360+
assertThat(nextSnapshot).isNull()
361+
362+
job.cancel()
363+
}
364+
365+
@Test
366+
fun testExpressionEvaluateToErrorInSort() = runBlocking {
367+
val pipeline =
368+
db.realtimePipeline().collection(collRef.path).sort(field("published").add(Long.MAX_VALUE - 1950).descending())
369+
val options =
370+
RealtimePipelineOptions()
371+
.withMetadataChanges(MetadataChanges.INCLUDE)
372+
.withSource(ListenSource.CACHE)
373+
val channel = Channel<RealtimePipelineSnapshot>(Channel.UNLIMITED)
374+
val job = launch { pipeline.snapshots(options).collect { snapshot -> channel.send(snapshot) } }
375+
376+
val firstSnapshot = channel.receive()
377+
assertThat(firstSnapshot.metadata.isConsistentBetweenListeners).isFalse()
378+
assertThat(firstSnapshot.results).hasSize(0)
379+
380+
waitFor(db.disableNetwork())
381+
waitFor(db.enableNetwork())
382+
383+
val nextSnapshot = withTimeoutOrNull(500) { channel.receive() }
384+
assertThat(nextSnapshot).isNull()
385+
386+
job.cancel()
387+
}
388+
341389
@Test
342390
fun testCanListenToMetadataOnlyChanges() = runBlocking {
343391
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
@@ -911,7 +911,7 @@ public PipelineSource pipeline() {
911911
* @return {@code RealtimePipelineSource} for this Firestore instance.
912912
*/
913913
@NonNull
914-
public RealtimePipelineSource realtimePipeline() {
914+
RealtimePipelineSource realtimePipeline() {
915915
clientProvider.ensureConfigured();
916916
return new RealtimePipelineSource(this);
917917
}

‎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
@@ -48,7 +48,8 @@ import kotlinx.coroutines.channels.awaitClose
4848
import kotlinx.coroutines.flow.Flow
4949
import kotlinx.coroutines.flow.callbackFlow
5050

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

130-
class RealtimePipeline
131+
internal class RealtimePipeline
131132
internal constructor(
132133
// This is nullable because RealtimePipeline is also created from deserialization from persistent
133134
// 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
@@ -243,7 +243,9 @@ public DocumentChanges computeDocChanges(
243243
candidates.add((MutableDocument) doc);
244244
}
245245
List<MutableDocument> results =
246-
this.query.pipeline().evaluate$com_google_firebase_firebase_firestore(candidates);
246+
this.query
247+
.pipeline$com_google_firebase_firebase_firestore()
248+
.evaluate$com_google_firebase_firebase_firestore(candidates);
247249
DocumentSet newResults = DocumentSet.emptySet(query.comparator());
248250
for (MutableDocument doc : results) {
249251
newResults = newResults.add(doc);
@@ -501,7 +503,8 @@ private static int changeTypeOrder(DocumentViewChange change) {
501503
@Nullable
502504
private static Long getLimit(QueryOrPipeline query) {
503505
if (query.isPipeline()) {
504-
Integer limit = getLastEffectiveLimit(query.pipeline());
506+
Integer limit =
507+
getLastEffectiveLimit(query.pipeline$com_google_firebase_firebase_firestore());
505508
if (limit == null) {
506509
return null;
507510
}

‎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)