Pub/Sub Setup

You are a Google Cloud Platform engineer. The user wants to set up Google Pub/Sub messaging with topics, subscriptions, and basic publish/subscribe operations.

What to check first

• Run gcloud auth list to verify you have an authenticated GCP account
• Run gcloud config get-value project to confirm your active GCP project ID
• Verify the Pub/Sub API is enabled: gcloud services list --enabled | grep pubsub

Steps

1. Enable the Pub/Sub API with gcloud services enable pubsub.googleapis.com
2. Create a topic using gcloud pubsub topics create my-topic
3. Create a subscription to that topic with gcloud pubsub subscriptions create my-subscription --topic=my-topic
4. Install the Google Cloud Pub/Sub client library: pip install google-cloud-pubsub
5. Set the GOOGLE_APPLICATION_CREDENTIALS environment variable to point to your service account JSON key file
6. Write publisher code to send messages to the topic using the PublisherClient class
7. Write subscriber code to receive messages using the SubscriberClient class with a callback function
8. Test by running the publisher and subscriber scripts, verifying messages flow end-to-end

Code

from google.cloud import pubsub_v1
import json
import time
from concurrent.futures import TimeoutError

# Configuration
PROJECT_ID = "your-gcp-project-id"
TOPIC_ID = "my-topic"
SUBSCRIPTION_ID = "my-subscription"

# Publisher: Send a message to the topic
def publish_message(project_id, topic_id, message_data):
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path(project_id, topic_id)
    
    # Convert message to JSON bytes
    message_json = json.dumps(message_data)
    message_bytes = message_json.encode('utf-8')
    
    # Publish the message
    future = publisher.publish(topic_path, message_bytes)
    message_id = future.result()
    
    print(f"Published message ID: {message_id}")
    return message_id

# Subscriber: Listen for messages from the subscription
def subscribe_to_messages(project_id, subscription_id, timeout=10):
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(project_id, subscription_id)
    
    # Define callback function to process messages
    def callback(message):
        print(f"Received message: {message.data.decode('utf-8')}")
        # Acknowledge the message so it won't be redelivered
        message.ack()
    
    # Subscribe with the callback
    streaming_pull_future = subscriber.subscribe(
        subscription_path, 
        callback=callback
    )
    
    print(

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change