How to see when google pub/sub completes

user_651174 0
Viewed 1348

From a client, I have the following code:

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)
future = publisher.publish(topic_path, data=json.dumps(dict(op='create_master', review_id=1273612)))

Is there a way to check when the item has finished being processed? If so, how would that be done? Now, I have not way of knowing if someone 'works' or not.

google-cloud-pubsub google-cloud-platform python
2 Answers

To know that the message has been published successfully, you would need to look at the result of the future. The preferred way is to do this asynchronously:

def callback(future):
  try:
    print(future.result()) # future.result() is the message ID for the published message.
  except Exception as e:
    print("Error publishing: " + str(e))

future = publisher.publish(topic_path, data=json.dumps(dict(op='create_master', review_id=1273612)))
future.add_done_callback(callback)

You can also do this synchronously if you want. Calling result() on the future will block until the result of the publish is available:

future = publisher.publish(topic_path, data=json.dumps(dict(op='create_master', review_id=1273612)))
try:
  print(future.result()) # future.result() is the message ID for the published message.
except Exception as e:
  print("Error publishing: " + str(e))

There is not a built-in way to know when subscribers have finished processing the message. Requiring publishers to know when subscribers have processed messages is an anti-pattern; publishers and subscribers are meant to separate entities that aren't directly aware of each other. That being said, if you need this kind of information, the best way to do it is by setting up a second topic where your original subscribers publish messages when they have finished processing that your original publishers can subscriber to in order to know when the processing is complete.

user_11212880

One way to set this up is to store it a database based on the message_id. For example, here is some example server code:

def callback(message):

    # Message has been received by the Server/Subscriber
    cursor.execute('INSERT IGNORE INTO pubsub (id, message, received) VALUES (%s, %s, NOW())', (message.message_id, message.data))
    connection.commit()

    # Message is processed by the Server/Subscriber
    data_obj = loads(message.data)
    _process(data_obj)

    # Message has finished being processed by the Server/Subscriber
    cursor.execute('UPDATE pubsub SET completed=NOW() WHERE id=%s', (message.message_id,))
    connection.commit()
    message.ack()

The client has access to the id via the future.result(), so can easily query that to see the status. This can be especially helpful if viewing statuses in a separate process (for example, if 100 long-running processes are running and we want to keep track of which have been completed).

user_6511740
Related