Listening to Firestore Document Changes Through RPC API

Question

I'm currently trying to consume the RPC API provided by Google Firestore through an Unity Application via the following Code Snippet. The snippet is based on the Protobuffer Definitions and the official gRPC Tutorial showing how to implement and consume bidirectional streams:

public async void ListenTest()
{
    // create cancellation token:
    CancellationTokenSource source = new CancellationTokenSource();
    CancellationToken cancellationToken = source.Token;

    using (AsyncDuplexStreamingCall<ListenRequest, ListenResponse> call = client.Listen())
    {
        // create listenRequest with corresponding properties:
        var listenRequest = new ListenRequest();
        listenRequest.Database = database;
        listenRequest.AddTarget = new Target();
        listenRequest.AddTarget.Documents = new Target.Types.DocumentsTarget
        {
            Documents = { database + "/documents/MyCollection/MyDocument" }
        };

        // task to handle response stream messages
        var responseReaderTask = Task.Run(async () =>
        {
            while (await call.ResponseStream.MoveNext(cancellationToken))
            {
                ListenResponse loc = call.ResponseStream.Current;
                Debug.Log("Received " + loc);
            }
        });

        Debug.Log("Listening for " + database + "/documents/MyCollection/MyDocument");

        // 1. asynchronously send listenRequest
        await call.RequestStream.WriteAsync(listenRequest);

        // 2. wait until listenRequest is sent
        await call.RequestStream.CompleteAsync();

        // 3. handle responses...
        await responseReaderTask;
    }
}

Executing this immediately terminates responseReaderTask, no changes made within Firestore reach the application.

Is there something wrong with this code? I read a yet unanswered similar question, which explains the same problem for GoLang. May be the same issue.

Retrieving a single document via client.GetDocument(getDocumentRequest) works.

Answer 1

Ok, I found the error I made, maybe this helps future generations, I just removed the following line to make it work:

    // 2. wait until listenRequest is sent
    await call.RequestStream.CompleteAsync();

It seems in an AsyncDuplexStreamingCall the streams are not independent from each other (unlike the two bidirectional streams in the example I used as a basis). On completion of the RequestStream the ResponseStream also terminated, so I wasn't able to observe any changes. The Duplex in AsyncDuplexStreamingCall kind of implies this behaviour.