Testing Google Cloud PubSub push endpoints locally

Question

Trying to figure out the best way to test PubSub push endpoints locally. We tried with ngrok.io, but you must own the domain in order to whitelist (the tool for doing so is also broken… resulting in an infinite redirect loop). We also tried emulating PubSub locally. I am able to publish and pull, but I cannot get the push subscriptions working. We are using a local Flask webserver like so:

@app.route('/_ah/push-handlers/events', methods=['POST'])
def handle_message():
    print request.json
    return jsonify({'ok': 1}), 200

The following produces no result:

client = pubsub.Client()
topic = client('events')
topic.create()
subscription = topic.subscription('test_push', push_endpoint='http://localhost:5000/_ah/push-handlers/events')
subscription.create()
topic.publish('{"test": 123}')

It does yell at us when we attempt to create a subscription to an HTTP endpoint (whereas live PubSub will if you do not use HTTPS). Perhaps this is by design? Pull works just fine… Any ideas on how to best develop PubSub push endpoints locally?

Answer 1

Following the latest PubSub library documentation at the time of writing, the following example creates a subscription with a push configuration.

Requirements

I have tested with the following requirements :

• Google Cloud SDK 285.0.1 (for PubSub local emulator)
• Python 3.8.1
• Python packages (requirements.txt) :

flask==1.1.1
google-cloud-pubsub==1.3.1

Run PubSub emulator locally

export PUBSUB_PROJECT_ID=fake-project
gcloud beta emulators pubsub start --project=$PUBSUB_PROJECT_ID

By default, PubSub emulator starts on port 8085. Project argument can be anything and does not matter.

Flask server

Considering the following server.py :

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.route('/_ah/push-handlers/events', methods=['POST'])
def handle_message():
    print(request.json)
    return jsonify({'ok': 1}), 200


if __name__ == "__main__":
    app.run(port=5000)

Run the server (starts on port 5000) :

python server.py

PubSub example

Considering the following pubsub.py :

import sys

from google.cloud import pubsub_v1


if __name__ == "__main__":
    project_id = sys.argv[1]

    # 1. create topic (events)
    publisher_client = pubsub_v1.PublisherClient()
    topic_path = publisher_client.topic_path(project_id, "events")
    publisher_client.create_topic(topic_path)

    # 2. create subscription (test_push with push_config)
    subscriber_client = pubsub_v1.SubscriberClient()
    subscription_path = subscriber_client.subscription_path(
        project_id, "test_push"
    )
    subscriber_client.create_subscription(
        subscription_path,
        topic_path,
        push_config={
          'push_endpoint': 'http://localhost:5000/_ah/push-handlers/events'
        }
    )

    # 3. publish a test message
    publisher_client.publish(
        topic_path,
        data='{"test": 123}'.encode("utf-8")
    )

Finally, run this script :

PUBSUB_EMULATOR_HOST=localhost:8085 \
PUBSUB_PROJECT_ID=fake-project \
python pubsub.py $PUBSUB_PROJECT_ID

Results

Then, you can see the results in Flask server's log :

{'subscription': 'projects/fake-project/subscriptions/test_push', 'message': {'data': 'eyJ0ZXN0IjogMTIzfQ==', 'messageId': '1', 'attributes': {}}}
127.0.0.1 - - [22/Mar/2020 12:11:00] "POST /_ah/push-handlers/events HTTP/1.1" 200 -

Note that you can retrieve the message sent, encoded here in base64 (message.data) :

$ echo "eyJ0ZXN0IjogMTIzfQ==" | base64 -d
{"test": 123}

Of course, you can also do the decoding in Python.