@chemelnucfin I'm not very familiar with the feature. To start, which RPC does watch correspond to? (The only BIDI RPCs are Listen and Write and it seems Listen is a closer synonym to Watch)

A basic implementation would like like this:

import threading
import uuid

from six.moves import queue as queue_mod

from google.cloud import firestore


STOP = uuid.uuid4()


def make_request_gen(queue):
    while True:
        request = queue.get()  # Blocks "forever"
        if request == STOP:
            break

        yield request


def do_watch(queue, firestore_api):
    request_gen = make_requent_gen(queue)
    response_gen = firestore_api.listen(request_gen)

    for response in response_gen:
        do_something(response)

    queue.put(STOP)


client = firestore.Client()
firestore_api = client._firestore_api

queue = queue_mod.Queue()
thread = threading.Thread(
    target=do_watch, args=(queue, firestore_api), name='Meh')
thread.start()

Also seems to me that threads would be ok and I don't need multiprocessing?

I would leave it open so that users can specify which type of concurrency they want (with the above that would mean swapping out the queue and the thread for other primitives)

But the question I have is how do I tell grpc to send the ListenResponse to the queue that I have?

The

Also, would it be better to have one queue per Watch instance or one queue per system?

The queue is for your requests. It seems in many cases you'd only ever send a single initial ListenRequest so the queue is just a way for you to send STOP (which is actually necessary).

I don't think you'd need a queue for the ListenResponse messages that come in via response_gen, you just need a way to process them (i.e. do_something). But that is me speaking from ignorance of how the "watch" feature is supposed to work.

firestore: Implement Watch #4513

Description

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions