Pubsub: How to flush unsent messages on program exit?

[mckingho]

I have set caching configuration which may cache some messages and messages are not sent to pubsub yet. Is there a flush method when I want to exit the program.

[anguillanneuf]

A workaround is to use Seek and seek to the current timestamp. This way, messages that were published before this time are marked as acknowledged, and messages retained in the subscription that were published after this time are marked as unacknowledged.

[sduskis]

@plamut do you have any advice about this question?

[plamut]

I'm not 100% sure if I understand the question correctly, but does "caching" refer to the batch settings used when publishing messages? E.g the max_latency setting that sets how long the publisher client should wait before actually publishing a batch of messages to the server (for messages added by the publisher.publish() method)?

If so, one way of forcing the publisher to actually send the messages to the server (e.g. on program exit) is to manually invoke the batch.commit() method.

The publisher client internally stores message batches for each topic here, thus forcing a batch to commit messages early could be done with the following:

topic_batch = publisher._batches.get(TOPIC_PATH)
if topic_batch is not None:
    topic_batch.commit()

(invoking commit() is thread-safe, and will simply do nothing if batch commit is already in progress)

---

@mckingho Does that answer your question? Or maybe what @anguillanneuf wrote earlier? If not and "caching configuration" refers to something else, we would need a bit more info.

Looking forward to hearing again from you!

[sduskis]

@plamut: I think the user wants to have a method to ensure that all messages have successfully been sent... i.e. all futures have completed.

[sduskis]

We may need documentation (or a new method) for how to wait for shut down until all futures are complete.