diff --git a/.github/workflows/android_unit_tests.yaml b/.github/workflows/android_unit_tests.yaml index 1ff80014f6a7..4f7db4f372d4 100644 --- a/.github/workflows/android_unit_tests.yaml +++ b/.github/workflows/android_unit_tests.yaml @@ -43,7 +43,11 @@ jobs: # Tasks are listed per package rather than using the root testDebugUnitTest # task, which would also build the Dart sources of the aggregate test app. # Add a task here when a package gains an android/src/test directory. - run: cd tests/android && ./gradlew :firebase_crashlytics:testDebugUnitTest + run: | + cd tests/android + ./gradlew \ + :cloud_firestore:testDebugUnitTest \ + :firebase_crashlytics:testDebugUnitTest - name: 'Upload test reports' if: failure() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a diff --git a/packages/cloud_firestore/cloud_firestore/android/build.gradle b/packages/cloud_firestore/cloud_firestore/android/build.gradle index 66d5b12c19c5..3d4bb93ce24e 100755 --- a/packages/cloud_firestore/cloud_firestore/android/build.gradle +++ b/packages/cloud_firestore/cloud_firestore/android/build.gradle @@ -62,6 +62,8 @@ android { api firebaseCoreProject implementation platform("com.google.firebase:firebase-bom:${getRootProjectExtOrCoreProperty("FirebaseSDKVersion", firebaseCoreProject)}") implementation 'com.google.firebase:firebase-firestore' + testImplementation 'junit:junit:4.13.2' + testImplementation 'org.mockito:mockito-core:5.14.2' } } diff --git a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkProtocol.java b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkProtocol.java new file mode 100644 index 000000000000..a8bdc8795432 --- /dev/null +++ b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkProtocol.java @@ -0,0 +1,83 @@ +/* + * Copyright 2026, the Chromium project authors. Please see the AUTHORS file + * for details. All rights reserved. Use of this source code is governed by a + * BSD-style license that can be found in the LICENSE file. + */ + +package io.flutter.plugins.firebase.firestore.streamhandler; + +import io.flutter.plugin.common.StandardMethodCodec; +import io.flutter.plugins.firebase.firestore.GeneratedAndroidFirebaseFirestore; +import java.nio.ByteBuffer; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** Wire messages for bounded Android query-snapshot transport. */ +final class QuerySnapshotChunkProtocol { + // Keep these values in sync with query_snapshot_chunk_assembler.dart. + private static final String CHUNK_MARKER_KEY = "firestoreQuerySnapshotChunk"; + private static final String SNAPSHOT_ID_KEY = "snapshotId"; + private static final String KIND_KEY = "kind"; + private static final String PAYLOAD_KEY = "payload"; + private static final String DOCUMENT_COUNT_KEY = "documentCount"; + private static final String DOCUMENT_CHANGE_COUNT_KEY = "documentChangeCount"; + + static final int START_KIND = 0; + static final int DOCUMENTS_KIND = 1; + static final int DOCUMENT_CHANGES_KIND = 2; + static final int END_KIND = 3; + + /** + * Keeps the transient platform-channel envelope far below the 18–31 MiB allocations observed in + * the crash report. A single Firestore document may exceed this target, but Firestore itself + * bounds that exceptional message to one 1 MiB document plus codec overhead. + */ + static final int MAX_ENCODED_ENVELOPE_BYTES = 512 * 1024; + + private static final StandardMethodCodec METHOD_CODEC = + new StandardMethodCodec(GeneratedAndroidFirebaseFirestore.PigeonCodec.INSTANCE); + + private QuerySnapshotChunkProtocol() {} + + static Map start( + long snapshotId, + int documentCount, + int documentChangeCount, + GeneratedAndroidFirebaseFirestore.InternalSnapshotMetadata metadata) { + Map message = base(snapshotId, START_KIND); + message.put(PAYLOAD_KEY, metadata); + message.put(DOCUMENT_COUNT_KEY, documentCount); + message.put(DOCUMENT_CHANGE_COUNT_KEY, documentChangeCount); + return message; + } + + static Map itemChunk(long snapshotId, int kind, List items) { + Map message = base(snapshotId, kind); + message.put(PAYLOAD_KEY, items); + return message; + } + + static Map end(long snapshotId) { + return base(snapshotId, END_KIND); + } + + static int encodedMessageSize(Object message) { + ByteBuffer encoded = + GeneratedAndroidFirebaseFirestore.PigeonCodec.INSTANCE.encodeMessage(message); + return encoded == null ? 0 : encoded.position(); + } + + static int encodedEnvelopeSize(Object message) { + ByteBuffer encoded = METHOD_CODEC.encodeSuccessEnvelope(message); + return encoded.position(); + } + + private static Map base(long snapshotId, int kind) { + Map message = new LinkedHashMap<>(); + message.put(CHUNK_MARKER_KEY, true); + message.put(SNAPSHOT_ID_KEY, snapshotId); + message.put(KIND_KEY, kind); + return message; + } +} diff --git a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunker.java b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunker.java new file mode 100644 index 000000000000..e8103afb4091 --- /dev/null +++ b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunker.java @@ -0,0 +1,78 @@ +/* + * Copyright 2026, the Chromium project authors. Please see the AUTHORS file + * for details. All rights reserved. Use of this source code is governed by a + * BSD-style license that can be found in the LICENSE file. + */ + +package io.flutter.plugins.firebase.firestore.streamhandler; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.function.Function; +import java.util.function.ToIntFunction; + +/** + * Incrementally converts and partitions snapshot items without retaining a converted copy of the + * complete query snapshot. + * + *

The inexpensive per-item size estimate chooses an initial boundary. The exact envelope sizer + * then removes tail items until the actual platform message fits. A single item is always allowed + * through even when it exceeds the target, because Firestore already bounds an individual document + * to 1 MiB and the stream must continue making progress. + */ +final class QuerySnapshotChunker { + private final Iterator source; + private final Function converter; + private final ToIntFunction itemSize; + private final int targetBytes; + private final ArrayDeque pending = new ArrayDeque<>(); + + QuerySnapshotChunker( + Iterator source, Function converter, ToIntFunction itemSize, int targetBytes) { + if (targetBytes <= 0) { + throw new IllegalArgumentException("targetBytes must be positive"); + } + this.source = source; + this.converter = converter; + this.itemSize = itemSize; + this.targetBytes = targetBytes; + } + + boolean hasNext() { + return !pending.isEmpty() || source.hasNext(); + } + + List nextChunk(ToIntFunction> exactEnvelopeSize) { + if (!hasNext()) { + throw new NoSuchElementException("No snapshot items remain"); + } + + ArrayList chunk = new ArrayList<>(); + int estimatedBytes = 0; + + while (hasNext()) { + T item = pending.isEmpty() ? converter.apply(source.next()) : pending.removeFirst(); + int encodedItemBytes = Math.max(0, itemSize.applyAsInt(item)); + + if (!chunk.isEmpty() && estimatedBytes + encodedItemBytes > targetBytes) { + pending.addFirst(item); + break; + } + + chunk.add(item); + estimatedBytes += encodedItemBytes; + if (estimatedBytes >= targetBytes) { + break; + } + } + + while (chunk.size() > 1 && exactEnvelopeSize.applyAsInt(chunk) > targetBytes) { + pending.addFirst(chunk.remove(chunk.size() - 1)); + } + + return chunk; + } +} diff --git a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandler.java b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandler.java index 8aade3f8a9c2..0c15fda26596 100644 --- a/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandler.java +++ b/packages/cloud_firestore/cloud_firestore/android/src/main/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandler.java @@ -10,23 +10,28 @@ import android.os.Handler; import android.os.Looper; +import com.google.firebase.firestore.DocumentChange; import com.google.firebase.firestore.DocumentSnapshot; import com.google.firebase.firestore.ListenSource; import com.google.firebase.firestore.ListenerRegistration; import com.google.firebase.firestore.MetadataChanges; import com.google.firebase.firestore.Query; +import com.google.firebase.firestore.QuerySnapshot; import com.google.firebase.firestore.SnapshotListenOptions; import io.flutter.plugin.common.EventChannel.EventSink; import io.flutter.plugin.common.EventChannel.StreamHandler; +import io.flutter.plugins.firebase.firestore.GeneratedAndroidFirebaseFirestore; import io.flutter.plugins.firebase.firestore.utils.ExceptionConverter; import io.flutter.plugins.firebase.firestore.utils.PigeonParser; +import java.util.ArrayDeque; +import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicLong; public class QuerySnapshotsStreamHandler implements StreamHandler { - ListenerRegistration listenerRegistration; - private final Handler mainHandler = new Handler(Looper.getMainLooper()); Query query; MetadataChanges metadataChanges; @@ -34,6 +39,12 @@ public class QuerySnapshotsStreamHandler implements StreamHandler { ListenSource source; Executor snapshotExecutor; + private final Executor eventExecutor; + private final AtomicLong nextSnapshotId = new AtomicLong(); + private final ArrayDeque pendingSnapshots = new ArrayDeque<>(); + private EventSink eventSink; + private boolean emissionActive; + private boolean cancelled; public QuerySnapshotsStreamHandler( Query query, @@ -41,51 +52,277 @@ public QuerySnapshotsStreamHandler( DocumentSnapshot.ServerTimestampBehavior serverTimestampBehavior, ListenSource source, Executor snapshotExecutor) { + this( + query, + includeMetadataChanges, + serverTimestampBehavior, + source, + snapshotExecutor, + mainThreadExecutor()); + } + + QuerySnapshotsStreamHandler( + Query query, + Boolean includeMetadataChanges, + DocumentSnapshot.ServerTimestampBehavior serverTimestampBehavior, + ListenSource source, + Executor snapshotExecutor, + Executor eventExecutor) { this.query = query; this.metadataChanges = includeMetadataChanges ? MetadataChanges.INCLUDE : MetadataChanges.EXCLUDE; this.serverTimestampBehavior = serverTimestampBehavior; this.source = source; this.snapshotExecutor = snapshotExecutor; + this.eventExecutor = eventExecutor; } @Override public void onListen(Object arguments, EventSink events) { + synchronized (this) { + cancelled = false; + emissionActive = false; + pendingSnapshots.clear(); + eventSink = events; + } + SnapshotListenOptions.Builder optionsBuilder = new SnapshotListenOptions.Builder(); optionsBuilder.setMetadataChanges(metadataChanges); optionsBuilder.setSource(source); optionsBuilder.setExecutor(snapshotExecutor); - listenerRegistration = + ListenerRegistration registration = query.addSnapshotListener( optionsBuilder.build(), (querySnapshot, exception) -> { if (exception != null) { - Map exceptionDetails = ExceptionConverter.createDetails(exception); - mainHandler.post( - () -> { - events.error(DEFAULT_ERROR_CODE, exception.getMessage(), exceptionDetails); - events.endOfStream(); - }); - - onCancel(null); + emitError(exception); } else { - // Emit the Pigeon object directly; the Pigeon-aware codec serializes - // nested `InternalDocumentSnapshot` / `InternalDocumentChange` / - // `InternalSnapshotMetadata` with their proper type codes. Pigeon 26 - // no longer flattens nested types via `.toList()`. - Object pigeonSnapshot = - PigeonParser.toPigeonQuerySnapshot(querySnapshot, serverTimestampBehavior); - mainHandler.post(() -> events.success(pigeonSnapshot)); + enqueueSnapshot(Objects.requireNonNull(querySnapshot)); } }); + + retainListenerRegistration(registration); + } + + void retainListenerRegistration(ListenerRegistration registration) { + boolean removeRegistration; + synchronized (this) { + removeRegistration = cancelled; + if (!removeRegistration) { + listenerRegistration = registration; + } + } + if (removeRegistration) { + registration.remove(); + } + } + + void emitSnapshotForTesting(QuerySnapshot snapshot, EventSink events) { + synchronized (this) { + if (eventSink != null && eventSink != events) { + throw new IllegalStateException("A different event sink is already active"); + } + cancelled = false; + eventSink = events; + } + enqueueSnapshot(snapshot); + } + + private void enqueueSnapshot(QuerySnapshot snapshot) { + boolean shouldStart; + synchronized (this) { + if (cancelled) { + return; + } + pendingSnapshots.addLast(snapshot); + shouldStart = !emissionActive; + if (shouldStart) { + emissionActive = true; + } + } + + if (shouldStart) { + emitNextSnapshot(); + } + } + + private void emitNextSnapshot() { + QuerySnapshot snapshot; + synchronized (this) { + if (cancelled) { + return; + } + snapshot = pendingSnapshots.pollFirst(); + if (snapshot == null) { + emissionActive = false; + return; + } + } + + try { + SnapshotEmission emission = new SnapshotEmission(nextSnapshotId.incrementAndGet(), snapshot); + postSuccess( + QuerySnapshotChunkProtocol.start( + emission.id, emission.documentCount, emission.documentChangeCount, emission.metadata), + () -> emitNextChunk(emission)); + } catch (Exception exception) { + emitError(exception); + } + } + + private void emitNextChunk(SnapshotEmission emission) { + if (isCancelled()) { + return; + } + + try { + if (emission.documents.hasNext()) { + List chunk = + emission.documents.nextChunk( + values -> + QuerySnapshotChunkProtocol.encodedEnvelopeSize( + QuerySnapshotChunkProtocol.itemChunk( + emission.id, QuerySnapshotChunkProtocol.DOCUMENTS_KIND, values))); + postSuccess( + QuerySnapshotChunkProtocol.itemChunk( + emission.id, QuerySnapshotChunkProtocol.DOCUMENTS_KIND, chunk), + () -> emitNextChunk(emission)); + return; + } + + if (emission.documentChanges.hasNext()) { + List chunk = + emission.documentChanges.nextChunk( + values -> + QuerySnapshotChunkProtocol.encodedEnvelopeSize( + QuerySnapshotChunkProtocol.itemChunk( + emission.id, + QuerySnapshotChunkProtocol.DOCUMENT_CHANGES_KIND, + values))); + postSuccess( + QuerySnapshotChunkProtocol.itemChunk( + emission.id, QuerySnapshotChunkProtocol.DOCUMENT_CHANGES_KIND, chunk), + () -> emitNextChunk(emission)); + return; + } + + postSuccess(QuerySnapshotChunkProtocol.end(emission.id), this::emitNextSnapshot); + } catch (Exception exception) { + emitError(exception); + } + } + + private void postSuccess(Object message, Runnable afterSuccess) { + EventSink sink; + synchronized (this) { + if (cancelled || eventSink == null) { + return; + } + sink = eventSink; + } + + eventExecutor.execute( + () -> { + if (isCancelled()) { + return; + } + try { + sink.success(message); + snapshotExecutor.execute(afterSuccess); + } catch (Exception exception) { + emitError(exception); + } + }); + } + + private void emitError(Exception exception) { + EventSink sink; + ListenerRegistration registration; + synchronized (this) { + if (cancelled) { + return; + } + cancelled = true; + pendingSnapshots.clear(); + emissionActive = false; + sink = eventSink; + eventSink = null; + registration = listenerRegistration; + listenerRegistration = null; + } + + if (registration != null) { + registration.remove(); + } + if (sink == null) { + return; + } + + Map exceptionDetails = ExceptionConverter.createDetails(exception); + eventExecutor.execute( + () -> { + sink.error(DEFAULT_ERROR_CODE, exception.getMessage(), exceptionDetails); + sink.endOfStream(); + }); + } + + private synchronized boolean isCancelled() { + return cancelled; } @Override public void onCancel(Object arguments) { - if (listenerRegistration != null) { - listenerRegistration.remove(); + ListenerRegistration registration; + synchronized (this) { + cancelled = true; + pendingSnapshots.clear(); + emissionActive = false; + eventSink = null; + registration = listenerRegistration; listenerRegistration = null; } + if (registration != null) { + registration.remove(); + } + } + + private static Executor mainThreadExecutor() { + Handler mainHandler = new Handler(Looper.getMainLooper()); + return command -> mainHandler.post(command); + } + + private final class SnapshotEmission { + final long id; + final int documentCount; + final int documentChangeCount; + final GeneratedAndroidFirebaseFirestore.InternalSnapshotMetadata metadata; + final QuerySnapshotChunker< + DocumentSnapshot, GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot> + documents; + final QuerySnapshotChunker< + DocumentChange, GeneratedAndroidFirebaseFirestore.InternalDocumentChange> + documentChanges; + + SnapshotEmission(long id, QuerySnapshot snapshot) { + this.id = id; + List nativeDocuments = snapshot.getDocuments(); + List nativeDocumentChanges = snapshot.getDocumentChanges(); + documentCount = nativeDocuments.size(); + documentChangeCount = nativeDocumentChanges.size(); + metadata = PigeonParser.toPigeonSnapshotMetadata(snapshot.getMetadata()); + documents = + new QuerySnapshotChunker<>( + nativeDocuments.iterator(), + value -> PigeonParser.toPigeonDocumentSnapshot(value, serverTimestampBehavior), + QuerySnapshotChunkProtocol::encodedMessageSize, + QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + documentChanges = + new QuerySnapshotChunker<>( + nativeDocumentChanges.iterator(), + value -> PigeonParser.toPigeonDocumentChange(value, serverTimestampBehavior), + QuerySnapshotChunkProtocol::encodedMessageSize, + QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + } } } diff --git a/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkProtocolTest.java b/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkProtocolTest.java new file mode 100644 index 000000000000..aea52e93b574 --- /dev/null +++ b/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkProtocolTest.java @@ -0,0 +1,166 @@ +/* + * Copyright 2026, the Chromium project authors. Please see the AUTHORS file + * for details. All rights reserved. Use of this source code is governed by a + * BSD-style license that can be found in the LICENSE file. + */ + +package io.flutter.plugins.firebase.firestore.streamhandler; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import io.flutter.plugins.firebase.firestore.GeneratedAndroidFirebaseFirestore; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.junit.Test; + +public class QuerySnapshotChunkProtocolTest { + @Test + public void documentChunkEnvelopesStayWithinTheTransportTarget() { + List documents = new ArrayList<>(); + String payload = String.join("", Collections.nCopies(2048, "abcdefgh")); + for (int index = 0; index < 100; index++) { + documents.add(document("projects/p/assets/" + index, payload)); + } + + QuerySnapshotChunker< + GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot, + GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot> + chunker = + new QuerySnapshotChunker<>( + documents.iterator(), + value -> value, + QuerySnapshotChunkProtocol::encodedMessageSize, + QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + List received = new ArrayList<>(); + + while (chunker.hasNext()) { + List chunk = + chunker.nextChunk( + values -> + QuerySnapshotChunkProtocol.encodedEnvelopeSize( + QuerySnapshotChunkProtocol.itemChunk( + 42, QuerySnapshotChunkProtocol.DOCUMENTS_KIND, values))); + Map message = + QuerySnapshotChunkProtocol.itemChunk( + 42, QuerySnapshotChunkProtocol.DOCUMENTS_KIND, chunk); + + assertTrue( + QuerySnapshotChunkProtocol.encodedEnvelopeSize(message) + <= QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + received.addAll(chunk); + } + + assertEquals(documents, received); + } + + @Test + public void oneOversizedDocumentStillProducesOneBoundedByFirestoreItemMessage() { + GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot document = + document( + "projects/p/assets/large", String.join("", Collections.nCopies(131072, "abcdefgh"))); + QuerySnapshotChunker< + GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot, + GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot> + chunker = + new QuerySnapshotChunker<>( + Collections.singletonList(document).iterator(), + value -> value, + QuerySnapshotChunkProtocol::encodedMessageSize, + QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + + List chunk = + chunker.nextChunk( + values -> + QuerySnapshotChunkProtocol.encodedEnvelopeSize( + QuerySnapshotChunkProtocol.itemChunk( + 1, QuerySnapshotChunkProtocol.DOCUMENTS_KIND, values))); + + assertEquals(Collections.singletonList(document), chunk); + assertTrue( + QuerySnapshotChunkProtocol.encodedEnvelopeSize( + QuerySnapshotChunkProtocol.itemChunk( + 1, QuerySnapshotChunkProtocol.DOCUMENTS_KIND, chunk)) + < 2 * 1024 * 1024); + } + + @Test + public void productionSizedSnapshotReplacesOneHugeEnvelopeWithBoundedMessages() { + int documentCount = 6112; + String payload = String.join("", Collections.nCopies(344, "abcdefgh")); + List documents = + new ArrayList<>(documentCount); + List changes = + new ArrayList<>(documentCount); + for (int index = 0; index < documentCount; index++) { + GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot document = + document("projects/p/assets/" + index, payload); + documents.add(document); + changes.add( + new GeneratedAndroidFirebaseFirestore.InternalDocumentChange.Builder() + .setType(GeneratedAndroidFirebaseFirestore.DocumentChangeType.ADDED) + .setDocument(document) + .setOldIndex(-1L) + .setNewIndex((long) index) + .build()); + } + GeneratedAndroidFirebaseFirestore.InternalSnapshotMetadata metadata = + new GeneratedAndroidFirebaseFirestore.InternalSnapshotMetadata.Builder() + .setHasPendingWrites(false) + .setIsFromCache(false) + .build(); + GeneratedAndroidFirebaseFirestore.InternalQuerySnapshot original = + new GeneratedAndroidFirebaseFirestore.InternalQuerySnapshot.Builder() + .setDocuments(documents) + .setDocumentChanges(changes) + .setMetadata(metadata) + .build(); + + int originalEnvelopeBytes = QuerySnapshotChunkProtocol.encodedEnvelopeSize(original); + assertTrue( + "Expected the original envelope to exceed 30 MiB, got " + originalEnvelopeBytes, + originalEnvelopeBytes > 30 * 1024 * 1024); + + assertBoundedChunks(documents, QuerySnapshotChunkProtocol.DOCUMENTS_KIND); + assertBoundedChunks(changes, QuerySnapshotChunkProtocol.DOCUMENT_CHANGES_KIND); + } + + private void assertBoundedChunks(List items, int kind) { + QuerySnapshotChunker chunker = + new QuerySnapshotChunker<>( + items.iterator(), + value -> value, + QuerySnapshotChunkProtocol::encodedMessageSize, + QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + int received = 0; + while (chunker.hasNext()) { + List chunk = + chunker.nextChunk( + values -> + QuerySnapshotChunkProtocol.encodedEnvelopeSize( + QuerySnapshotChunkProtocol.itemChunk(1, kind, values))); + assertTrue( + QuerySnapshotChunkProtocol.encodedEnvelopeSize( + QuerySnapshotChunkProtocol.itemChunk(1, kind, chunk)) + <= QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + received += chunk.size(); + } + assertEquals(items.size(), received); + } + + private GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot document( + String path, String payload) { + GeneratedAndroidFirebaseFirestore.InternalSnapshotMetadata metadata = + new GeneratedAndroidFirebaseFirestore.InternalSnapshotMetadata.Builder() + .setHasPendingWrites(false) + .setIsFromCache(false) + .build(); + return new GeneratedAndroidFirebaseFirestore.InternalDocumentSnapshot.Builder() + .setPath(path) + .setData(Collections.singletonMap("payload", payload)) + .setMetadata(metadata) + .build(); + } +} diff --git a/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkerTest.java b/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkerTest.java new file mode 100644 index 000000000000..7a245f935788 --- /dev/null +++ b/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotChunkerTest.java @@ -0,0 +1,92 @@ +/* + * Copyright 2026, the Chromium project authors. Please see the AUTHORS file + * for details. All rights reserved. Use of this source code is governed by a + * BSD-style license that can be found in the LICENSE file. + */ + +package io.flutter.plugins.firebase.firestore.streamhandler; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; + +public class QuerySnapshotChunkerTest { + @Test + public void chunksItemsWithoutReordering() { + QuerySnapshotChunker chunker = + new QuerySnapshotChunker<>( + Arrays.asList(1, 2, 3).iterator(), value -> value, value -> 4, 10); + + assertEquals(Arrays.asList(1, 2), chunker.nextChunk(this::encodedSizeWithOneByteHeader)); + assertEquals( + Collections.singletonList(3), chunker.nextChunk(this::encodedSizeWithOneByteHeader)); + assertFalse(chunker.hasNext()); + } + + @Test + public void shrinksAChunkWhenTheExactEnvelopeExceedsTheTarget() { + QuerySnapshotChunker chunker = + new QuerySnapshotChunker<>( + Arrays.asList(1, 2, 3).iterator(), value -> value, value -> 4, 10); + + assertEquals( + Collections.singletonList(1), chunker.nextChunk(this::encodedSizeWithFourByteHeader)); + assertEquals( + Collections.singletonList(2), chunker.nextChunk(this::encodedSizeWithFourByteHeader)); + assertEquals( + Collections.singletonList(3), chunker.nextChunk(this::encodedSizeWithFourByteHeader)); + } + + @Test + public void allowsOneOversizedItemSoProgressCannotStall() { + QuerySnapshotChunker chunker = + new QuerySnapshotChunker<>( + Collections.singletonList(1).iterator(), value -> value, value -> 14, 10); + + assertEquals(Collections.singletonList(1), chunker.nextChunk(values -> 15)); + assertFalse(chunker.hasNext()); + } + + @Test + public void convertsEverySourceItemExactlyOnce() { + AtomicInteger conversions = new AtomicInteger(); + QuerySnapshotChunker chunker = + new QuerySnapshotChunker<>( + Arrays.asList(1, 2, 3).iterator(), + value -> { + conversions.incrementAndGet(); + return value; + }, + value -> 4, + 10); + + while (chunker.hasNext()) { + assertTrue(chunker.nextChunk(this::encodedSizeWithFourByteHeader).size() >= 1); + } + + assertEquals(3, conversions.get()); + } + + @Test + public void emptyInputHasNoChunk() { + QuerySnapshotChunker chunker = + new QuerySnapshotChunker<>( + Collections.emptyList().iterator(), value -> value, value -> 1, 10); + + assertFalse(chunker.hasNext()); + } + + private int encodedSizeWithOneByteHeader(List values) { + return 1 + (values.size() * 4); + } + + private int encodedSizeWithFourByteHeader(List values) { + return 4 + (values.size() * 4); + } +} diff --git a/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandlerTest.java b/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandlerTest.java new file mode 100644 index 000000000000..6707aad31caf --- /dev/null +++ b/packages/cloud_firestore/cloud_firestore/android/src/test/java/io/flutter/plugins/firebase/firestore/streamhandler/QuerySnapshotsStreamHandlerTest.java @@ -0,0 +1,207 @@ +/* + * Copyright 2026, the Chromium project authors. Please see the AUTHORS file + * for details. All rights reserved. Use of this source code is governed by a + * BSD-style license that can be found in the LICENSE file. + */ + +package io.flutter.plugins.firebase.firestore.streamhandler; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.firebase.firestore.DocumentChange; +import com.google.firebase.firestore.DocumentReference; +import com.google.firebase.firestore.DocumentSnapshot; +import com.google.firebase.firestore.ListenSource; +import com.google.firebase.firestore.ListenerRegistration; +import com.google.firebase.firestore.Query; +import com.google.firebase.firestore.QueryDocumentSnapshot; +import com.google.firebase.firestore.QuerySnapshot; +import com.google.firebase.firestore.SnapshotMetadata; +import io.flutter.plugin.common.EventChannel; +import io.flutter.plugins.firebase.firestore.GeneratedAndroidFirebaseFirestore; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executor; +import org.junit.Test; + +public class QuerySnapshotsStreamHandlerTest { + @Test + public void emitsOneLogicalSnapshotAsBoundedProtocolMessages() { + Query query = mock(Query.class); + QuerySnapshot snapshot = mock(QuerySnapshot.class); + QueryDocumentSnapshot document = mock(QueryDocumentSnapshot.class); + DocumentReference reference = mock(DocumentReference.class); + SnapshotMetadata metadata = mock(SnapshotMetadata.class); + DocumentChange change = mock(DocumentChange.class); + when(snapshot.getMetadata()).thenReturn(metadata); + when(snapshot.getDocuments()).thenReturn(Collections.singletonList(document)); + when(snapshot.getDocumentChanges()).thenReturn(Collections.singletonList(change)); + when(metadata.hasPendingWrites()).thenReturn(false); + when(metadata.isFromCache()).thenReturn(false); + when(document.getMetadata()).thenReturn(metadata); + when(document.getReference()).thenReturn(reference); + when(reference.getPath()).thenReturn("projects/p/assets/a"); + when(document.getData(DocumentSnapshot.ServerTimestampBehavior.NONE)) + .thenReturn(Collections.singletonMap("title", "Asset A")); + when(change.getType()).thenReturn(DocumentChange.Type.ADDED); + when(change.getDocument()).thenReturn(document); + when(change.getOldIndex()).thenReturn(-1); + when(change.getNewIndex()).thenReturn(0); + QuerySnapshotsStreamHandler handler = + new QuerySnapshotsStreamHandler( + query, + false, + DocumentSnapshot.ServerTimestampBehavior.NONE, + ListenSource.DEFAULT, + Runnable::run, + Runnable::run); + RecordingEventSink sink = new RecordingEventSink(); + handler.emitSnapshotForTesting(snapshot, sink); + + assertEquals(4, sink.events.size()); + for (Object event : sink.events) { + assertTrue(event instanceof Map); + assertFalse(event instanceof GeneratedAndroidFirebaseFirestore.InternalQuerySnapshot); + assertTrue( + QuerySnapshotChunkProtocol.encodedEnvelopeSize(event) + <= QuerySnapshotChunkProtocol.MAX_ENCODED_ENVELOPE_BYTES); + } + assertEquals(QuerySnapshotChunkProtocol.START_KIND, kind(sink.events.get(0))); + assertEquals(QuerySnapshotChunkProtocol.DOCUMENTS_KIND, kind(sink.events.get(1))); + assertEquals(QuerySnapshotChunkProtocol.DOCUMENT_CHANGES_KIND, kind(sink.events.get(2))); + assertEquals(QuerySnapshotChunkProtocol.END_KIND, kind(sink.events.get(3))); + assertNull(sink.errorCode); + } + + @Test + public void serializesQueuedSnapshotsWithoutInterleaving() { + Query query = mock(Query.class); + QuerySnapshot snapshot = emptySnapshot(); + ManualExecutor executor = new ManualExecutor(); + QuerySnapshotsStreamHandler handler = + new QuerySnapshotsStreamHandler( + query, + false, + DocumentSnapshot.ServerTimestampBehavior.NONE, + ListenSource.DEFAULT, + executor, + executor); + RecordingEventSink sink = new RecordingEventSink(); + + handler.emitSnapshotForTesting(snapshot, sink); + handler.emitSnapshotForTesting(snapshot, sink); + executor.runAll(); + + assertEquals(4, sink.events.size()); + assertEquals(QuerySnapshotChunkProtocol.START_KIND, kind(sink.events.get(0))); + assertEquals(QuerySnapshotChunkProtocol.END_KIND, kind(sink.events.get(1))); + assertEquals(QuerySnapshotChunkProtocol.START_KIND, kind(sink.events.get(2))); + assertEquals(QuerySnapshotChunkProtocol.END_KIND, kind(sink.events.get(3))); + assertEquals(1L, snapshotId(sink.events.get(0))); + assertEquals(1L, snapshotId(sink.events.get(1))); + assertEquals(2L, snapshotId(sink.events.get(2))); + assertEquals(2L, snapshotId(sink.events.get(3))); + } + + @Test + public void cancellationDropsQueuedSnapshotMessages() { + ManualExecutor executor = new ManualExecutor(); + QuerySnapshotsStreamHandler handler = + new QuerySnapshotsStreamHandler( + mock(Query.class), + false, + DocumentSnapshot.ServerTimestampBehavior.NONE, + ListenSource.DEFAULT, + executor, + executor); + RecordingEventSink sink = new RecordingEventSink(); + + handler.emitSnapshotForTesting(emptySnapshot(), sink); + handler.onCancel(null); + executor.runAll(); + + assertTrue(sink.events.isEmpty()); + } + + @Test + public void registrationReturnedAfterCancellationIsRemoved() { + ListenerRegistration registration = mock(ListenerRegistration.class); + QuerySnapshotsStreamHandler handler = + new QuerySnapshotsStreamHandler( + mock(Query.class), + false, + DocumentSnapshot.ServerTimestampBehavior.NONE, + ListenSource.DEFAULT, + Runnable::run, + Runnable::run); + + handler.onCancel(null); + handler.retainListenerRegistration(registration); + + verify(registration).remove(); + } + + private QuerySnapshot emptySnapshot() { + QuerySnapshot snapshot = mock(QuerySnapshot.class); + SnapshotMetadata metadata = mock(SnapshotMetadata.class); + when(metadata.hasPendingWrites()).thenReturn(false); + when(metadata.isFromCache()).thenReturn(false); + when(snapshot.getMetadata()).thenReturn(metadata); + when(snapshot.getDocuments()).thenReturn(Collections.emptyList()); + when(snapshot.getDocumentChanges()).thenReturn(Collections.emptyList()); + return snapshot; + } + + private int kind(Object event) { + return (Integer) ((Map) event).get("kind"); + } + + private long snapshotId(Object event) { + return (Long) ((Map) event).get("snapshotId"); + } + + private static final class RecordingEventSink implements EventChannel.EventSink { + final List events = new ArrayList<>(); + String errorCode; + int endOfStreamCount; + + @Override + public void success(Object event) { + events.add(event); + } + + @Override + public void error(String code, String message, Object details) { + errorCode = code; + } + + @Override + public void endOfStream() { + endOfStreamCount++; + } + } + + private static final class ManualExecutor implements Executor { + final ArrayDeque pending = new ArrayDeque<>(); + + @Override + public void execute(Runnable command) { + pending.addLast(command); + } + + void runAll() { + while (!pending.isEmpty()) { + pending.removeFirst().run(); + } + } + } +} diff --git a/packages/cloud_firestore/cloud_firestore_platform_interface/lib/src/method_channel/method_channel_query.dart b/packages/cloud_firestore/cloud_firestore_platform_interface/lib/src/method_channel/method_channel_query.dart index 12604b472d7c..077e7f865235 100644 --- a/packages/cloud_firestore/cloud_firestore_platform_interface/lib/src/method_channel/method_channel_query.dart +++ b/packages/cloud_firestore/cloud_firestore_platform_interface/lib/src/method_channel/method_channel_query.dart @@ -16,6 +16,7 @@ import 'package:flutter/services.dart'; import 'method_channel_aggregate_query.dart'; import 'method_channel_firestore.dart'; import 'method_channel_query_snapshot.dart'; +import 'query_snapshot_chunk_assembler.dart'; import 'utils/exception.dart'; /// An implementation of [QueryPlatform] that uses [MethodChannel] to @@ -161,6 +162,7 @@ class MethodChannelQuery extends QueryPlatform { controller; // ignore: close_sinks StreamSubscription? snapshotStreamSubscription; + final snapshotAssembler = QuerySnapshotChunkAssembler(); controller = StreamController.broadcast( onListen: () async { @@ -185,16 +187,24 @@ class MethodChannelQuery extends QueryPlatform { ) .listen( (snapshot) { - // With Pigeon 26, the native side emits the generated Pigeon class - // directly through the Pigeon-aware codec, so we receive a fully - // decoded `InternalQuerySnapshot` here (no manual decode required). - final result = snapshot as InternalQuerySnapshot; - controller.add(MethodChannelQuerySnapshot(firestore, result)); + try { + final result = snapshotAssembler.add(snapshot); + if (result != null) { + controller.add(MethodChannelQuerySnapshot(firestore, result)); + } + } catch (error, stackTrace) { + snapshotAssembler.reset(); + controller.addError(error, stackTrace); + } + }, + onError: (Object error, StackTrace stackTrace) { + snapshotAssembler.reset(); + controller.addError(error, stackTrace); }, - onError: controller.addError, ); }, onCancel: () { + snapshotAssembler.reset(); snapshotStreamSubscription?.cancel(); }, ); diff --git a/packages/cloud_firestore/cloud_firestore_platform_interface/lib/src/method_channel/query_snapshot_chunk_assembler.dart b/packages/cloud_firestore/cloud_firestore_platform_interface/lib/src/method_channel/query_snapshot_chunk_assembler.dart new file mode 100644 index 000000000000..f0afd4cbd4e2 --- /dev/null +++ b/packages/cloud_firestore/cloud_firestore_platform_interface/lib/src/method_channel/query_snapshot_chunk_assembler.dart @@ -0,0 +1,249 @@ +// Copyright 2026, the Chromium project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import '../pigeon/messages.pigeon.dart'; + +// These map keys and chunk kinds are mirrored by QuerySnapshotsStreamHandler +// on Android. The chunk protocol deliberately uses types already supported by +// PigeonCodec so older Apple and desktop implementations can continue emitting +// InternalQuerySnapshot directly. +const String _chunkMarkerKey = 'firestoreQuerySnapshotChunk'; +const String _snapshotIdKey = 'snapshotId'; +const String _kindKey = 'kind'; +const String _payloadKey = 'payload'; +const String _documentCountKey = 'documentCount'; +const String _documentChangeCountKey = 'documentChangeCount'; + +const int _startKind = 0; +const int _documentsKind = 1; +const int _documentChangesKind = 2; +const int _endKind = 3; + +/// Reassembles the bounded Android transport messages for one logical query +/// snapshot. +/// +/// Other platforms still emit [InternalQuerySnapshot] directly, which is +/// passed through unchanged. A chunked snapshot is returned only after its end +/// marker and declared item counts have been received, preserving the public +/// stream's atomic snapshot contract. +class QuerySnapshotChunkAssembler { + _ActiveQuerySnapshot? _active; + + /// Adds one decoded event-channel [message]. + /// + /// Returns a complete snapshot, or `null` while a chunked snapshot is still + /// being assembled. Throws [StateError] when the transport protocol is + /// malformed or interleaved. + InternalQuerySnapshot? add(Object? message) { + if (message is InternalQuerySnapshot) { + if (_active != null) { + throw StateError( + 'Received a complete query snapshot while chunk assembly was active.', + ); + } + return message; + } + + if (message is! Map || message[_chunkMarkerKey] != true) { + throw StateError('Received an unknown query snapshot message.'); + } + + final snapshotId = _requiredInt(message, _snapshotIdKey); + final kind = _requiredInt(message, _kindKey); + + switch (kind) { + case _startKind: + _start(message, snapshotId); + case _documentsKind: + _appendDocuments(message, snapshotId); + case _documentChangesKind: + _appendDocumentChanges(message, snapshotId); + case _endKind: + return _finish(snapshotId); + default: + throw StateError('Unknown query snapshot chunk kind: $kind.'); + } + + return null; + } + + /// Abandons an interrupted snapshot after cancellation or a stream error. + void reset() { + _active = null; + } + + void _start(Map message, int snapshotId) { + if (_active != null) { + throw StateError( + 'Query snapshot $snapshotId started before the active snapshot ended.', + ); + } + + final documentCount = _requiredInt(message, _documentCountKey); + final documentChangeCount = _requiredInt(message, _documentChangeCountKey); + final metadata = message[_payloadKey]; + if (documentCount < 0 || documentChangeCount < 0) { + throw StateError('Query snapshot item counts cannot be negative.'); + } + if (metadata is! InternalSnapshotMetadata) { + throw StateError('Query snapshot start is missing metadata.'); + } + + _active = _ActiveQuerySnapshot( + id: snapshotId, + expectedDocumentCount: documentCount, + expectedDocumentChangeCount: documentChangeCount, + metadata: metadata, + ); + } + + void _appendDocuments(Map message, int snapshotId) { + final active = _requireActive(snapshotId); + final payload = message[_payloadKey]; + if (payload is! List) { + throw StateError('Query snapshot document chunk has an invalid payload.'); + } + + active.documents.addAll(payload.cast()); + if (active.documents.length > active.expectedDocumentCount) { + throw StateError('Query snapshot received too many documents.'); + } + } + + void _appendDocumentChanges( + Map message, + int snapshotId, + ) { + final active = _requireActive(snapshotId); + final payload = message[_payloadKey]; + if (payload is! List) { + throw StateError( + 'Query snapshot document-change chunk has an invalid payload.', + ); + } + + active.documentChanges.addAll(payload.cast()); + if (active.documentChanges.length > active.expectedDocumentChangeCount) { + throw StateError('Query snapshot received too many document changes.'); + } + } + + InternalQuerySnapshot _finish(int snapshotId) { + final active = _requireActive(snapshotId); + _active = null; + + if (active.documents.length != active.expectedDocumentCount || + active.documentChanges.length != active.expectedDocumentChangeCount) { + throw StateError( + 'Query snapshot $snapshotId ended with ' + '${active.documents.length}/${active.expectedDocumentCount} documents ' + 'and ${active.documentChanges.length}/' + '${active.expectedDocumentChangeCount} document changes.', + ); + } + + return InternalQuerySnapshot( + documents: active.documents, + documentChanges: active.documentChanges, + metadata: active.metadata, + ); + } + + _ActiveQuerySnapshot _requireActive(int snapshotId) { + final active = _active; + if (active == null) { + throw StateError( + 'Received a chunk for query snapshot $snapshotId before its start.', + ); + } + if (active.id != snapshotId) { + throw StateError( + 'Received a chunk for query snapshot $snapshotId while assembling ' + '${active.id}.', + ); + } + return active; + } +} + +class _ActiveQuerySnapshot { + _ActiveQuerySnapshot({ + required this.id, + required this.expectedDocumentCount, + required this.expectedDocumentChangeCount, + required this.metadata, + }); + + final int id; + final int expectedDocumentCount; + final int expectedDocumentChangeCount; + final InternalSnapshotMetadata metadata; + final List documents = + []; + final List documentChanges = + []; +} + +int _requiredInt(Map message, String key) { + final value = message[key]; + if (value is! int) { + throw StateError('Query snapshot chunk is missing integer "$key".'); + } + return value; +} + +/// Creates an Android query-snapshot start message for protocol tests. +Map querySnapshotChunkStart({ + required int snapshotId, + required int documentCount, + required int documentChangeCount, + required InternalSnapshotMetadata metadata, +}) { + return { + _chunkMarkerKey: true, + _snapshotIdKey: snapshotId, + _kindKey: _startKind, + _payloadKey: metadata, + _documentCountKey: documentCount, + _documentChangeCountKey: documentChangeCount, + }; +} + +/// Creates an Android query-snapshot document message for protocol tests. +Map querySnapshotDocumentChunk( + int snapshotId, + List documents, +) { + return _itemChunk(snapshotId, _documentsKind, documents); +} + +/// Creates an Android query-snapshot change message for protocol tests. +Map querySnapshotDocumentChangeChunk( + int snapshotId, + List documentChanges, +) { + return _itemChunk(snapshotId, _documentChangesKind, documentChanges); +} + +/// Creates an Android query-snapshot end message for protocol tests. +Map querySnapshotChunkEnd(int snapshotId) { + return { + _chunkMarkerKey: true, + _snapshotIdKey: snapshotId, + _kindKey: _endKind, + }; +} + +Map _itemChunk( + int snapshotId, + int kind, + List payload, +) { + return { + _chunkMarkerKey: true, + _snapshotIdKey: snapshotId, + _kindKey: kind, + _payloadKey: payload, + }; +} diff --git a/packages/cloud_firestore/cloud_firestore_platform_interface/test/method_channel/query_snapshot_chunk_assembler_test.dart b/packages/cloud_firestore/cloud_firestore_platform_interface/test/method_channel/query_snapshot_chunk_assembler_test.dart new file mode 100644 index 000000000000..715e3e8a43ff --- /dev/null +++ b/packages/cloud_firestore/cloud_firestore_platform_interface/test/method_channel/query_snapshot_chunk_assembler_test.dart @@ -0,0 +1,225 @@ +// Copyright 2026, the Chromium project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'package:cloud_firestore_platform_interface/cloud_firestore_platform_interface.dart'; +import 'package:cloud_firestore_platform_interface/src/method_channel/query_snapshot_chunk_assembler.dart'; +import 'package:flutter/services.dart'; +import 'package:flutter_test/flutter_test.dart'; + +void main() { + group('QuerySnapshotChunkAssembler', () { + test('passes through the existing platform snapshot message', () { + final assembler = QuerySnapshotChunkAssembler(); + final snapshot = _snapshot('projects/p/assets/a'); + + expect(assembler.add(snapshot), same(snapshot)); + }); + + test('reassembles chunked documents and changes atomically', () { + final assembler = QuerySnapshotChunkAssembler(); + final first = _document('projects/p/assets/a', 1); + final second = _document('projects/p/assets/b', 2); + final metadata = _metadata(isFromCache: false); + + expect( + assembler.add( + querySnapshotChunkStart( + snapshotId: 7, + documentCount: 2, + documentChangeCount: 2, + metadata: metadata, + ), + ), + isNull, + ); + expect( + assembler.add(querySnapshotDocumentChunk(7, [first])), + isNull, + ); + expect( + assembler.add(querySnapshotDocumentChunk(7, [second])), + isNull, + ); + expect( + assembler.add( + querySnapshotDocumentChangeChunk(7, [ + _change(first, 0), + _change(second, 1), + ]), + ), + isNull, + ); + + final result = assembler.add(querySnapshotChunkEnd(7)); + + expect(result, isA()); + final complete = result!; + expect(complete.documents, [first, second]); + expect(complete.documentChanges, hasLength(2)); + expect(complete.metadata, metadata); + }); + + test('rejects a new snapshot before the active snapshot ends', () { + final assembler = QuerySnapshotChunkAssembler(); + assembler.add( + querySnapshotChunkStart( + snapshotId: 1, + documentCount: 0, + documentChangeCount: 0, + metadata: _metadata(), + ), + ); + + expect( + () => assembler.add( + querySnapshotChunkStart( + snapshotId: 2, + documentCount: 0, + documentChangeCount: 0, + metadata: _metadata(), + ), + ), + throwsStateError, + ); + }); + + test('rejects chunks for a different snapshot', () { + final assembler = QuerySnapshotChunkAssembler(); + assembler.add( + querySnapshotChunkStart( + snapshotId: 4, + documentCount: 1, + documentChangeCount: 0, + metadata: _metadata(), + ), + ); + + expect( + () => assembler.add( + querySnapshotDocumentChunk( + 5, + [_document('projects/p/assets/a', 1)], + ), + ), + throwsStateError, + ); + }); + + test('rejects an incomplete snapshot', () { + final assembler = QuerySnapshotChunkAssembler(); + assembler.add( + querySnapshotChunkStart( + snapshotId: 9, + documentCount: 2, + documentChangeCount: 0, + metadata: _metadata(), + ), + ); + assembler.add( + querySnapshotDocumentChunk( + 9, + [_document('projects/p/assets/a', 1)], + ), + ); + + expect( + () => assembler.add(querySnapshotChunkEnd(9)), + throwsStateError, + ); + }); + + test('reset abandons an interrupted snapshot', () { + final assembler = QuerySnapshotChunkAssembler(); + assembler.add( + querySnapshotChunkStart( + snapshotId: 1, + documentCount: 1, + documentChangeCount: 0, + metadata: _metadata(), + ), + ); + + assembler.reset(); + + expect( + assembler.add( + querySnapshotChunkStart( + snapshotId: 2, + documentCount: 0, + documentChangeCount: 0, + metadata: _metadata(), + ), + ), + isNull, + ); + expect( + assembler.add(querySnapshotChunkEnd(2)), + isA(), + ); + }); + + test('reassembles messages after a Pigeon event-channel round trip', () { + const codec = StandardMethodCodec(PigeonCodec()); + final assembler = QuerySnapshotChunkAssembler(); + final document = _document('projects/p/assets/a', 1); + final messages = [ + querySnapshotChunkStart( + snapshotId: 3, + documentCount: 1, + documentChangeCount: 1, + metadata: _metadata(), + ), + querySnapshotDocumentChunk(3, [document]), + querySnapshotDocumentChangeChunk(3, [_change(document, 0)]), + querySnapshotChunkEnd(3), + ]; + + InternalQuerySnapshot? result; + for (final message in messages) { + final envelope = codec.encodeSuccessEnvelope(message); + result = assembler.add(codec.decodeEnvelope(envelope)); + } + + expect(result, isNotNull); + expect(result!.documents.single, document); + expect(result.documentChanges.single!.document, document); + }); + }); +} + +InternalQuerySnapshot _snapshot(String path) { + final document = _document(path, 1); + return InternalQuerySnapshot( + documents: [document], + documentChanges: [_change(document, 0)], + metadata: _metadata(), + ); +} + +InternalDocumentSnapshot _document(String path, int value) { + return InternalDocumentSnapshot( + path: path, + data: {'value': value}, + metadata: _metadata(), + ); +} + +InternalDocumentChange _change( + InternalDocumentSnapshot document, + int newIndex, +) { + return InternalDocumentChange( + type: DocumentChangeType.added, + document: document, + oldIndex: -1, + newIndex: newIndex, + ); +} + +InternalSnapshotMetadata _metadata({bool isFromCache = true}) { + return InternalSnapshotMetadata( + hasPendingWrites: false, + isFromCache: isFromCache, + ); +} diff --git a/tests/android/app/build.gradle b/tests/android/app/build.gradle index fc3d9adc9b24..bd3d7d05625d 100644 --- a/tests/android/app/build.gradle +++ b/tests/android/app/build.gradle @@ -1,5 +1,6 @@ plugins { id "com.android.application" + id "org.jetbrains.kotlin.android" // The Flutter Gradle Plugin must be applied after the Android Gradle plugin. id "dev.flutter.flutter-gradle-plugin" } diff --git a/tests/pubspec.yaml b/tests/pubspec.yaml index e8735c1a8ed6..26eed1c645c2 100644 --- a/tests/pubspec.yaml +++ b/tests/pubspec.yaml @@ -10,6 +10,8 @@ environment: flutter: '>=3.22.0' dependencies: + cloud_firestore: ^6.8.0 + cloud_firestore_platform_interface: ^8.0.6 cloud_functions: ^6.3.6 cloud_functions_platform_interface: ^6.0.6 cloud_functions_web: ^5.1.12 diff --git a/tests/windows/flutter/generated_plugin_registrant.cc b/tests/windows/flutter/generated_plugin_registrant.cc index b6d55c18d245..45647a7ffb7d 100644 --- a/tests/windows/flutter/generated_plugin_registrant.cc +++ b/tests/windows/flutter/generated_plugin_registrant.cc @@ -6,6 +6,7 @@ #include "generated_plugin_registrant.h" +#include #include #include #include @@ -14,6 +15,8 @@ #include void RegisterPlugins(flutter::PluginRegistry* registry) { + CloudFirestorePluginCApiRegisterWithRegistrar( + registry->GetRegistrarForPlugin("CloudFirestorePluginCApi")); FirebaseAppCheckPluginCApiRegisterWithRegistrar( registry->GetRegistrarForPlugin("FirebaseAppCheckPluginCApi")); FirebaseAuthPluginCApiRegisterWithRegistrar( diff --git a/tests/windows/flutter/generated_plugins.cmake b/tests/windows/flutter/generated_plugins.cmake index 89f9a20f171a..09c830ce9260 100644 --- a/tests/windows/flutter/generated_plugins.cmake +++ b/tests/windows/flutter/generated_plugins.cmake @@ -3,6 +3,7 @@ # list(APPEND FLUTTER_PLUGIN_LIST + cloud_firestore firebase_app_check firebase_auth firebase_core