Skip to content
This repository was archived by the owner on Aug 30, 2026. It is now read-only.

fix: retry watch on InternalException #875

Merged
schmidt-sebastian merged 1 commit into from
Feb 22, 2022
Merged

fix: retry watch on InternalException #875

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions google-cloud-firestore/src/main/java/com/google/cloud/firestore/Watch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -570,6 +570,9 @@ private static Status getStatus(Throwable throwable) {
return ((StatusRuntimeException) throwable).getStatus();
} else if (throwable instanceof StatusException) {
return ((StatusException) throwable).getStatus();
} else if (throwable instanceof ApiException
&& ((ApiException) throwable).getStatusCode().getTransportCode() instanceof Code) {
return ((Code) ((ApiException) throwable).getStatusCode().getTransportCode()).toStatus();
}
return null;
}
Expand Down
21 changes: 20 additions & 1 deletion google-cloud-firestore/src/test/java/com/google/cloud/firestore/WatchTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,11 @@
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;

import com.google.api.gax.grpc.GrpcStatusCode;
import com.google.api.gax.rpc.BidiStreamObserver;
import com.google.api.gax.rpc.BidiStreamingCallable;
import com.google.api.gax.rpc.ClientStream;
import com.google.api.gax.rpc.InternalException;
import com.google.cloud.firestore.Query.Direction;
import com.google.cloud.firestore.WatchTest.SnapshotDocument.ChangeType;
import com.google.cloud.firestore.spi.v1.FirestoreRpc;
Expand Down Expand Up @@ -393,6 +395,19 @@ public void queryWatchRetriesBasedOnErrorCode() throws InterruptedException {
}
}

@Test
public void queryWatchRetriesOnInternalException() throws InterruptedException {
addQueryListener();
awaitAddTarget();
send(addTarget());
destroy(new InternalException(null, GrpcStatusCode.of(Code.INTERNAL), true));
awaitAddTarget();
send(addTarget());
send(current());
send(snapshot());
awaitQuerySnapshot();
}

@Test
public void queryWatchHandlesDocumentChange() throws InterruptedException {
addQueryListener();
Expand Down Expand Up @@ -991,7 +1006,11 @@ private void send(ListenResponse response) {
}

private void destroy(Code code) {
streamObserverCapture.getValue().onError(new StatusException(io.grpc.Status.fromCode(code)));
destroy(new StatusException(io.grpc.Status.fromCode(code)));
}

private void destroy(Exception e) {
streamObserverCapture.getValue().onError(e);
}

private void close() {
Expand Down