Allow clients to cancel ApiStreamObserver early

[mdietrichstein]

Hello,

This issue is similar to #574.

I want to use query.stream and perform some client-side filtering. In our case, this is necessary to work around two Firestore limitations:

• No inequality filters on multiple fields.
• Only a single array-contains per disjunction is allowed.

So, we perform one inequality check and one array-contains in the Firestore query and check for the remaining constraints client-side. The thing is, we only need the first X matching results. After the first X results, we want to stop the stream. Here's some code:

query.stream(
  new ApiStreamObserver<DocumentSnapshot>() {

    private static final int X = 10;

    private int numMatches = 0;

    @Override
    public void onNext(DocumentSnapshot value) {
        if (matches(value)) {
            numMatches++;
            publishMatch(value);
        }

        if (numMatches > X) {
          // Stop the stream early somehow
        }
    }

    @Override
    public void onError(Throwable t) {
      // ...
    }

    @Override
    public void onCompleted() {
      // ...
    }

    private boolean matches(DocumentSnapshot value) {
      // ...
    }

    private void publishMatch(DocumentSnapshot value) {
      // ...
    }
  }
);

I found that ResponseQueryObserver.onStart in Query.internalStream gets passed a StreamController instance which has a cancel method. Could this be exposed to the user?

I think this is a common use case and would be a nice feature to have, especially when working with large collections where only a handful of results are needed per query.

[dconeybe]

Hi @mdietrichstein. You mention that this issue is "similar" to #574. But, to me, it looks identical. Could you highlight what the difference is? We do indeed have this feature request in our backlog (to add cancellation support to query.stream()) but we have not prioritized its work at the moment.

[mdietrichstein]

Hi @dconeybe

You're right, the issue is actually identical. It's great to hear that this feature is on your backlog.

I'm open to contributing by implementing the feature myself and submitting a pull request. If you could provide some guidance or point me in the right direction, I'd be willing to take on the task if this is in your interest as well.

Let me know how I can be of assistance in moving this forward.

[dconeybe]

Hi @mdietrichstein. Thanks for your offer to author a pull request for this feature. That is definitely something we'd be open to. Still, I don't personally have the time to devote to the code review process, but I'll check with my team and see if I can find someone who does. Regardless, you are welcome to open a pull request, even if it remains unreviewed for some time.

One thing to note is that the pull request will include changes to the public API. That's fine, but we have extra internal review processes for public API changes. Your pull request would also need to include unit and/or integration tests (if appropriate). So it is likely a non-trivial time commitment for you with all of the back and forth that will likely be involved.

Something that would be of particular interest to us is your thoughts on what the public API change would look like. It would need to be backwards compatible (so existing applications continue to work). I have a preliminary idea, but would like to hear your thoughts.

[tom-andersen]

@mdietrichstein You can throw exception within onNext callback. This will cancel the stream. You will then receive an onError with exception com.google.api.gax.rpc.CancelledException. If you dig into the cause property, you will find the exception you threw earlier.

The above is not the most elegant solution, but it works.

[mdietrichstein]

Thanks @tom-andersen I will try that.

@dconeybe Great, I'll check it out once we've wrapped up our current release and think about how to implement this feature and its effect on the public API 👍